minigraf 1.1.1

use super::evaluator::{StratifiedEvaluator, evaluate_not_join};
use super::functions::{AggImpl, FunctionRegistry, apply_builtin_aggregate, value_cmp};
use super::matcher::{PatternMatcher, edn_to_entity_id, edn_to_value};
use super::optimizer;
use super::rules::RuleRegistry;
use super::types::{
    AsOf, AttributeSpec, BinOp, DatalogCommand, DatalogQuery, EdnValue, Expr, FindSpec, Order,
    Pattern, Rule, Transaction, UnaryOp, ValidAt, WhereClause, WindowFunc,
};
use crate::graph::FactStorage;
use crate::graph::types::{Fact, TransactOptions, TxId, Value, tx_id_now};
use anyhow::{Result, anyhow};
use std::collections::HashMap;
use std::sync::{Arc, RwLock};

/// Returns true if any where clause (at any depth) contains a per-fact
/// pseudo-attribute pattern (ValidFrom / ValidTo / TxCount / TxId).
/// Used to enforce the `:any-valid-time` requirement.
fn query_uses_per_fact_pseudo_attr(query: &DatalogQuery) -> bool {
    fn check_clauses(clauses: &[WhereClause]) -> bool {
        clauses.iter().any(|c| match c {
            WhereClause::Pattern(p) => matches!(
                &p.attribute,
                AttributeSpec::Pseudo(pa) if pa.is_per_fact()
            ),
            WhereClause::Not(inner) => check_clauses(inner),
            WhereClause::NotJoin { clauses: inner, .. } => check_clauses(inner),
            WhereClause::Or(branches) => branches.iter().any(|b| check_clauses(b)),
            WhereClause::OrJoin { branches, .. } => branches.iter().any(|b| check_clauses(b)),
            _ => false,
        })
    }
    check_clauses(&query.where_clauses)
}

/// Recursively collect all `Pattern` clauses from a slice of where clauses,
/// including those nested inside `Not`, `NotJoin`, `Or`, and `OrJoin` bodies.
/// Used by `selective_fact_fetch` to ensure every pattern that references a fact
/// (including not-body patterns) is considered when deciding which indexes to query.
fn collect_all_patterns(clauses: &[WhereClause]) -> Vec<Pattern> {
    let mut patterns = Vec::new();
    for clause in clauses {
        match clause {
            WhereClause::Pattern(p) => patterns.push(p.clone()),
            WhereClause::Not(inner) => patterns.extend(collect_all_patterns(inner)),
            WhereClause::NotJoin { clauses: inner, .. } => {
                patterns.extend(collect_all_patterns(inner))
            }
            WhereClause::Or(branches) | WhereClause::OrJoin { branches, .. } => {
                for branch in branches {
                    patterns.extend(collect_all_patterns(branch));
                }
            }
            WhereClause::RuleInvocation { .. } | WhereClause::Expr { .. } => {}
        }
    }
    patterns
}

/// The result of executing a Datalog command via [`crate::db::Minigraf::execute`].
///
/// Pattern-match on this to distinguish query results from write confirmations:
///
/// ```
/// # use minigraf::{Minigraf, QueryResult};
/// # let db = Minigraf::in_memory().unwrap();
/// # db.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// match db.execute("(query [:find ?name :where [?e :person/name ?name]])").unwrap() {
///     QueryResult::QueryResults { vars, results } => {
///         for row in &results {
///             println!("{}: {:?}", vars[0], row[0]);
///         }
///     }
///     QueryResult::Transacted(tx_id) => println!("tx {}", tx_id),
///     QueryResult::Retracted(tx_id) => println!("retracted tx {}", tx_id),
///     QueryResult::Ok => {}
/// }
/// ```
#[derive(Debug, Clone, PartialEq)]
pub enum QueryResult {
    /// Transaction completed successfully. The inner value is the transaction ID
    /// (Unix milliseconds). Use [`crate::db::Minigraf::current_tx_count`] to retrieve
    /// the monotonic counter (`:as-of N` value) after a write.
    Transacted(TxId),
    /// Retraction completed successfully. The inner value is the transaction ID
    /// (Unix milliseconds). Use [`crate::db::Minigraf::current_tx_count`] to retrieve
    /// the monotonic counter (`:as-of N` value) after a write.
    Retracted(TxId),
    /// Query results: list of variable bindings
    QueryResults {
        /// The variable names in the order they appear in the `:find` clause.
        vars: Vec<String>,
        /// Each inner `Vec<Value>` is one result row, aligned with `vars`.
        results: Vec<Vec<Value>>,
    },
    /// Acknowledgement for commands that produce no data (e.g. rule definitions
    /// inside a [`crate::db::WriteTransaction`]).
    Ok,
}

/// Executor for Datalog commands
pub struct DatalogExecutor {
    storage: FactStorage,
    facts_override: Option<Arc<[Fact]>>,
    read_now_floor: Option<i64>,
    rules: Arc<RwLock<RuleRegistry>>,
    // RwLock pre-wired for 7.7b register_aggregate API.
    functions: Arc<RwLock<FunctionRegistry>>,
    indexes: Arc<crate::storage::index::Indexes>,
    max_derived_facts: usize,
    max_results: usize,
}

impl DatalogExecutor {
    #[allow(dead_code)]
    pub fn new(storage: FactStorage) -> Self {
        DatalogExecutor {
            storage,
            facts_override: None,
            read_now_floor: None,
            rules: Arc::new(RwLock::new(RuleRegistry::new())),
            functions: Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
            indexes: Arc::new(crate::storage::index::Indexes::new()),
            max_derived_facts: crate::query::datalog::evaluator::DEFAULT_MAX_DERIVED_FACTS,
            max_results: crate::query::datalog::evaluator::DEFAULT_MAX_RESULTS,
        }
    }

    /// Create a `DatalogExecutor` with a shared rule registry and function registry.
    ///
    /// Used by `Minigraf` to share registries across all `execute()` calls.
    pub fn new_with_rules_and_functions(
        storage: FactStorage,
        rules: Arc<RwLock<RuleRegistry>>,
        functions: Arc<RwLock<FunctionRegistry>>,
    ) -> Self {
        DatalogExecutor {
            storage,
            facts_override: None,
            read_now_floor: None,
            rules,
            functions,
            indexes: Arc::new(crate::storage::index::Indexes::new()),
            max_derived_facts: crate::query::datalog::evaluator::DEFAULT_MAX_DERIVED_FACTS,
            max_results: crate::query::datalog::evaluator::DEFAULT_MAX_RESULTS,
        }
    }

    /// Create a `DatalogExecutor` over a merged fact slice while sharing rules and functions.
    pub(crate) fn new_from_facts_with_rules_and_functions(
        facts: Arc<[Fact]>,
        pending_read_now_floor: Option<i64>,
        rules: Arc<RwLock<RuleRegistry>>,
        functions: Arc<RwLock<FunctionRegistry>>,
    ) -> Self {
        DatalogExecutor {
            storage: FactStorage::new(),
            facts_override: Some(facts),
            read_now_floor: pending_read_now_floor,
            rules,
            functions,
            indexes: Arc::new(crate::storage::index::Indexes::new()),
            max_derived_facts: crate::query::datalog::evaluator::DEFAULT_MAX_DERIVED_FACTS,
            max_results: crate::query::datalog::evaluator::DEFAULT_MAX_RESULTS,
        }
    }

    /// Convenience constructor for tests. Shares `rules` with other executors but creates
    /// a fresh `FunctionRegistry::with_builtins()`. Production code uses
    /// [`new_with_rules_and_functions`] to share the registry from `Minigraf::Inner`.
    #[allow(dead_code)]
    pub fn new_with_rules(storage: FactStorage, rules: Arc<RwLock<RuleRegistry>>) -> Self {
        Self::new_with_rules_and_functions(
            storage,
            rules,
            Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
        )
    }

    /// Create a `DatalogExecutor` with custom complexity limits.
    ///
    /// Used by `Minigraf` when `OpenOptions` specifies non-default limits.
    #[allow(dead_code)]
    pub fn new_with_limits(
        storage: FactStorage,
        rules: Arc<RwLock<RuleRegistry>>,
        functions: Arc<RwLock<FunctionRegistry>>,
        max_derived_facts: usize,
        max_results: usize,
    ) -> Self {
        let indexes = storage.pending_indexes_snapshot();
        DatalogExecutor {
            storage,
            facts_override: None,
            read_now_floor: None,
            rules,
            functions,
            indexes: Arc::new(indexes),
            max_derived_facts,
            max_results,
        }
    }

    /// Set complexity limits on an existing executor.
    pub fn set_limits(&mut self, max_derived_facts: usize, max_results: usize) {
        self.max_derived_facts = max_derived_facts;
        self.max_results = max_results;
    }

    /// Read-time "now" for query visibility.
    ///
    /// Overlay reads floor the wall clock to the newest staged fact so buffered
    /// writes remain visible even if their synthetic metadata is slightly ahead
    /// of the current millisecond.
    fn read_now(&self) -> i64 {
        let now = tx_id_now().cast_signed();
        self.read_now_floor.map_or(now, |floor| now.max(floor))
    }

    /// Execute a Datalog command
    pub fn execute(&self, command: DatalogCommand) -> Result<QueryResult> {
        match command {
            DatalogCommand::Transact(tx) => self.execute_transact(tx),
            DatalogCommand::Retract(tx) => self.execute_retract(tx),
            DatalogCommand::Query(query) => self.execute_query(query),
            DatalogCommand::Rule(rule) => self.execute_rule(rule),
        }
    }

    /// Execute a transact command: add facts to storage
    fn execute_transact(&self, tx: Transaction) -> Result<QueryResult> {
        // Transaction-level valid-time options (fallback when no per-fact override)
        let tx_opts = if tx.valid_from.is_some() || tx.valid_to.is_some() {
            Some(TransactOptions::new(tx.valid_from, tx.valid_to))
        } else {
            None
        };

        // Collect all facts into a single batch so they share one tx_count.
        // Each fact carries its own per-fact opts (or None to fall back to tx_opts).
        let mut fact_tuples = Vec::new();
        for pattern in tx.facts {
            let entity_id =
                edn_to_entity_id(&pattern.entity).map_err(|e| anyhow!("Invalid entity: {}", e))?;

            let attribute = match &pattern.attribute {
                AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
                AttributeSpec::Real(_) => return Err(anyhow!("Attribute must be a keyword")),
                AttributeSpec::Pseudo(_) => {
                    return Err(anyhow!("Cannot transact a pseudo-attribute"));
                }
            };

            let value =
                edn_to_value(&pattern.value).map_err(|e| anyhow!("Invalid value: {}", e))?;

            let per_fact_opts = if pattern.valid_from.is_some() || pattern.valid_to.is_some() {
                Some(TransactOptions::new(pattern.valid_from, pattern.valid_to))
            } else {
                None
            };

            fact_tuples.push((entity_id, attribute, value, per_fact_opts));
        }

        let (tx_id, _tx_count) = self
            .storage
            .transact_batch(fact_tuples, tx_opts)
            .map_err(|e| anyhow!("Transaction failed: {}", e))?;

        Ok(QueryResult::Transacted(tx_id))
    }

    /// Execute a retract command: retract facts from storage
    fn execute_retract(&self, tx: Transaction) -> Result<QueryResult> {
        let mut fact_tuples = Vec::new();

        for pattern in tx.facts {
            let entity_id =
                edn_to_entity_id(&pattern.entity).map_err(|e| anyhow!("Invalid entity: {}", e))?;

            let attribute = match &pattern.attribute {
                AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
                AttributeSpec::Real(_) => return Err(anyhow!("Attribute must be a keyword")),
                AttributeSpec::Pseudo(_) => {
                    return Err(anyhow!("Cannot transact a pseudo-attribute"));
                }
            };

            let value =
                edn_to_value(&pattern.value).map_err(|e| anyhow!("Invalid value: {}", e))?;

            fact_tuples.push((entity_id, attribute, value));
        }

        let (tx_id, _tx_count) = self
            .storage
            .retract(fact_tuples)
            .map_err(|e| anyhow!("Retraction failed: {}", e))?;

        Ok(QueryResult::Retracted(tx_id))
    }

    /// Build a filtered fact snapshot for a query's temporal constraints.
    ///
    /// Step 1: apply transaction-time filter (`:as-of`) — defaults to all facts.
    /// Step 2: discard retracted facts within the tx window (`net_asserted_facts`).
    /// Step 3: apply valid-time filter (`:valid-at`) — defaults to "currently valid".
    ///
    /// Returns an `Arc<[Fact]>` snapshot. `.clone()` is a cheap Arc refcount increment,
    /// so `or`-branches and `not`/`not-join` sub-evaluations share the same allocation.
    /// The three steps above are paid exactly once per `execute_query` /
    /// `execute_query_with_rules` call.
    ///
    /// Step 1 uses selective index-backed fetches when query patterns bind concrete entities
    /// or attributes (up to 4 distinct lookups); falls back to `get_all_facts()` otherwise.
    /// Step 2 (caching `net_asserted_facts()`) remains a future optimisation opportunity.
    fn filter_facts_for_query(&self, query: &DatalogQuery) -> Result<Arc<[Fact]>> {
        let now = self.read_now();

        let source_facts: Vec<Fact> = match (&self.facts_override, query.as_of.as_ref()) {
            (Some(facts), Some(as_of)) => {
                crate::graph::storage::filter_facts_as_of(facts.iter().cloned().collect(), as_of)
            }
            (Some(facts), None) => facts.iter().cloned().collect(),
            (None, Some(as_of)) => self.storage.get_facts_as_of(as_of)?,
            (None, None) => {
                // Selective fetch is only safe when no rule invocations are present —
                // rules require the full fact base to evaluate correctly.
                if !query.uses_rules() {
                    let patterns = collect_all_patterns(&query.where_clauses);
                    match self.selective_fact_fetch(&patterns, 4) {
                        Some(facts) => facts,
                        None => self.storage.get_all_facts()?,
                    }
                } else {
                    self.storage.get_all_facts()?
                }
            }
        };

        let tx_filtered = source_facts;

        // Step 2: compute net-asserted view — for each (entity, attribute, value) triple,
        // keep it only if the record with the highest tx_count is an assertion.
        // This correctly hides facts that have been retracted.
        let asserted = crate::graph::storage::net_asserted_facts(tx_filtered);

        // Step 3: valid-time filter
        let valid_filtered: Vec<Fact> = match &query.valid_at {
            Some(ValidAt::Timestamp(t)) => asserted
                .into_iter()
                .filter(|f| f.valid_from <= *t && *t < f.valid_to)
                .collect(),
            Some(ValidAt::AnyValidTime) => asserted,
            Some(ValidAt::Slot(_)) => {
                return Err(anyhow!(
                    "internal: unsubstituted :valid-at bind slot reached the executor"
                ));
            }
            None => asserted
                .into_iter()
                .filter(|f| f.valid_from <= now && now < f.valid_to)
                .collect(),
        };

        Ok(Arc::from(valid_filtered))
    }

    /// Attempt a selective index-backed fact fetch for the given patterns.
    ///
    /// Inspects `patterns` for bound entity literals (UUID or keyword → deterministic UUID)
    /// and bound attribute keywords. If the total distinct lookup count is 0 (nothing bound)
    /// or exceeds `threshold` (too many — full scan is cheaper), returns `None`.
    /// Otherwise returns `Some(facts)` — the union of all selectively fetched facts,
    /// deduplicated by `(entity, attribute, tx_count)`.
    fn selective_fact_fetch(&self, patterns: &[Pattern], threshold: usize) -> Option<Vec<Fact>> {
        use std::collections::HashSet;

        let mut entity_ids: HashSet<uuid::Uuid> = HashSet::new();
        let mut attributes: HashSet<String> = HashSet::new();

        for pattern in patterns {
            // Bound entity: UUID literal or keyword that resolves deterministically
            match &pattern.entity {
                EdnValue::Uuid(u) => {
                    entity_ids.insert(*u);
                }
                EdnValue::Keyword(_) => {
                    if let Ok(uid) = edn_to_entity_id(&pattern.entity) {
                        entity_ids.insert(uid);
                    }
                }
                _ => {}
            }
            // Bound attribute: non-variable keyword
            if let AttributeSpec::Real(EdnValue::Keyword(attr)) = &pattern.attribute {
                attributes.insert(attr.clone());
            }
        }

        let total = entity_ids.len() + attributes.len();
        if total == 0 || total > threshold {
            return None;
        }

        // Dedup key: (entity uuid, attribute string, tx_count) — avoids Value debug formatting.
        let mut seen: HashSet<(uuid::Uuid, String, u64)> = HashSet::new();
        let mut all_facts: Vec<Fact> = Vec::new();

        for uid in &entity_ids {
            match self.storage.get_facts_by_entity(uid) {
                Ok(facts) => {
                    for fact in facts {
                        let key = (fact.entity, fact.attribute.clone(), fact.tx_count);
                        if seen.insert(key) {
                            all_facts.push(fact);
                        }
                    }
                }
                Err(_) => return None,
            }
        }

        for attr in &attributes {
            match self.storage.get_facts_by_attribute(attr) {
                Ok(facts) => {
                    for fact in facts {
                        let key = (fact.entity, fact.attribute.clone(), fact.tx_count);
                        if seen.insert(key) {
                            all_facts.push(fact);
                        }
                    }
                }
                Err(_) => return None,
            }
        }

        Some(all_facts)
    }

    /// Execute a query: find matching facts and return specified variables
    fn execute_query(&self, query: DatalogQuery) -> Result<QueryResult> {
        // Check if query uses rules
        if query.uses_rules() {
            // Use StratifiedEvaluator for queries with rule invocations (handles negation and strata)
            return self.execute_query_with_rules(query);
        }

        // Warn about queries with no binding mechanism
        if !query.has_binding_mechanism() {
            return Err(anyhow!(
                "query has no :where clause, rules, or aggregates — nothing binds the variables. \
                 Add a :where clause (e.g., [:find ?e ?a ?v :where [?e ?a ?v]]) or use an aggregate."
            ));
        }

        // Compute query-level valid_at value for :db/valid-at pseudo-attribute binding.
        let now = self.read_now();
        let valid_at_value = match &query.valid_at {
            Some(ValidAt::Timestamp(t)) => Value::Integer(*t),
            Some(ValidAt::AnyValidTime) => Value::Null,
            Some(ValidAt::Slot(_)) => {
                return Err(anyhow!(
                    "internal: unsubstituted :valid-at bind slot reached the executor"
                ));
            }
            None => Value::Integer(now),
        };

        // Hard-error: per-fact pseudo-attrs require :any-valid-time.
        if query_uses_per_fact_pseudo_attr(&query)
            && !matches!(query.valid_at, Some(ValidAt::AnyValidTime))
        {
            return Err(anyhow!(
                "temporal pseudo-attributes :db/valid-from, :db/valid-to, :db/tx-count, and \
                 :db/tx-id require :any-valid-time; add :any-valid-time to your query"
            ));
        }

        // Apply temporal filters before pattern matching
        let filtered_facts = self.filter_facts_for_query(&query)?;
        let matcher = PatternMatcher::from_slice_with_valid_at(
            filtered_facts.clone(),
            valid_at_value.clone(),
        );
        // Acquire function registry before the plan loop — needed for inline Expr evaluation.
        let registry = self
            .functions
            .read()
            .map_err(|_| anyhow!("functions lock poisoned"))?;

        // Pre-validate UDF predicate names: surface unknown predicates as errors before
        // processing any rows (matches the behaviour of the former apply_expr_clauses post-pass).
        for clause in &query.where_clauses {
            if let WhereClause::Expr {
                expr: Expr::UnaryOp(UnaryOp::Udf(name), _),
                ..
            } = clause
                && registry.get_predicate(name).is_none()
            {
                anyhow::bail!("unknown predicate: '{}'", name);
            }
        }

        // Collect Pattern and Expr top-level clauses for the planner.
        // Not/NotJoin/Or/OrJoin are extracted separately below and applied as post-filters.
        let plan_clauses: Vec<WhereClause> = query
            .where_clauses
            .iter()
            .filter(|c| matches!(c, WhereClause::Pattern(_) | WhereClause::Expr { .. }))
            .cloned()
            .collect();

        let planned = optimizer::plan(plan_clauses, &self.indexes);

        // Process planned clauses in order: Pattern → expand bindings, Expr → filter/extend.
        let mut bindings: Vec<Binding> = vec![Binding::new()];
        for (clause, hint) in planned {
            match clause {
                WhereClause::Pattern(p) => {
                    bindings = matcher.match_with_hint_seeded(
                        bindings,
                        &p,
                        hint.as_ref().unwrap_or(&optimizer::IndexHint::Eavt),
                    );
                }
                WhereClause::Expr { expr, binding: out } => {
                    bindings = bindings
                        .into_iter()
                        .filter_map(|mut b| match eval_expr(&expr, &b, Some(&registry)) {
                            Ok(v) => {
                                if let Some(var) = &out {
                                    b.insert(var.clone(), v);
                                    Some(b)
                                } else if is_truthy(&v) {
                                    Some(b)
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        })
                        .collect();
                }
                _ => {}
            }
        }

        // Apply Or/OrJoin clauses (post-pass: after pattern matching, before not/expr)
        let rules_guard = self
            .rules
            .read()
            .map_err(|_| anyhow!("rules lock poisoned"))?;
        let bindings = apply_or_clauses(
            &query.where_clauses,
            bindings,
            filtered_facts.clone(),
            &rules_guard,
            query.as_of.clone(),
            query.valid_at.clone(),
            &registry,
        )?;
        drop(rules_guard);

        // Apply not-filter for WhereClause::Not and WhereClause::NotJoin clauses
        // (no rules involved — pure post-filter)
        #[cfg_attr(feature = "wasm", allow(unused_mut))]
        let mut not_clauses: Vec<&Vec<WhereClause>> = query
            .where_clauses
            .iter()
            .filter_map(|c| match c {
                WhereClause::Not(inner) => Some(inner),
                _ => None,
            })
            .collect();

        #[cfg_attr(feature = "wasm", allow(unused_mut))]
        let mut not_join_clauses: Vec<(Vec<String>, Vec<WhereClause>)> = query
            .where_clauses
            .iter()
            .filter_map(|c| match c {
                WhereClause::NotJoin { join_vars, clauses } => {
                    Some((join_vars.clone(), clauses.clone()))
                }
                _ => None,
            })
            .collect();

        // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
        #[cfg(not(feature = "wasm"))]
        not_clauses.sort_by_key(|body| optimizer::clause_cost(&WhereClause::Not(body.to_vec())));
        // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
        #[cfg(not(feature = "wasm"))]
        not_join_clauses.sort_by_key(|(vars, clauses)| {
            optimizer::clause_cost(&WhereClause::NotJoin {
                join_vars: vars.clone(),
                clauses: clauses.clone(),
            })
        });

        let not_filtered: Vec<_> = if not_clauses.is_empty() && not_join_clauses.is_empty() {
            bindings
        } else {
            // Pre-compute exclusion sets — one evaluation per not-body, not per outer binding.
            //
            // For each not-body, run pattern matching once against `filtered_facts` to get
            // all bindings where the body is satisfiable. Then collect the "join keys" (the
            // subset of variables that appear in the outer bindings) into a HashSet.
            // The filter loop below does one O(1) probe per outer binding instead of a full
            // pattern match.
            //
            // Edge case: expr-only bodies (no patterns) produce no pre-computed set and fall
            // through to `not_body_matches` as before (rare, already fast).
            use std::collections::HashSet;

            // --- Not bodies ---
            // Each element: either Some((has_expr, HashSet of excluded join-key tuples)) or None
            // (expr-only, use slow path). `has_expr` is computed once here during pre-compute,
            // not inside the per-binding filter closure.
            let not_exclusion_sets: Vec<NotExclusionEntry> = not_clauses
                .iter()
                .map(|not_body| {
                    let has_expr = not_body
                        .iter()
                        .any(|c| matches!(c, WhereClause::Expr { .. }));
                    let patterns: Vec<_> = not_body
                        .iter()
                        .filter_map(|c| match c {
                            WhereClause::Pattern(p) => Some(p.clone()),
                            _ => None,
                        })
                        .collect();
                    if patterns.is_empty() {
                        // Expr-only body: no pre-computation possible.
                        return None;
                    }
                    let matcher = PatternMatcher::from_slice_with_valid_at(
                        filtered_facts.clone(),
                        valid_at_value.clone(),
                    );
                    let body_bindings = matcher.match_patterns(&patterns);
                    // Store all body bindings as sorted (key, value) vecs for probing.
                    // Normalize values (e.g. keyword entities → Ref) so that probe keys from
                    // the outer binding match body bindings regardless of representation.
                    let exclusion_set: HashSet<Vec<(String, Value)>> = body_bindings
                        .into_iter()
                        .map(|mut b| {
                            // Drop hidden metadata keys (prefixed `__f`)
                            b.retain(|k, _| !k.starts_with("__f"));
                            let mut kv: Vec<(String, Value)> = b
                                .into_iter()
                                .map(|(k, v)| (k, normalize_value(&v)))
                                .collect();
                            kv.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            kv
                        })
                        .collect();
                    Some((has_expr, exclusion_set))
                })
                .collect();

            // --- Not-join bodies ---
            // Each entry: Some((has_expr, key_vars, HashSet)) where key_vars are the join_vars
            // that actually appear in ALL body binding rows (the intersection of join_vars and
            // vars bound in every row). This handles cases like `not-join [?u ?r]` where the
            // body only binds `?r`.
            //
            // Values are normalized: Value::Keyword(k) that represents an entity keyword is
            // converted to Value::Ref(uuid) so that probe keys from the outer binding (which
            // may store entity references as keywords) match the body bindings (which bind
            // entity fields as Value::Ref). `has_expr` is computed once here, not per-binding.
            let not_join_exclusion_sets: Vec<NotJoinExclusionEntry> = not_join_clauses
                .iter()
                .map(|(join_vars, nj_clauses)| {
                    let has_expr = nj_clauses
                        .iter()
                        .any(|c| matches!(c, WhereClause::Expr { .. }));
                    let patterns: Vec<_> = nj_clauses
                        .iter()
                        .filter_map(|c| match c {
                            WhereClause::Pattern(p) => Some(p.clone()),
                            _ => None,
                        })
                        .collect();
                    if patterns.is_empty() {
                        return None;
                    }
                    let matcher = PatternMatcher::from_slice_with_valid_at(
                        filtered_facts.clone(),
                        valid_at_value.clone(),
                    );
                    let body_bindings = matcher.match_patterns(&patterns);
                    if body_bindings.is_empty() {
                        return Some((has_expr, join_vars.clone(), HashSet::new()));
                    }
                    let key_vars: Vec<String> = join_vars
                        .iter()
                        .filter(|v| body_bindings.iter().all(|b| b.contains_key(*v)))
                        .cloned()
                        .collect();
                    let exclusion_set: HashSet<Vec<(String, Value)>> = body_bindings
                        .into_iter()
                        .map(|b| {
                            let mut kv: Vec<(String, Value)> = key_vars
                                .iter()
                                .filter_map(|v| {
                                    b.get(v).map(|val| (v.clone(), normalize_value(val)))
                                })
                                .collect();
                            kv.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            kv
                        })
                        .collect();
                    Some((has_expr, key_vars, exclusion_set))
                })
                .collect();

            bindings
                .into_iter()
                .filter(|binding| {
                    // Check not-bodies via pre-computed exclusion sets (fast path) or
                    // via not_body_matches (slow path for expr-only bodies).
                    for (not_body, exclusion_entry) in
                        not_clauses.iter().zip(not_exclusion_sets.iter())
                    {
                        match exclusion_entry {
                            Some((has_expr, exclusion_set)) => {
                                if exclusion_set.is_empty() && !has_expr {
                                    // No excluding bindings → this outer binding is safe.
                                    continue;
                                }
                                if !has_expr {
                                    // Fast path: probe exclusion set using the outer binding's
                                    // values for the join variables, normalized for consistency.
                                    if let Some(sample) = exclusion_set.iter().next() {
                                        let key: Vec<(String, Value)> = sample
                                            .iter()
                                            .filter_map(|(var, _)| {
                                                binding
                                                    .get(var)
                                                    .map(|val| (var.clone(), normalize_value(val)))
                                            })
                                            .collect();
                                        if key.len() == sample.len() {
                                            // All join vars are bound in the outer binding.
                                            if exclusion_set.contains(&key) {
                                                return false;
                                            }
                                            continue;
                                        }
                                        // Outer binding is underspecified (fewer vars than
                                        // the exclusion set key) — fall back to slow path.
                                        if not_body_matches(
                                            not_body,
                                            binding,
                                            filtered_facts.clone(),
                                            valid_at_value.clone(),
                                            &registry,
                                        ) {
                                            return false;
                                        }
                                        continue;
                                    }
                                }
                                // Slow path fallback (expr clauses or empty exclusion set with exprs)
                                if not_body_matches(
                                    not_body,
                                    binding,
                                    filtered_facts.clone(),
                                    valid_at_value.clone(),
                                    &registry,
                                ) {
                                    return false;
                                }
                            }
                            None => {
                                // Expr-only body: slow path.
                                if not_body_matches(
                                    not_body,
                                    binding,
                                    filtered_facts.clone(),
                                    valid_at_value.clone(),
                                    &registry,
                                ) {
                                    return false;
                                }
                            }
                        }
                    }

                    // Check not-join-bodies.
                    for ((join_vars, nj_clauses), nj_exclusion_entry) in
                        not_join_clauses.iter().zip(not_join_exclusion_sets.iter())
                    {
                        match nj_exclusion_entry {
                            Some((has_expr, key_vars, exclusion_set)) => {
                                if !has_expr {
                                    if key_vars.is_empty() {
                                        // Body bound no join vars: if exclusion set non-empty,
                                        // exclude all outer bindings (body always succeeds).
                                        if !exclusion_set.is_empty() {
                                            return false;
                                        }
                                        continue;
                                    }
                                    // Build probe key from outer binding using key_vars.
                                    // Normalize values so keyword entities match ref entities.
                                    let mut key: Vec<(String, Value)> = key_vars
                                        .iter()
                                        .filter_map(|v| {
                                            binding
                                                .get(v)
                                                .map(|val| (v.clone(), normalize_value(val)))
                                        })
                                        .collect();
                                    key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                                    if key.len() == key_vars.len() {
                                        if exclusion_set.contains(&key) {
                                            return false;
                                        }
                                        continue;
                                    }
                                    // Outer binding underspecified relative to the not-join body —
                                    // fall back to the slow path so the body is correctly evaluated.
                                    if evaluate_not_join(
                                        join_vars,
                                        nj_clauses,
                                        binding,
                                        filtered_facts.clone(),
                                        &registry,
                                    ) {
                                        return false;
                                    }
                                    continue;
                                }
                                // Fall through to slow path if expr clauses present.
                                if evaluate_not_join(
                                    join_vars,
                                    nj_clauses,
                                    binding,
                                    filtered_facts.clone(),
                                    &registry,
                                ) {
                                    return false;
                                }
                            }
                            None => {
                                if evaluate_not_join(
                                    join_vars,
                                    nj_clauses,
                                    binding,
                                    filtered_facts.clone(),
                                    &registry,
                                ) {
                                    return false;
                                }
                            }
                        }
                    }
                    true
                })
                .collect()
        };

        let results =
            apply_post_processing(not_filtered, &query.find, &query.with_vars, &registry)?;

        Ok(QueryResult::QueryResults {
            vars: query.find.iter().map(|s| s.display_name()).collect(),
            results,
        })
    }

    /// Execute a query that uses recursive rules
    fn execute_query_with_rules(&self, query: DatalogQuery) -> Result<QueryResult> {
        // Extract ALL predicates (including inside not bodies) so the StratifiedEvaluator
        // evaluates every referenced rule. This is needed for not-post-filter to work.
        let all_rule_invocations = query.get_rule_invocations();
        let predicates: Vec<String> = all_rule_invocations
            .iter()
            .map(|(pred, _)| pred.clone())
            .collect();

        // Compute query-level valid_at value for :db/valid-at pseudo-attribute binding.
        let now = self.read_now();
        let valid_at_value = match &query.valid_at {
            Some(ValidAt::Timestamp(t)) => Value::Integer(*t),
            Some(ValidAt::AnyValidTime) => Value::Null,
            Some(ValidAt::Slot(_)) => {
                return Err(anyhow!(
                    "internal: unsubstituted :valid-at bind slot reached the executor"
                ));
            }
            None => Value::Integer(now),
        };

        // Hard-error: per-fact pseudo-attrs require :any-valid-time.
        if query_uses_per_fact_pseudo_attr(&query)
            && !matches!(query.valid_at, Some(ValidAt::AnyValidTime))
        {
            return Err(anyhow!(
                "temporal pseudo-attributes :db/valid-from, :db/valid-to, :db/tx-count, and \
                 :db/tx-id require :any-valid-time; add :any-valid-time to your query"
            ));
        }

        // Apply temporal filters before evaluating recursive rules
        let filtered_facts = self.filter_facts_for_query(&query)?;

        // Convert to FactStorage for StratifiedEvaluator (needs mutable accumulation)
        // TODO (post-1.0): use FactStorage::new_noindex() once profiling confirms rules-path
        // index rebuild is also a bottleneck.
        let filtered_storage = FactStorage::new();
        for fact in filtered_facts.iter().cloned() {
            filtered_storage.load_fact(fact)?;
        }

        // Create StratifiedEvaluator — handles negation, stratification, and positive-only rules
        let evaluator = StratifiedEvaluator::new(
            filtered_storage,
            self.rules.clone(),
            self.functions.clone(),
            1000, // max iterations
            self.max_derived_facts,
            self.max_results,
        );

        let derived_storage = evaluator.evaluate(&predicates)?;

        // Compute derived_facts Arc once; reuse for plan loop, or-clauses and not-post-filter.
        // Must use derived_storage (includes rule-derived facts), not filtered_facts (base only).
        let derived_facts: Arc<[Fact]> =
            Arc::from(derived_storage.get_asserted_facts().unwrap_or_default());

        let matcher =
            PatternMatcher::from_slice_with_valid_at(derived_facts.clone(), valid_at_value.clone());

        // Acquire function registry before the plan loop — needed for inline Expr evaluation.
        let registry = self
            .functions
            .read()
            .map_err(|_| anyhow!("functions lock poisoned"))?;

        // Pre-validate UDF predicate names.
        for clause in &query.where_clauses {
            if let WhereClause::Expr {
                expr: Expr::UnaryOp(UnaryOp::Udf(name), _),
                ..
            } = clause
                && registry.get_predicate(name).is_none()
            {
                anyhow::bail!("unknown predicate: '{}'", name);
            }
        }

        // Collect Pattern and Expr top-level clauses for the planner.
        // Rule invocations are converted to WhereClause::Pattern against derived_storage.
        let mut plan_clauses: Vec<WhereClause> = query
            .where_clauses
            .iter()
            .filter(|c| matches!(c, WhereClause::Pattern(_) | WhereClause::Expr { .. }))
            .cloned()
            .collect();

        for (predicate, args) in query.get_top_level_rule_invocations() {
            let pattern = match args.len() {
                1 => {
                    #[allow(clippy::indexing_slicing)]
                    let entity = args[0].clone();
                    Pattern::new(
                        entity,
                        EdnValue::Keyword(format!(":{}", predicate)),
                        EdnValue::Symbol("?_rule_value".to_string()),
                    )
                }
                2 => {
                    #[allow(clippy::indexing_slicing)]
                    let entity = args[0].clone();
                    #[allow(clippy::indexing_slicing)]
                    let value = args[1].clone();
                    Pattern::new(entity, EdnValue::Keyword(format!(":{}", predicate)), value)
                }
                n => {
                    return Err(anyhow!(
                        "Rule invocation '{}' must have 1 or 2 arguments, got {}",
                        predicate,
                        n
                    ));
                }
            };
            plan_clauses.push(WhereClause::Pattern(pattern));
        }

        let planned = optimizer::plan(plan_clauses, &self.indexes);

        // Process planned clauses in order: Pattern → expand, Expr → filter/extend.
        let mut bindings: Vec<Binding> = vec![Binding::new()];
        for (clause, hint) in planned {
            match clause {
                WhereClause::Pattern(p) => {
                    bindings = matcher.match_with_hint_seeded(
                        bindings,
                        &p,
                        hint.as_ref().unwrap_or(&optimizer::IndexHint::Eavt),
                    );
                }
                WhereClause::Expr { expr, binding: out } => {
                    bindings = bindings
                        .into_iter()
                        .filter_map(|mut b| match eval_expr(&expr, &b, Some(&registry)) {
                            Ok(v) => {
                                if let Some(var) = &out {
                                    b.insert(var.clone(), v);
                                    Some(b)
                                } else if is_truthy(&v) {
                                    Some(b)
                                } else {
                                    None
                                }
                            }
                            Err(_) => None,
                        })
                        .collect();
                }
                _ => {}
            }
        }

        // Apply Or/OrJoin clauses against derived facts (rules already evaluated)
        let rules_guard = self
            .rules
            .read()
            .map_err(|_| anyhow!("rules lock poisoned"))?;
        let bindings = apply_or_clauses(
            &query.where_clauses,
            bindings,
            derived_facts.clone(),
            &rules_guard,
            query.as_of.clone(),
            query.valid_at.clone(),
            &registry,
        )?;
        drop(rules_guard);

        // Apply not-post-filter for WhereClause::Not and WhereClause::NotJoin clauses
        // in the query body. (The StratifiedEvaluator handles `not`/`not-join` in rule
        // bodies; this handles them appearing directly in the query body alongside rule
        // invocations.)
        #[cfg_attr(feature = "wasm", allow(unused_mut))]
        let mut not_clauses: Vec<&Vec<WhereClause>> = query
            .where_clauses
            .iter()
            .filter_map(|c| match c {
                WhereClause::Not(inner) => Some(inner),
                _ => None,
            })
            .collect();

        #[cfg_attr(feature = "wasm", allow(unused_mut))]
        let mut not_join_clauses: Vec<(Vec<String>, Vec<WhereClause>)> = query
            .where_clauses
            .iter()
            .filter_map(|c| match c {
                WhereClause::NotJoin { join_vars, clauses } => {
                    Some((join_vars.clone(), clauses.clone()))
                }
                _ => None,
            })
            .collect();

        // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
        #[cfg(not(feature = "wasm"))]
        not_clauses.sort_by_key(|body| optimizer::clause_cost(&WhereClause::Not(body.to_vec())));
        // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
        #[cfg(not(feature = "wasm"))]
        not_join_clauses.sort_by_key(|(vars, clauses)| {
            optimizer::clause_cost(&WhereClause::NotJoin {
                join_vars: vars.clone(),
                clauses: clauses.clone(),
            })
        });

        let not_filtered: Vec<_> = if not_clauses.is_empty() && not_join_clauses.is_empty() {
            bindings
        } else {
            bindings
                .into_iter()
                .filter(|binding| {
                    for not_body in &not_clauses {
                        // Collect pattern and rule-invocation clauses into patterns.
                        let substituted: Vec<Pattern> = not_body
                            .iter()
                            .filter_map(|c| match c {
                                WhereClause::Pattern(p) => {
                                    Some(crate::query::datalog::evaluator::substitute_pattern(
                                        p, binding,
                                    ))
                                }
                                WhereClause::RuleInvocation { predicate, args } => {
                                    // Convert rule invocation to a pattern against derived storage.
                                    // Apply the current binding to any variables in args first.
                                    let resolved_args: Vec<EdnValue> = args
                                        .iter()
                                        .map(|a| match a {
                                            EdnValue::Symbol(s) if s.starts_with('?') => {
                                                // Look up the bound value and convert back to EdnValue
                                                binding
                                                    .get(s)
                                                    .map(|v| match v {
                                                        Value::Keyword(k) => {
                                                            EdnValue::Keyword(k.clone())
                                                        }
                                                        Value::String(s) => {
                                                            EdnValue::String(s.clone())
                                                        }
                                                        Value::Integer(i) => EdnValue::Integer(*i),
                                                        Value::Float(f) => EdnValue::Float(*f),
                                                        Value::Boolean(b) => EdnValue::Boolean(*b),
                                                        Value::Ref(u) => EdnValue::Uuid(*u),
                                                        Value::Null => EdnValue::Nil,
                                                    })
                                                    .unwrap_or_else(|| a.clone())
                                            }
                                            other => other.clone(),
                                        })
                                        .collect();
                                    // Safety: match arms guarantee len()==1 or len()==2.
                                    let pattern = match resolved_args.len() {
                                        1 => {
                                            #[allow(clippy::indexing_slicing)]
                                            let entity = resolved_args[0].clone();
                                            Pattern::new(
                                                entity,
                                                EdnValue::Keyword(format!(":{}", predicate)),
                                                EdnValue::Symbol("?_rule_value".to_string()),
                                            )
                                        }
                                        2 => {
                                            #[allow(clippy::indexing_slicing)]
                                            let entity = resolved_args[0].clone();
                                            #[allow(clippy::indexing_slicing)]
                                            let value = resolved_args[1].clone();
                                            Pattern::new(
                                                entity,
                                                EdnValue::Keyword(format!(":{}", predicate)),
                                                value,
                                            )
                                        }
                                        _ => return None,
                                    };
                                    Some(crate::query::datalog::evaluator::substitute_pattern(
                                        &pattern, binding,
                                    ))
                                }
                                _ => None,
                            })
                            .collect();

                        // Compute not_bindings: if no patterns, seed with current binding.
                        let m = PatternMatcher::from_slice_with_valid_at(
                            derived_facts.clone(),
                            valid_at_value.clone(),
                        );
                        let mut not_bindings: Vec<Binding> = if substituted.is_empty() {
                            vec![binding.clone()]
                        } else {
                            m.match_patterns(&substituted)
                                .into_iter()
                                .map(|mut nb| {
                                    for (k, v) in binding {
                                        nb.entry(k.clone()).or_insert_with(|| v.clone());
                                    }
                                    nb
                                })
                                .collect()
                        };

                        // Apply Expr clauses from the not body.
                        // Errors (e.g. unknown UDF predicate) are treated as "no match".
                        not_bindings = apply_expr_clauses(not_bindings, not_body, &registry)
                            .unwrap_or_default();
                        if !not_bindings.is_empty() {
                            return false; // not condition violated
                        }
                    }
                    // Use the already-acquired registry instead of re-acquiring the lock.
                    for (join_vars, nj_clauses) in &not_join_clauses {
                        if evaluate_not_join(
                            join_vars,
                            nj_clauses,
                            binding,
                            derived_facts.clone(),
                            &registry,
                        ) {
                            return false;
                        }
                    }
                    true
                })
                .collect()
        };

        let results =
            apply_post_processing(not_filtered, &query.find, &query.with_vars, &registry)?;

        Ok(QueryResult::QueryResults {
            vars: query.find.iter().map(|s| s.display_name()).collect(),
            results,
        })
    }

    /// Execute a rule command: register the rule for later use
    fn execute_rule(&self, rule: Rule) -> Result<QueryResult> {
        // Extract predicate name from rule head
        // Head format: (predicate ?arg1 ?arg2 ...)
        let predicate = self.extract_predicate(&rule)?;

        // Register the rule
        self.rules
            .write()
            .map_err(|_| anyhow!("rules lock poisoned"))?
            .register_rule(predicate, rule)?;

        Ok(QueryResult::Ok)
    }

    /// Extract the predicate name from a rule head
    fn extract_predicate(&self, rule: &Rule) -> Result<String> {
        if rule.head.is_empty() {
            return Err(anyhow!("Rule head cannot be empty"));
        }

        // Safety: is_empty() check above guarantees index 0 exists.
        #[allow(clippy::indexing_slicing)]
        match &rule.head[0] {
            EdnValue::Symbol(s) => Ok(s.clone()),
            _ => Err(anyhow!(
                "Rule head must start with a symbol (predicate name)"
            )),
        }
    }

    /// Get the underlying storage (for testing)
    #[allow(dead_code)]
    pub fn storage(&self) -> &FactStorage {
        &self.storage
    }

    /// Get the rule registry (for testing)
    #[cfg(test)]
    pub fn rules(&self) -> Arc<RwLock<RuleRegistry>> {
        self.rules.clone()
    }
}

/// Normalize a `Value` for use as a hash-join key.
///
/// Entity keywords (`:foo`) and entity refs (`Value::Ref(uuid)`) represent the same
/// entity but appear as different variants depending on whether the value was stored in
/// the entity position (→ `Ref`) or the value position (→ `Keyword`) of a fact.
/// Normalize both to `Value::Ref` so that exclusion-set probes work correctly across
/// these two representations.
fn normalize_value(v: &Value) -> Value {
    if let Value::Keyword(k) = v {
        use crate::query::datalog::matcher::edn_to_entity_id;
        use crate::query::datalog::types::EdnValue;
        if let Ok(uuid) = edn_to_entity_id(&EdnValue::Keyword(k.clone())) {
            return Value::Ref(uuid);
        }
    }
    v.clone()
}

/// Evaluate a `not` body against the current outer binding.
///
/// Returns true if the body "matches" (i.e., the outer binding should be excluded).
fn not_body_matches(
    not_body: &[WhereClause],
    outer: &Binding,
    storage: Arc<[Fact]>,
    valid_at: Value,
    registry: &FunctionRegistry,
) -> bool {
    use crate::query::datalog::evaluator::substitute_pattern;

    let patterns: Vec<_> = not_body
        .iter()
        .filter_map(|c| match c {
            WhereClause::Pattern(p) => Some(substitute_pattern(p, outer)),
            // INVARIANT: not_body_matches is only called from execute_query, which is
            // only reached when query.uses_rules() is false. uses_rules() descends into
            // Not bodies via rule_invocations(), so any not body containing a
            // RuleInvocation is routed to execute_query_with_rules instead.
            // WhereClause::Expr clauses are handled by apply_expr_clauses below.
            _ => None,
        })
        .collect();

    let matcher = crate::query::datalog::matcher::PatternMatcher::from_slice_with_valid_at(
        storage.clone(),
        valid_at,
    );
    let mut not_bindings: Vec<Binding> = if patterns.is_empty() {
        // Expr-only not body: start with the outer binding so variables resolve.
        vec![outer.clone()]
    } else {
        // Merge outer binding with pattern-match results.
        matcher
            .match_patterns(&patterns)
            .into_iter()
            .map(|mut nb| {
                for (k, v) in outer {
                    nb.entry(k.clone()).or_insert_with(|| v.clone());
                }
                nb
            })
            .collect()
    };

    // Apply Expr clauses from the not body.
    // Errors (e.g. unknown UDF predicate) are treated as "no match" — same as an
    // empty result — so the outer row is kept (not-condition not violated).
    not_bindings = apply_expr_clauses(not_bindings, not_body, registry).unwrap_or_default();
    !not_bindings.is_empty()
}

/// Extract plain variable values from bindings (non-aggregate path).
fn extract_variables(
    bindings: Vec<std::collections::HashMap<String, Value>>,
    find_specs: &[FindSpec],
) -> Vec<Vec<Value>> {
    let mut results = Vec::new();
    for binding in bindings {
        let mut row = Vec::new();
        for spec in find_specs {
            if let Some(value) = binding.get(spec.var()) {
                row.push(value.clone());
            } else {
                break;
            }
        }
        if row.len() == find_specs.len() {
            results.push(row);
        }
    }
    results
}

type Binding = std::collections::HashMap<String, Value>;
/// Internal type alias for pre-computed not-body exclusion sets.
type NotExclusionEntry = Option<(bool, std::collections::HashSet<Vec<(String, Value)>>)>;
/// Internal type alias for pre-computed not-join exclusion sets.
type NotJoinExclusionEntry = Option<(
    bool,
    Vec<String>,
    std::collections::HashSet<Vec<(String, Value)>>,
)>;

/// Unified post-processing: handles plain-variable extraction, aggregation,
/// window functions, and mixed (aggregate + window) queries.
///
/// - Plain variables only → `extract_variables` (no change from current path).
/// - Aggregates only → group-by collapse, then project.
/// - Windows only → partition/sort/accumulate per spec, then project.
/// - Mixed → aggregate collapses first, window runs over collapsed rows.
fn apply_post_processing(
    bindings: Vec<Binding>,
    find_specs: &[FindSpec],
    with_vars: &[String],
    registry: &FunctionRegistry,
) -> Result<Vec<Vec<Value>>> {
    let has_aggregates = find_specs
        .iter()
        .any(|s| matches!(s, FindSpec::Aggregate { .. }));
    let has_windows = find_specs.iter().any(|s| matches!(s, FindSpec::Window(_)));

    if !has_aggregates && !has_windows {
        return Ok(extract_variables(bindings, find_specs));
    }

    // Step 1: Aggregate (collapses rows, produces binding maps).
    let mut working: Vec<Binding> = if has_aggregates {
        compute_aggregation(bindings, find_specs, with_vars, registry)?
    } else {
        bindings
    };

    // Step 2: Window functions (annotate each row, no collapse).
    if has_windows {
        apply_window_functions(&mut working, find_specs, registry)?;
    }

    // Step 3: Project to output rows in find-spec order.
    Ok(project_find_specs(&working, find_specs))
}

/// Group bindings by non-aggregate find vars + with_vars, apply aggregate functions,
/// return one binding map per group. Aggregate results stored under `"__agg_{i}"`.
fn compute_aggregation(
    bindings: Vec<Binding>,
    find_specs: &[FindSpec],
    with_vars: &[String],
    registry: &FunctionRegistry,
) -> Result<Vec<Binding>> {
    let has_grouping_vars = find_specs
        .iter()
        .any(|s| matches!(s, FindSpec::Variable(_)));

    // Special case: zero bindings + all-count specs → one zero row.
    if bindings.is_empty() {
        let all_count = !has_grouping_vars
            && find_specs.iter().all(|s| {
                matches!(s, FindSpec::Aggregate { func, .. }
                    if func == "count" || func == "count-distinct")
            });
        if all_count {
            let mut b = Binding::new();
            for (i, _) in find_specs.iter().enumerate() {
                b.insert(format!("__agg_{}", i), Value::Integer(0));
            }
            return Ok(vec![b]);
        }
        return Ok(vec![]);
    }

    // In a mixed aggregate+window query, :with vars must NOT be added to the
    // grouping key. The window phase runs after aggregation, so :with vars that
    // are used only by window specs (var, order_by) would otherwise inflate the
    // number of groups. Even :with vars used by aggregate specs (e.g. ?e in
    // count(?e)) should not split groups — the aggregate operates over all rows
    // in the base group determined by the Variable find specs.
    let has_windows = find_specs.iter().any(|s| matches!(s, FindSpec::Window(_)));

    // Grouping key = Variable find specs (in find order).
    // In pure-aggregate queries, also include with_vars (Datomic semantics: :with
    // prevents pre-aggregation de-duplication by adding vars to the group key).
    let mut group_var_names: Vec<&str> = find_specs
        .iter()
        .filter_map(|s| match s {
            FindSpec::Variable(v) => Some(v.as_str()),
            _ => None,
        })
        .collect();
    if !has_windows {
        // Pure aggregate: with_vars add to grouping key.
        group_var_names.extend(with_vars.iter().map(|s| s.as_str()));
    }

    // Group using BTreeMap keyed by group key (O(log g) instead of O(g) per binding).
    use std::collections::BTreeMap;
    let mut groups: BTreeMap<Vec<Value>, Vec<Binding>> = BTreeMap::new();
    for b in bindings {
        let key: Vec<Value> = group_var_names
            .iter()
            .map(|v| b.get(*v).cloned().unwrap_or(Value::Null))
            .collect();
        groups.entry(key).or_default().push(b);
    }

    // Build a position map for Variable specs only (indices 0..n_vars in the key vector).
    // with_vars occupy key positions n_vars..end and are used only for grouping, not for output.
    // Map of Variable spec name → its index in the group key Vec.
    let mut group_key_idx: std::collections::HashMap<&str, usize> =
        std::collections::HashMap::new();
    {
        let mut var_pos = 0usize;
        for spec in find_specs {
            if let FindSpec::Variable(v) = spec {
                group_key_idx.insert(v.as_str(), var_pos);
                var_pos += 1;
            }
        }
    }

    let mut results: Vec<Binding> = Vec::new();
    for (key, group_bindings) in groups.iter() {
        let mut binding = Binding::new();
        let mut skip = false;

        // Plain variable values from group key.
        for (v, &idx) in &group_key_idx {
            if let Some(val) = key.get(idx) {
                binding.insert((*v).to_string(), val.clone());
            }
        }

        // Aggregate values stored under "__agg_{i}".
        for (i, spec) in find_specs.iter().enumerate() {
            if let FindSpec::Aggregate { func, var } = spec {
                let non_null: Vec<&Value> = group_bindings
                    .iter()
                    .filter_map(|b| b.get(var.as_str()))
                    .filter(|v| !matches!(v, Value::Null))
                    .collect();
                let agg_val: anyhow::Result<Value> = match registry.get(func.as_str()) {
                    Some(desc) if desc.is_builtin => {
                        // Built-in: use batch path which enforces strict type-error semantics.
                        apply_builtin_aggregate(func, &non_null)
                    }
                    Some(desc) => {
                        if let AggImpl::Udf(ops) = &desc.impl_ {
                            if non_null.is_empty() {
                                Ok(Value::Null)
                            } else {
                                let mut acc = (ops.init)();
                                for v in &non_null {
                                    (ops.step)(&mut acc, v);
                                }
                                Ok((ops.finalise)(&acc, non_null.len()))
                            }
                        } else {
                            // AggImpl::Builtin with is_builtin=false shouldn't happen
                            apply_builtin_aggregate(func, &non_null)
                        }
                    }
                    None => Err(anyhow::anyhow!("unknown aggregate function: '{}'", func)),
                };
                match agg_val {
                    Ok(v) => {
                        binding.insert(format!("__agg_{}", i), v);
                    }
                    Err(e) => {
                        let msg = e.to_string();
                        if msg.contains("no non-null values in group") {
                            skip = true;
                            break;
                        }
                        return Err(e);
                    }
                }
            }
        }

        if !skip {
            results.push(binding);
        }
    }

    Ok(results)
}

/// Compute window function values for each row and store under `"__win_{i}"`.
/// Modifies `bindings` in place.
fn apply_window_functions(
    bindings: &mut [Binding],
    find_specs: &[FindSpec],
    registry: &FunctionRegistry,
) -> Result<()> {
    for (i, spec) in find_specs.iter().enumerate() {
        let FindSpec::Window(ws) = spec else {
            continue;
        };
        let key = format!("__win_{}", i);

        // Build partitions: (partition_key, sorted row indices).
        let mut partitions: HashMap<Option<Value>, Vec<usize>> = HashMap::new();
        for (row_idx, binding) in bindings.iter().enumerate() {
            let part_key = ws
                .partition_by
                .as_ref()
                .and_then(|pv| binding.get(pv))
                .cloned();
            partitions.entry(part_key).or_default().push(row_idx);
        }

        // For each partition: sort, compute window values, write back.
        for row_indices in partitions.values_mut() {
            // Pre-extract order_by values into a contiguous Vec so the sort
            // comparator never touches the HashMap — O(n) lookups here instead
            // of O(n log n) random HashMap accesses inside sort_by.
            // Safety: row_indices are populated from 0..bindings.len() enumeration above,
            // so all indices are valid.
            #[allow(clippy::indexing_slicing)]
            let mut keyed: Vec<(Value, usize)> = row_indices
                .iter()
                .map(|&i| {
                    let k = bindings[i]
                        .get(&ws.order_by)
                        .cloned()
                        .unwrap_or(Value::Null);
                    (k, i)
                })
                .collect();
            keyed.sort_by(|(a, _), (b, _)| {
                let cmp = value_cmp(a, b);
                match ws.order {
                    Order::Asc => cmp,
                    Order::Desc => cmp.reverse(),
                }
            });
            // Rewrite row_indices in sorted order for the write-back step.
            for (dest, (_, src)) in row_indices.iter_mut().zip(keyed.iter()) {
                *dest = *src;
            }

            // Compute one window value per row in sorted order.
            let window_values: Vec<Value> = match ws.func {
                WindowFunc::RowNumber => {
                    let mut values = Vec::with_capacity(keyed.len());
                    for pos in 1..=keyed.len() {
                        values.push(Value::Integer(
                            i64::try_from(pos).map_err(|_| anyhow!("row number overflow"))?,
                        ));
                    }
                    values
                }

                WindowFunc::Rank => {
                    // Reuse pre-extracted keys for tie-detection — no extra HashMap lookups.
                    let mut values = Vec::with_capacity(keyed.len());
                    let mut rank = 1i64;
                    let mut prev: Option<&Value> = None;
                    for (row_num, (key, _)) in (1i64..).zip(keyed.iter()) {
                        if prev != Some(key) {
                            rank = row_num;
                            prev = Some(key);
                        }
                        values.push(Value::Integer(rank));
                    }
                    values
                }

                _ => {
                    // Accumulator-based: built-ins (sum, count, min, max, avg) and UDF aggregates.
                    // UDF aggregates (WindowFunc::Udf) also route here via registry lookup.
                    let func_name = ws.func_name();
                    let desc = registry.get(&func_name).ok_or_else(|| {
                        anyhow::anyhow!(
                            "unknown window function '{}' — register it with register_aggregate() before querying",
                            func_name
                        )
                    })?;

                    let mut values = Vec::with_capacity(keyed.len());
                    // Safety: row_idx values come from enumerate() over bindings, so indices are valid.
                    #[allow(clippy::indexing_slicing)]
                    match &desc.impl_ {
                        AggImpl::Builtin(ops) => {
                            let mut acc = (ops.init)();
                            for (_, row_idx) in keyed.iter() {
                                let val = ws
                                    .var
                                    .as_ref()
                                    .and_then(|v| bindings[*row_idx].get(v))
                                    .unwrap_or(&Value::Null);
                                (ops.step)(&mut acc, val);
                                values.push((ops.finalise)(&acc));
                            }
                        }
                        AggImpl::Udf(ops) => {
                            let mut acc = (ops.init)();
                            let mut row_count = 0usize;
                            for (_, row_idx) in keyed.iter() {
                                let val = ws
                                    .var
                                    .as_ref()
                                    .and_then(|v| bindings[*row_idx].get(v))
                                    .unwrap_or(&Value::Null);
                                (ops.step)(&mut acc, val);
                                row_count += 1;
                                values.push((ops.finalise)(&acc, row_count));
                            }
                        }
                    }
                    values
                }
            };

            // Write window values back to rows.
            // Safety: row_idx values come from enumerate() over bindings, so indices are valid.
            for (&row_idx, window_val) in row_indices.iter().zip(window_values) {
                #[allow(clippy::indexing_slicing)]
                bindings[row_idx].insert(key.clone(), window_val);
            }
        }
    }
    Ok(())
}

/// Project binding maps to output rows in find-spec order.
fn project_find_specs(bindings: &[Binding], find_specs: &[FindSpec]) -> Vec<Vec<Value>> {
    let mut results = Vec::new();
    for binding in bindings {
        let mut row = Vec::new();
        let mut complete = true;
        for (i, spec) in find_specs.iter().enumerate() {
            let val = match spec {
                FindSpec::Variable(v) => binding.get(v).cloned(),
                FindSpec::Aggregate { .. } => binding.get(&format!("__agg_{}", i)).cloned(),
                FindSpec::Window(_) => binding.get(&format!("__win_{}", i)).cloned(),
            };
            match val {
                Some(v) => row.push(v),
                // Invariant: all __agg_{i} and __win_{i} keys are populated for non-skipped rows.
                // None here only occurs for skipped aggregate groups (e.g. min/max on all-null input).
                None => {
                    complete = false;
                    break;
                }
            }
        }
        if complete {
            results.push(row);
        }
    }
    results
}

/// Evaluate a single branch of an `or`/`or-join` against incoming bindings.
///
/// Processing order (note: top-level execute_query now uses an interleaved plan loop
/// where Expr clauses are pushed down inline; branches retain their own Expr post-pass):
/// 1. Pattern/RuleInvocation → match_patterns_seeded
/// 2. Nested Or/OrJoin → apply_or_clauses (recursive)
/// 3. Not/NotJoin → post-filter
/// 4. Expr → apply_expr_clauses
pub(crate) fn evaluate_branch(
    branch: &[WhereClause],
    incoming: Vec<Binding>,
    storage: Arc<[Fact]>,
    rules: &crate::query::datalog::rules::RuleRegistry,
    as_of: Option<AsOf>,
    valid_at: Option<ValidAt>,
    registry: &FunctionRegistry,
) -> anyhow::Result<Vec<Binding>> {
    use crate::query::datalog::evaluator::rule_invocation_to_pattern;
    use crate::query::datalog::matcher::PatternMatcher;

    if incoming.is_empty() {
        return Ok(vec![]);
    }

    // Compute valid_at_value for pseudo-attribute binding in this branch.
    let branch_valid_at_value = match &valid_at {
        Some(ValidAt::Timestamp(t)) => Value::Integer(*t),
        Some(ValidAt::AnyValidTime) => Value::Null,
        Some(ValidAt::Slot(_)) => {
            return Err(anyhow!(
                "internal: unsubstituted :valid-at bind slot reached the executor"
            ));
        }
        None => Value::Integer(tx_id_now().cast_signed()),
    };

    // Step 1: Collect Pattern and RuleInvocation clauses
    let patterns: Vec<Pattern> = branch
        .iter()
        .filter_map(|c| match c {
            WhereClause::Pattern(p) => Some(p.clone()),
            WhereClause::RuleInvocation { predicate, args } => {
                rule_invocation_to_pattern(predicate, args).ok()
            }
            _ => None,
        })
        .collect();

    let matcher =
        PatternMatcher::from_slice_with_valid_at(storage.clone(), branch_valid_at_value.clone());
    let bindings = if patterns.is_empty() {
        incoming
    } else {
        matcher.match_patterns_seeded(&patterns, incoming)
    };

    if bindings.is_empty() {
        return Ok(vec![]);
    }

    // Step 2: Nested Or/OrJoin
    let bindings = apply_or_clauses(
        branch,
        bindings,
        storage.clone(),
        rules,
        as_of.clone(),
        valid_at.clone(),
        registry,
    )?;

    if bindings.is_empty() {
        return Ok(vec![]);
    }

    // Step 3: Not/NotJoin post-filter
    let not_clauses: Vec<&Vec<WhereClause>> = branch
        .iter()
        .filter_map(|c| match c {
            WhereClause::Not(inner) => Some(inner),
            _ => None,
        })
        .collect();

    let not_join_clauses: Vec<(Vec<String>, Vec<WhereClause>)> = branch
        .iter()
        .filter_map(|c| match c {
            WhereClause::NotJoin { join_vars, clauses } => {
                Some((join_vars.clone(), clauses.clone()))
            }
            _ => None,
        })
        .collect();

    let bindings = if not_clauses.is_empty() && not_join_clauses.is_empty() {
        bindings
    } else {
        bindings
            .into_iter()
            .filter(|binding| {
                for not_body in &not_clauses {
                    if not_body_matches(
                        not_body,
                        binding,
                        storage.clone(),
                        branch_valid_at_value.clone(),
                        registry,
                    ) {
                        return false;
                    }
                }
                for (join_vars, nj_clauses) in &not_join_clauses {
                    if evaluate_not_join(join_vars, nj_clauses, binding, storage.clone(), registry)
                    {
                        return false;
                    }
                }
                true
            })
            .collect()
    };

    // Step 4: Expr clauses
    let bindings = apply_expr_clauses(bindings, branch, registry)?;

    Ok(bindings)
}

/// Apply all Or/OrJoin clauses from `clauses` to `bindings` in sequence.
///
/// Non-Or/OrJoin clauses are ignored (handled elsewhere).
/// For `Or`: union results from all branches (deduplicated by full binding map).
/// For `OrJoin`: union results, then project out branch-private variables.
pub(crate) fn apply_or_clauses(
    clauses: &[WhereClause],
    mut bindings: Vec<Binding>,
    storage: Arc<[Fact]>,
    rules: &crate::query::datalog::rules::RuleRegistry,
    as_of: Option<AsOf>,
    valid_at: Option<ValidAt>,
    registry: &FunctionRegistry,
) -> anyhow::Result<Vec<Binding>> {
    for clause in clauses {
        match clause {
            WhereClause::Or(branches) => {
                let sorted_or_branches: Vec<&Vec<WhereClause>> = {
                    #[cfg_attr(feature = "wasm", allow(unused_mut))]
                    let mut b: Vec<&Vec<WhereClause>> = branches.iter().collect();
                    // Sort branches by cost ascending so cheaper branches evaluate first.
                    // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
                    // Note: all branches still evaluated (no short-circuit); ordering is
                    // infrastructure for issue #250.
                    #[cfg(not(feature = "wasm"))]
                    b.sort_by_key(|br| optimizer::branch_cost(br));
                    b
                };
                // If any branch contains Not/NotJoin clauses (which need bound variables
                // from the outer scope to evaluate correctly), fall back to the classic
                // seeded-branch evaluation to preserve correctness.
                let any_branch_has_not = sorted_or_branches.iter().any(|b| {
                    b.iter()
                        .any(|c| matches!(c, WhereClause::Not(_) | WhereClause::NotJoin { .. }))
                });

                if any_branch_has_not {
                    // Classic O(N·B) seeded evaluation: each incoming binding seeds each branch.
                    let mut seen: std::collections::HashSet<Vec<(String, Value)>> =
                        std::collections::HashSet::new();
                    let mut result: Vec<Binding> = Vec::new();
                    for branch in &sorted_or_branches {
                        let branch_result = evaluate_branch(
                            branch,
                            bindings.clone(),
                            storage.clone(),
                            rules,
                            as_of.clone(),
                            valid_at.clone(),
                            registry,
                        )?;
                        for b in branch_result {
                            let mut key: Vec<_> = b
                                .iter()
                                .filter(|(k, _)| !k.starts_with("__"))
                                .map(|(k, v)| (k.clone(), v.clone()))
                                .collect();
                            key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            if seen.insert(key) {
                                result.push(b);
                            }
                        }
                    }
                    bindings = result;
                    continue;
                }

                // Fast path: no Not/NotJoin in any branch.
                // Evaluate every branch from an empty seed (independent of incoming bindings),
                // then hash-join the union back onto incoming bindings on shared variables.
                let empty_seed: Vec<Binding> = vec![HashMap::new()];
                let mut union_bindings: Vec<Binding> = Vec::new();
                let mut seen_keys: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();

                for branch in &sorted_or_branches {
                    let branch_result = evaluate_branch(
                        branch,
                        empty_seed.clone(),
                        storage.clone(),
                        rules,
                        as_of.clone(),
                        valid_at.clone(),
                        registry,
                    )?;
                    for b in branch_result {
                        // Deduplicate on user-visible variables only (exclude internal `__` keys).
                        let mut key: Vec<_> = b
                            .iter()
                            .filter(|(k, _)| !k.starts_with("__"))
                            .map(|(k, v)| (k.clone(), v.clone()))
                            .collect();
                        key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                        if seen_keys.insert(key) {
                            union_bindings.push(b);
                        }
                    }
                }

                // Determine shared variable names: variables present in both
                // incoming bindings and branch results.
                // Exclude internal metadata keys (prefixed with `__`) — those are
                // fact-specific and differ between patterns for the same entity.
                let branch_var_names: std::collections::HashSet<&str> = union_bindings
                    .iter()
                    .flat_map(|b| b.keys().map(|k| k.as_str()))
                    .filter(|k| !k.starts_with("__"))
                    .collect();
                let shared_vars: Vec<String> = bindings
                    .iter()
                    .flat_map(|b| b.keys().cloned())
                    .collect::<std::collections::HashSet<_>>()
                    .into_iter()
                    .filter(|v| !v.starts_with("__") && branch_var_names.contains(v.as_str()))
                    .collect();

                if shared_vars.is_empty() {
                    // No shared user-visible variables between incoming bindings and branch
                    // results. This is semantically equivalent to a cross-join, which only
                    // makes sense when `bindings` carries no meaningful state yet (i.e. it
                    // is a single empty binding — the query start state — or all incoming
                    // variables are fact-metadata keys that the branch does not reference).
                    // In that case the cross-join degenerates to "replace with branch union",
                    // which is what the original seeded evaluation produced. Incoming bindings
                    // that carry user-visible variables but none matching any branch variable
                    // would be silently dropped here — that situation should not arise given
                    // that `or` must share at least one variable with the surrounding clause.
                    bindings = union_bindings;
                    continue;
                }

                // Build HashMap: shared-key tuple → Vec<branch Binding>
                let mut branch_map: HashMap<Vec<(String, Value)>, Vec<Binding>> = HashMap::new();
                for b in union_bindings {
                    let key: Vec<(String, Value)> = shared_vars
                        .iter()
                        .filter_map(|v| b.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    branch_map.entry(key).or_default().push(b);
                }

                // For each incoming binding, look up matching branch results and merge.
                let mut result: Vec<Binding> = Vec::new();
                let mut seen_result: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();
                for incoming in &bindings {
                    let key: Vec<(String, Value)> = shared_vars
                        .iter()
                        .filter_map(|v| incoming.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    if let Some(matches) = branch_map.get(&key) {
                        for branch_binding in matches {
                            // Merge: start with incoming, extend with branch-introduced vars.
                            let mut merged = incoming.clone();
                            for (k, v) in branch_binding {
                                merged.entry(k.clone()).or_insert_with(|| v.clone());
                            }
                            let mut dedup_key: Vec<_> =
                                merged.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                            dedup_key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            if seen_result.insert(dedup_key) {
                                result.push(merged);
                            }
                        }
                    }
                }
                bindings = result;
            }

            WhereClause::OrJoin {
                join_vars,
                branches,
            } => {
                let sorted_oj_branches: Vec<&Vec<WhereClause>> = {
                    #[cfg_attr(feature = "wasm", allow(unused_mut))]
                    let mut b: Vec<&Vec<WhereClause>> = branches.iter().collect();
                    // Sort branches by cost ascending so cheaper branches evaluate first.
                    // WASM omission: small datasets + determinism — see optimizer::selectivity_score().
                    // Note: all branches still evaluated (no short-circuit); ordering is
                    // infrastructure for issue #250.
                    #[cfg(not(feature = "wasm"))]
                    b.sort_by_key(|br| optimizer::branch_cost(br));
                    b
                };
                let outer_keys: std::collections::HashSet<String> =
                    bindings.iter().flat_map(|b| b.keys().cloned()).collect();

                // Defensive check: every join_var must be bound in the incoming scope.
                // The parser validates this, but guard here to avoid silent wrong results
                // if a join_var is missing from outer_keys (hash-join key would be partial).
                for jv in join_vars.iter() {
                    if !outer_keys.contains(jv.as_str()) {
                        anyhow::bail!("or-join variable {} is not bound in the incoming scope", jv);
                    }
                }

                let empty_seed: Vec<Binding> = vec![HashMap::new()];
                let mut projected: Vec<Binding> = Vec::new();
                let mut seen_proj: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();

                for branch in &sorted_oj_branches {
                    let branch_result = evaluate_branch(
                        branch,
                        empty_seed.clone(),
                        storage.clone(),
                        rules,
                        as_of.clone(),
                        valid_at.clone(),
                        registry,
                    )?;
                    for mut b in branch_result {
                        if !join_vars.iter().all(|v| b.contains_key(v)) {
                            continue;
                        }
                        // Project to outer_keys (preserves join_vars since join_vars ⊆ outer_keys)
                        b.retain(|k, _| outer_keys.contains(k));
                        let mut key: Vec<_> =
                            b.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                        key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                        if seen_proj.insert(key) {
                            projected.push(b);
                        }
                    }
                }

                // Build HashMap keyed on join_vars tuple.
                let mut branch_map: HashMap<Vec<(String, Value)>, Vec<Binding>> = HashMap::new();
                for b in projected {
                    let key: Vec<(String, Value)> = join_vars
                        .iter()
                        .filter_map(|v| b.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    branch_map.entry(key).or_default().push(b);
                }

                let mut result: Vec<Binding> = Vec::new();
                let mut seen_result: std::collections::HashSet<Vec<(String, Value)>> =
                    std::collections::HashSet::new();
                for incoming in &bindings {
                    let key: Vec<(String, Value)> = join_vars
                        .iter()
                        .filter_map(|v| incoming.get(v).map(|val| (v.clone(), val.clone())))
                        .collect();
                    if let Some(matches) = branch_map.get(&key) {
                        for branch_binding in matches {
                            let mut merged = incoming.clone();
                            for (k, v) in branch_binding {
                                merged.entry(k.clone()).or_insert_with(|| v.clone());
                            }
                            let mut dedup_key: Vec<_> =
                                merged.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
                            dedup_key.sort_unstable_by(|a, b| a.0.cmp(&b.0));
                            if seen_result.insert(dedup_key) {
                                result.push(merged);
                            }
                        }
                    }
                }
                bindings = result;
            }

            _ => {} // Other clause types handled elsewhere
        }
    }
    Ok(bindings)
}

/// Returns true for Boolean(true), non-zero Integer, non-zero Float.
/// All other Value variants (String, Keyword, Ref, Null, Float(0.0)) → false.
/// Note: `Float(-0.0)` is falsy because `-0.0 == 0.0` in IEEE 754.
pub(crate) fn is_truthy(v: &Value) -> bool {
    match v {
        Value::Boolean(b) => *b,
        Value::Integer(i) => *i != 0,
        Value::Float(f) => *f != 0.0,
        _ => false,
    }
}

/// Promote both values to f64 for numeric comparison / float arithmetic.
/// Returns Err(()) if either operand is not Integer or Float.
fn to_float_pair(l: &Value, r: &Value) -> Result<(f64, f64), ()> {
    let lf = match l {
        Value::Integer(i) => *i as f64,
        Value::Float(f) => *f,
        _ => return Err(()),
    };
    let rf = match r {
        Value::Integer(i) => *i as f64,
        Value::Float(f) => *f,
        _ => return Err(()),
    };
    Ok((lf, rf))
}

fn eval_binop(op: &BinOp, l: Value, r: Value) -> Result<Value, ()> {
    match op {
        // Structural equality — works for all Value variants; no type mismatch error.
        BinOp::Eq => return Ok(Value::Boolean(l == r)),
        BinOp::Neq => return Ok(Value::Boolean(l != r)),
        _ => {}
    }

    match op {
        // Numeric comparisons — require both numeric; int/float promotion via to_float_pair.
        BinOp::Lt | BinOp::Gt | BinOp::Lte | BinOp::Gte => {
            let (lf, rf) = to_float_pair(&l, &r)?;
            Ok(Value::Boolean(match op {
                BinOp::Lt => lf < rf,
                BinOp::Gt => lf > rf,
                BinOp::Lte => lf <= rf,
                BinOp::Gte => lf >= rf,
                #[allow(clippy::unreachable)]
                _ => unreachable!(),
            }))
        }

        // Arithmetic: integer-integer stays integer; any float promotes result to float.
        BinOp::Add | BinOp::Sub | BinOp::Mul | BinOp::Div => match (&l, &r) {
            (Value::Integer(a), Value::Integer(b)) => match op {
                BinOp::Add => Ok(Value::Integer(a.wrapping_add(*b))),
                BinOp::Sub => Ok(Value::Integer(a.wrapping_sub(*b))),
                BinOp::Mul => Ok(Value::Integer(a.wrapping_mul(*b))),
                BinOp::Div => {
                    if *b == 0 {
                        Err(())
                    } else {
                        Ok(Value::Integer(a / b))
                    }
                }
                #[allow(clippy::unreachable)]
                _ => unreachable!(),
            },
            _ => {
                let (lf, rf) = to_float_pair(&l, &r)?;
                match op {
                    BinOp::Add => {
                        let r = lf + rf;
                        if r.is_nan() {
                            Err(())
                        } else {
                            Ok(Value::Float(r))
                        }
                    }
                    BinOp::Sub => {
                        let r = lf - rf;
                        if r.is_nan() {
                            Err(())
                        } else {
                            Ok(Value::Float(r))
                        }
                    }
                    BinOp::Mul => {
                        let r = lf * rf;
                        if r.is_nan() {
                            Err(())
                        } else {
                            Ok(Value::Float(r))
                        }
                    }
                    BinOp::Div => {
                        if rf == 0.0 || rf.is_nan() {
                            Err(())
                        } else {
                            Ok(Value::Float(lf / rf))
                        }
                    }
                    #[allow(clippy::unreachable)]
                    _ => unreachable!(),
                }
            }
        },

        // String predicates — both operands must be String.
        BinOp::StartsWith => match (l, r) {
            (Value::String(s), Value::String(prefix)) => {
                Ok(Value::Boolean(s.starts_with(prefix.as_str())))
            }
            _ => Err(()),
        },
        BinOp::EndsWith => match (l, r) {
            (Value::String(s), Value::String(suffix)) => {
                Ok(Value::Boolean(s.ends_with(suffix.as_str())))
            }
            _ => Err(()),
        },
        BinOp::Contains => match (l, r) {
            (Value::String(s), Value::String(needle)) => {
                Ok(Value::Boolean(s.contains(needle.as_str())))
            }
            _ => Err(()),
        },
        BinOp::Matches { regex: re, .. } => match (l, r) {
            (Value::String(s), Value::String(_)) => Ok(Value::Boolean(re.is_match(&s))),
            _ => Err(()),
        },

        // Eq/Neq handled above
        BinOp::Eq | BinOp::Neq => unreachable!(),
    }
}

/// Evaluate an Expr against a binding map.
///
/// Returns `Err(())` on: unbound variable, type mismatch, division by zero, unknown UDF predicate.
pub(crate) fn eval_expr(
    expr: &Expr,
    binding: &std::collections::HashMap<String, Value>,
    registry: Option<&FunctionRegistry>,
) -> Result<Value, ()> {
    match expr {
        Expr::Var(v) => binding.get(v).cloned().ok_or(()),
        Expr::Lit(val) => Ok(val.clone()),
        Expr::UnaryOp(op, arg) => {
            let v = eval_expr(arg, binding, registry)?;
            match op {
                UnaryOp::StringQ => Ok(Value::Boolean(matches!(v, Value::String(_)))),
                UnaryOp::IntegerQ => Ok(Value::Boolean(matches!(v, Value::Integer(_)))),
                UnaryOp::FloatQ => Ok(Value::Boolean(matches!(v, Value::Float(_)))),
                UnaryOp::BooleanQ => Ok(Value::Boolean(matches!(v, Value::Boolean(_)))),
                UnaryOp::NilQ => Ok(Value::Boolean(matches!(v, Value::Null))),
                UnaryOp::Udf(name) => {
                    let desc = registry.and_then(|r| r.get_predicate(name)).ok_or(())?;
                    Ok(Value::Boolean((desc.f)(&v)))
                }
            }
        }
        Expr::BinOp(op, lhs, rhs) => {
            let l = eval_expr(lhs, binding, registry)?;
            let r = eval_expr(rhs, binding, registry)?;
            eval_binop(op, l, r)
        }
        Expr::Slot(_) => {
            // Unsubstituted bind slot — treat as eval error (unbound variable equivalent).
            Err(())
        }
    }
}

/// Apply all WhereClause::Expr clauses from `where_clauses` to `bindings`.
///
/// Filter-form (`binding: None`) drops the row if the expr is not truthy or errors.
/// Binding-form (`binding: Some(var)`) extends the row with the computed value.
/// Type mismatches and errors silently drop the row.
///
/// Pre-validates UDF predicate names: returns `Err` if a named UDF predicate is not
/// registered, so callers get a clear error rather than silently empty results.
pub(crate) fn apply_expr_clauses(
    mut bindings: Vec<Binding>,
    where_clauses: &[WhereClause],
    registry: &FunctionRegistry,
) -> anyhow::Result<Vec<Binding>> {
    // Pre-validate: surface unknown UDF predicate names as errors before filtering rows.
    for clause in where_clauses {
        if let WhereClause::Expr {
            expr: Expr::UnaryOp(UnaryOp::Udf(name), _),
            ..
        } = clause
            && registry.get_predicate(name).is_none()
        {
            anyhow::bail!("unknown predicate: '{}'", name);
        }
    }

    for clause in where_clauses {
        if let WhereClause::Expr { expr, binding: out } = clause {
            bindings = bindings
                .into_iter()
                .filter_map(|mut b| match eval_expr(expr, &b, Some(registry)) {
                    Ok(v) => {
                        if let Some(var) = out {
                            b.insert(var.clone(), v);
                            Some(b)
                        } else if is_truthy(&v) {
                            Some(b)
                        } else {
                            None
                        }
                    }
                    Err(()) => None,
                })
                .collect();
        }
    }
    Ok(bindings)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::query::datalog::parser::parse_datalog_command;
    use crate::query::datalog::types::WhereClause;
    use uuid::Uuid;

    #[test]
    fn test_execute_transact() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        let cmd = parse_datalog_command(
            r#"(transact [[:alice :person/name "Alice"]
                          [:alice :person/age 30]])"#,
        )
        .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::Transacted(tx_id) => {
                assert!(tx_id > 0);
            }
            _ => panic!("Expected Transacted result"),
        }

        // Verify facts were added
        let facts = executor.storage().get_asserted_facts().unwrap();
        assert_eq!(facts.len(), 2);
    }

    #[test]
    fn test_execute_simple_query() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Add some facts
        let alice_id = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        alice_id,
                        ":person/name".to_string(),
                        Value::String("Alice".to_string()),
                    ),
                    (alice_id, ":person/age".to_string(), Value::Integer(30)),
                ],
                None,
            )
            .unwrap();

        // Query for name
        let cmd = parse_datalog_command(r#"(query [:find ?name :where [?e :person/name ?name]])"#)
            .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?name"]);
                assert_eq!(results.len(), 1);
                assert_eq!(results[0][0], Value::String("Alice".to_string()));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_multi_pattern_query() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Add some facts
        let alice_id = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        alice_id,
                        ":person/name".to_string(),
                        Value::String("Alice".to_string()),
                    ),
                    (alice_id, ":person/age".to_string(), Value::Integer(30)),
                ],
                None,
            )
            .unwrap();

        // Query for both name and age
        let cmd = parse_datalog_command(
            r#"(query [:find ?name ?age
                       :where [?e :person/name ?name]
                              [?e :person/age ?age]])"#,
        )
        .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?name", "?age"]);
                assert_eq!(results.len(), 1);
                assert_eq!(results[0][0], Value::String("Alice".to_string()));
                assert_eq!(results[0][1], Value::Integer(30));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_query_no_results() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Query with no matching facts
        let cmd = parse_datalog_command(r#"(query [:find ?name :where [?e :person/name ?name]])"#)
            .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?name"]);
                assert_eq!(results.len(), 0);
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_retract() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Add a fact
        let alice_id = Uuid::new_v4();
        storage
            .transact(
                vec![(alice_id, ":person/age".to_string(), Value::Integer(30))],
                None,
            )
            .unwrap();

        // Verify it exists
        let current_value = storage
            .get_current_value(&alice_id, &":person/age".to_string())
            .unwrap();
        assert_eq!(current_value, Some(Value::Integer(30)));

        // Small delay to ensure different timestamp
        std::thread::sleep(std::time::Duration::from_millis(2));

        // Retract it using UUID-based entity reference
        let cmd = parse_datalog_command(
            format!(r#"(retract [[#uuid "{}" :person/age 30]])"#, alice_id).as_str(),
        )
        .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::Retracted(tx_id) => {
                assert!(tx_id > 0);
            }
            _ => panic!("Expected Retracted result"),
        }

        // Verify it's retracted (current value should be None)
        let current_value = storage
            .get_current_value(&alice_id, &":person/age".to_string())
            .unwrap();
        assert_eq!(current_value, None);
    }

    #[test]
    fn test_transact_with_keyword_entity() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Transact with keyword-based entity (will be converted to deterministic UUID)
        let cmd = parse_datalog_command(
            r#"(transact [[:alice :person/name "Alice"]
                          [:alice :person/age 30]])"#,
        )
        .unwrap();

        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::Transacted(_) => {}
            _ => panic!("Expected Transacted result"),
        }

        // Query to verify both facts share the same entity
        let query_cmd = parse_datalog_command(
            r#"(query [:find ?name ?age
                       :where [?e :person/name ?name]
                              [?e :person/age ?age]])"#,
        )
        .unwrap();

        let result = executor.execute(query_cmd).unwrap();
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1);
                assert_eq!(results[0][0], Value::String("Alice".to_string()));
                assert_eq!(results[0][1], Value::Integer(30));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_register_rule() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Parse and execute a rule command
        let cmd =
            parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap();

        let result = executor.execute(cmd).unwrap();
        assert_eq!(result, QueryResult::Ok);

        // Verify rule was registered
        let registry = executor.rules();
        let rules = registry.read().unwrap().get_rules("reachable");
        assert_eq!(rules.len(), 1);
    }

    #[test]
    fn test_register_multiple_rules_same_predicate() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Register base case
        let cmd1 =
            parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap();
        executor.execute(cmd1).unwrap();

        // Register recursive case
        let cmd2 = parse_datalog_command(
            r#"(rule [(reachable ?x ?y) [?x :connected ?z] (reachable ?z ?y)])"#,
        )
        .unwrap();
        executor.execute(cmd2).unwrap();

        // Verify both rules registered
        let registry = executor.rules();
        let rules = registry.read().unwrap().get_rules("reachable");
        assert_eq!(rules.len(), 2);
    }

    #[test]
    fn test_register_rules_different_predicates() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Register reachable rule
        let cmd1 =
            parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap();
        executor.execute(cmd1).unwrap();

        // Register ancestor rule
        let cmd2 = parse_datalog_command(r#"(rule [(ancestor ?a ?d) [?a :parent ?d]])"#).unwrap();
        executor.execute(cmd2).unwrap();

        // Verify both predicates have rules
        let registry = executor.rules();
        let reg_read = registry.read().unwrap();
        assert!(reg_read.has_rule("reachable"));
        assert!(reg_read.has_rule("ancestor"));
        assert_eq!(reg_read.predicate_count(), 2);
    }

    #[test]
    fn test_query_with_rule_invocation() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Create graph: A->B, A->C
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let c = Uuid::new_v4();

        storage
            .transact(
                vec![
                    (a, ":connected".to_string(), Value::Ref(b)),
                    (a, ":connected".to_string(), Value::Ref(c)),
                ],
                None,
            )
            .unwrap();

        // Register reachable rule (base case only - no recursion yet)
        let rule1 =
            parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap();
        executor.execute(rule1).unwrap();

        // Query using rule invocation: find all nodes reachable from A
        let query_str = format!(
            r#"(query [:find ?to :where (reachable #uuid "{}" ?to)])"#,
            a
        );
        let query_cmd = parse_datalog_command(&query_str).unwrap();

        let result = executor.execute(query_cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?to"]);
                // Should find B and C (direct connections)
                assert_eq!(results.len(), 2);

                // Collect result UUIDs
                let result_uuids: Vec<Uuid> = results
                    .iter()
                    .map(|row| match &row[0] {
                        Value::Ref(uuid) => *uuid,
                        _ => panic!("Expected Ref value"),
                    })
                    .collect();

                assert!(result_uuids.contains(&b));
                assert!(result_uuids.contains(&c));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_query_mixed_pattern_and_rule() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Create graph with names: A->B, A->C, and give B a name
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let c = Uuid::new_v4();

        storage
            .transact(
                vec![
                    (a, ":connected".to_string(), Value::Ref(b)),
                    (a, ":connected".to_string(), Value::Ref(c)),
                    (
                        b,
                        ":person/name".to_string(),
                        Value::String("Bob".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();

        // Register reachable rule (base case only - no recursion yet)
        executor
            .execute(
                parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap(),
            )
            .unwrap();

        // Query: find names of nodes reachable from A
        let query_str = format!(
            r#"(query [:find ?name :where (reachable #uuid "{}" ?to) [?to :person/name ?name]])"#,
            a
        );
        let query_cmd = parse_datalog_command(&query_str).unwrap();

        let result = executor.execute(query_cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?name"]);
                assert_eq!(results.len(), 1);
                assert_eq!(results[0][0], Value::String("Bob".to_string()));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_query_with_recursive_transitive_closure() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Create graph: A->B->C
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let c = Uuid::new_v4();

        storage
            .transact(
                vec![
                    (a, ":connected".to_string(), Value::Ref(b)),
                    (b, ":connected".to_string(), Value::Ref(c)),
                ],
                None,
            )
            .unwrap();

        // Register reachable rules (base + recursive)
        executor
            .execute(
                parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap(),
            )
            .unwrap();

        executor
            .execute(
                parse_datalog_command(
                    r#"(rule [(reachable ?x ?y) [?x :connected ?z] (reachable ?z ?y)])"#,
                )
                .unwrap(),
            )
            .unwrap();

        // Query: find all nodes reachable from A
        let query_str = format!(
            r#"(query [:find ?to :where (reachable #uuid "{}" ?to)])"#,
            a
        );
        let query_cmd = parse_datalog_command(&query_str).unwrap();

        let result = executor.execute(query_cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?to"]);
                // Should find B and C via transitive closure
                assert_eq!(results.len(), 2);

                // Collect result UUIDs
                let result_uuids: Vec<Uuid> = results
                    .iter()
                    .map(|row| match &row[0] {
                        Value::Ref(uuid) => *uuid,
                        _ => panic!("Expected Ref value"),
                    })
                    .collect();

                assert!(result_uuids.contains(&b));
                assert!(result_uuids.contains(&c));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_default_query_filters_to_currently_valid() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());
        let alice = Uuid::new_v4();

        // Fact valid forever (default) - tx_count=1
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":person/name".to_string()),
                    EdnValue::String("Alice".to_string()),
                )],
                valid_from: None,
                valid_to: None,
            }))
            .unwrap();

        // Fact with valid_to in the past (expired) - tx_count=2
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":employment/status".to_string()),
                    EdnValue::Keyword(":active".to_string()),
                )],
                valid_from: Some(1000_i64),
                valid_to: Some(2000_i64), // expired long ago
            }))
            .unwrap();

        // Default query (no :valid-at) should only return the forever-valid fact
        let result = executor
            .execute(DatalogCommand::Query(DatalogQuery::new(
                vec![FindSpec::Variable("?attr".to_string())],
                vec![WhereClause::Pattern(Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Symbol("?attr".to_string()),
                    EdnValue::Symbol("?v".to_string()),
                ))],
            )))
            .unwrap();

        let rows = match result {
            QueryResult::QueryResults { results, .. } => results,
            _ => panic!("expected query results"),
        };
        assert_eq!(rows.len(), 1); // only the name fact
    }

    #[test]
    fn test_as_of_counter_shows_past_state() {
        use crate::query::datalog::types::AsOf;
        use crate::query::datalog::types::ValidAt;

        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let alice = Uuid::new_v4();

        // tx_count=1: assert name
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":person/name".to_string()),
                    EdnValue::String("Alice".to_string()),
                )],
                valid_from: None,
                valid_to: None,
            }))
            .unwrap();

        // tx_count=2: assert age
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":person/age".to_string()),
                    EdnValue::Integer(30),
                )],
                valid_from: None,
                valid_to: None,
            }))
            .unwrap();

        // :as-of 1 → only name fact visible (age was added at tx_count=2)
        let result = executor
            .execute(DatalogCommand::Query(DatalogQuery {
                find: vec![FindSpec::Variable("?attr".to_string())],
                where_clauses: vec![WhereClause::Pattern(Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Symbol("?attr".to_string()),
                    EdnValue::Symbol("?v".to_string()),
                ))],
                as_of: Some(AsOf::Counter(1)),
                valid_at: Some(ValidAt::AnyValidTime),
                with_vars: Vec::new(),
            }))
            .unwrap();

        let rows = match result {
            QueryResult::QueryResults { results, .. } => results,
            _ => panic!("expected query results"),
        };
        assert_eq!(rows.len(), 1);
    }

    #[test]
    fn test_valid_at_any_valid_time_shows_all() {
        use crate::query::datalog::types::ValidAt;

        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let alice = Uuid::new_v4();

        // Fact valid forever (default)
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":person/name".to_string()),
                    EdnValue::String("Alice".to_string()),
                )],
                valid_from: None,
                valid_to: None,
            }))
            .unwrap();

        // Fact with valid_to already in the past
        executor
            .execute(DatalogCommand::Transact(Transaction {
                facts: vec![Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Keyword(":employment/status".to_string()),
                    EdnValue::Keyword(":active".to_string()),
                )],
                valid_from: Some(1000_i64),
                valid_to: Some(2000_i64), // expired
            }))
            .unwrap();

        // :valid-at :any-valid-time → both facts returned
        let result = executor
            .execute(DatalogCommand::Query(DatalogQuery {
                find: vec![FindSpec::Variable("?attr".to_string())],
                where_clauses: vec![WhereClause::Pattern(Pattern::new(
                    EdnValue::Uuid(alice),
                    EdnValue::Symbol("?attr".to_string()),
                    EdnValue::Symbol("?v".to_string()),
                ))],
                as_of: None,
                valid_at: Some(ValidAt::AnyValidTime),
                with_vars: Vec::new(),
            }))
            .unwrap();

        let rows = match result {
            QueryResult::QueryResults { results, .. } => results,
            _ => panic!("expected query results"),
        };
        assert_eq!(rows.len(), 2);
    }

    #[test]
    fn test_query_recursive_with_mixed_patterns() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());

        // Create graph: A->B->C, give C a name
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let c = Uuid::new_v4();

        storage
            .transact(
                vec![
                    (a, ":connected".to_string(), Value::Ref(b)),
                    (b, ":connected".to_string(), Value::Ref(c)),
                    (
                        c,
                        ":person/name".to_string(),
                        Value::String("Charlie".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();

        // Register recursive reachable rules
        executor
            .execute(
                parse_datalog_command(r#"(rule [(reachable ?x ?y) [?x :connected ?y]])"#).unwrap(),
            )
            .unwrap();

        executor
            .execute(
                parse_datalog_command(
                    r#"(rule [(reachable ?x ?y) [?x :connected ?z] (reachable ?z ?y)])"#,
                )
                .unwrap(),
            )
            .unwrap();

        // Query: find names of nodes transitively reachable from A
        let query_str = format!(
            r#"(query [:find ?name :where (reachable #uuid "{}" ?to) [?to :person/name ?name]])"#,
            a
        );
        let query_cmd = parse_datalog_command(&query_str).unwrap();

        let result = executor.execute(query_cmd).unwrap();
        match result {
            QueryResult::QueryResults { vars, results } => {
                assert_eq!(vars, vec!["?name"]);
                assert_eq!(results.len(), 1);
                assert_eq!(results[0][0], Value::String("Charlie".to_string()));
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_query_not_as_pure_filter() {
        // Query: [:find ?e :where [?e :applied true] (not [?e :rejected true])]
        // No rule invocations — pure not-filter path in execute_query.
        use crate::query::datalog::types::WhereClause;
        let storage = FactStorage::new();
        let alice = uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
        let bob = uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap();
        // alice: applied + rejected
        storage
            .transact(
                vec![
                    (alice, ":applied".to_string(), Value::Boolean(true)),
                    (alice, ":rejected".to_string(), Value::Boolean(true)),
                ],
                None,
            )
            .unwrap();
        // bob: applied only
        storage
            .transact(
                vec![(bob, ":applied".to_string(), Value::Boolean(true))],
                None,
            )
            .unwrap();

        let query = DatalogQuery::new(
            vec![FindSpec::Variable("?e".to_string())],
            vec![
                WhereClause::Pattern(Pattern::new(
                    EdnValue::Symbol("?e".to_string()),
                    EdnValue::Keyword(":applied".to_string()),
                    EdnValue::Boolean(true),
                )),
                WhereClause::Not(vec![WhereClause::Pattern(Pattern::new(
                    EdnValue::Symbol("?e".to_string()),
                    EdnValue::Keyword(":rejected".to_string()),
                    EdnValue::Boolean(true),
                ))]),
            ],
        );

        let executor = DatalogExecutor::new(storage);
        let result = executor
            .execute(crate::query::datalog::types::DatalogCommand::Query(query))
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "only bob should pass (alice is rejected)");
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_query_with_rules_not_in_query_body() {
        // Query: [:find ?x :where (reachable ?_a ?x) (not [?x :blocked true])]
        // rule invocation + pattern-not in same query body
        use crate::query::datalog::types::{Pattern, WhereClause};
        let storage = FactStorage::new();
        let a = uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
        let b = uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap();
        let c = uuid::Uuid::parse_str("00000000-0000-0000-0000-000000000003").unwrap();
        storage
            .transact(
                vec![
                    (a, ":connected".to_string(), Value::Ref(b)),
                    (a, ":connected".to_string(), Value::Ref(c)),
                    (c, ":blocked".to_string(), Value::Boolean(true)),
                ],
                None,
            )
            .unwrap();

        let rules = Arc::new(RwLock::new(RuleRegistry::new()));
        // reachable(?from ?to) :- [?from :connected ?to]
        {
            use crate::query::datalog::types::{Rule, WhereClause as WC};
            let rule = Rule {
                head: vec![
                    EdnValue::Symbol("reachable".to_string()),
                    EdnValue::Symbol("?from".to_string()),
                    EdnValue::Symbol("?to".to_string()),
                ],
                body: vec![WC::Pattern(Pattern::new(
                    EdnValue::Symbol("?from".to_string()),
                    EdnValue::Keyword(":connected".to_string()),
                    EdnValue::Symbol("?to".to_string()),
                ))],
            };
            rules
                .write()
                .unwrap()
                .register_rule("reachable".to_string(), rule)
                .unwrap();
        }

        let query = DatalogQuery::new(
            vec![FindSpec::Variable("?x".to_string())],
            vec![
                WhereClause::RuleInvocation {
                    predicate: "reachable".to_string(),
                    args: vec![
                        EdnValue::Symbol("?_a".to_string()),
                        EdnValue::Symbol("?x".to_string()),
                    ],
                },
                WhereClause::Not(vec![WhereClause::Pattern(Pattern::new(
                    EdnValue::Symbol("?x".to_string()),
                    EdnValue::Keyword(":blocked".to_string()),
                    EdnValue::Boolean(true),
                ))]),
            ],
        );

        let executor = DatalogExecutor::new_with_rules(storage, rules);
        let result = executor
            .execute(crate::query::datalog::types::DatalogCommand::Query(query))
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    1,
                    "c should be excluded (blocked), only b passes"
                );
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_query_not_join_basic() {
        // Query: find entities that have :submitted but NO blocked dependency
        // alice: submitted, has-dep dep1, dep1:blocked=true  -> excluded
        // bob:   submitted, no deps                          -> included
        let storage = FactStorage::new();
        let alice = Uuid::new_v4();
        let bob = Uuid::new_v4();
        let dep1 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (alice, ":submitted".to_string(), Value::Boolean(true)),
                    (alice, ":has-dep".to_string(), Value::Ref(dep1)),
                    (dep1, ":blocked".to_string(), Value::Boolean(true)),
                    (bob, ":submitted".to_string(), Value::Boolean(true)),
                ],
                None,
            )
            .unwrap();

        let query = DatalogQuery::new(
            vec![FindSpec::Variable("?x".to_string())],
            vec![
                WhereClause::Pattern(Pattern::new(
                    EdnValue::Symbol("?x".to_string()),
                    EdnValue::Keyword(":submitted".to_string()),
                    EdnValue::Boolean(true),
                )),
                WhereClause::NotJoin {
                    join_vars: vec!["?x".to_string()],
                    clauses: vec![
                        WhereClause::Pattern(Pattern::new(
                            EdnValue::Symbol("?x".to_string()),
                            EdnValue::Keyword(":has-dep".to_string()),
                            EdnValue::Symbol("?d".to_string()),
                        )),
                        WhereClause::Pattern(Pattern::new(
                            EdnValue::Symbol("?d".to_string()),
                            EdnValue::Keyword(":blocked".to_string()),
                            EdnValue::Boolean(true),
                        )),
                    ],
                },
            ],
        );

        let executor = DatalogExecutor::new(storage);
        let result = executor
            .execute(crate::query::datalog::types::DatalogCommand::Query(query))
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "only bob should be returned");
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_execute_query_with_rules_not_join_in_query_body() {
        // Rule: (reachable ?x ?y) :- [?x :edge ?y]
        // Query: find ?y reachable from root that do NOT have a blocked dep
        let storage = FactStorage::new();
        let root = Uuid::new_v4();
        let a = Uuid::new_v4();
        let b = Uuid::new_v4();
        let dep1 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (root, ":edge".to_string(), Value::Ref(a)),
                    (root, ":edge".to_string(), Value::Ref(b)),
                    (a, ":has-dep".to_string(), Value::Ref(dep1)),
                    (dep1, ":blocked".to_string(), Value::Boolean(true)),
                ],
                None,
            )
            .unwrap();

        let rules = Arc::new(RwLock::new(RuleRegistry::new()));
        {
            use crate::query::datalog::types::{Rule, WhereClause as WC};
            let rule = Rule {
                head: vec![
                    EdnValue::Symbol("reachable".to_string()),
                    EdnValue::Symbol("?x".to_string()),
                    EdnValue::Symbol("?y".to_string()),
                ],
                body: vec![WC::Pattern(Pattern::new(
                    EdnValue::Symbol("?x".to_string()),
                    EdnValue::Keyword(":edge".to_string()),
                    EdnValue::Symbol("?y".to_string()),
                ))],
            };
            rules
                .write()
                .unwrap()
                .register_rule("reachable".to_string(), rule)
                .unwrap();
        }

        let query = DatalogQuery::new(
            vec![FindSpec::Variable("?y".to_string())],
            vec![
                WhereClause::RuleInvocation {
                    predicate: "reachable".to_string(),
                    args: vec![EdnValue::Uuid(root), EdnValue::Symbol("?y".to_string())],
                },
                WhereClause::NotJoin {
                    join_vars: vec!["?y".to_string()],
                    clauses: vec![
                        WhereClause::Pattern(Pattern::new(
                            EdnValue::Symbol("?y".to_string()),
                            EdnValue::Keyword(":has-dep".to_string()),
                            EdnValue::Symbol("?d".to_string()),
                        )),
                        WhereClause::Pattern(Pattern::new(
                            EdnValue::Symbol("?d".to_string()),
                            EdnValue::Keyword(":blocked".to_string()),
                            EdnValue::Boolean(true),
                        )),
                    ],
                },
            ],
        );

        let executor = DatalogExecutor::new_with_rules(storage, rules);
        let result = executor
            .execute(crate::query::datalog::types::DatalogCommand::Query(query))
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                // a is excluded (has a blocked dep); b passes
                assert_eq!(results.len(), 1, "only b should pass");
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    #[test]
    fn test_optimizer_does_not_change_query_results() {
        // A multi-pattern query that the optimizer would reorder.
        // Results must be identical regardless of execution order.
        let storage = FactStorage::new();
        let alice = uuid::Uuid::new_v4();
        let bob = uuid::Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        alice,
                        ":name".to_string(),
                        Value::String("Alice".to_string()),
                    ),
                    (alice, ":friend".to_string(), Value::Ref(bob)),
                    (bob, ":name".to_string(), Value::String("Bob".to_string())),
                ],
                None,
            )
            .unwrap();

        let executor = DatalogExecutor::new(storage);
        // Simple query: find all names (no join reordering needed)
        let result = executor
            .execute(
                parse_datalog_command("(query [:find ?name :where [?e :name ?name]])").unwrap(),
            )
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 2, "Alice and Bob both have names");
            }
            _ => panic!("Expected QueryResults"),
        }
    }

    // Helper: build a binding map from key-value pairs
    fn binding(pairs: &[(&str, Value)]) -> std::collections::HashMap<String, Value> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.clone()))
            .collect()
    }

    #[test]
    fn test_apply_aggregation_count_basic() {
        let bindings = vec![
            binding(&[("?e", Value::Integer(1))]),
            binding(&[("?e", Value::Integer(2))]),
            binding(&[("?e", Value::Integer(3))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "count".to_string(),
            var: "?e".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0][0], Value::Integer(3));
    }

    #[test]
    fn test_apply_aggregation_count_with_grouping() {
        let bindings = vec![
            binding(&[
                ("?dept", Value::String("eng".to_string())),
                ("?e", Value::Integer(1)),
            ]),
            binding(&[
                ("?dept", Value::String("eng".to_string())),
                ("?e", Value::Integer(2)),
            ]),
            binding(&[
                ("?dept", Value::String("hr".to_string())),
                ("?e", Value::Integer(3)),
            ]),
        ];
        let find_specs = vec![
            FindSpec::Variable("?dept".to_string()),
            FindSpec::Aggregate {
                func: "count".to_string(),
                var: "?e".to_string(),
            },
        ];
        let mut results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        results.sort_by_key(|r| match &r[0] {
            Value::String(s) => s.clone(),
            _ => String::new(),
        });
        assert_eq!(results.len(), 2);
        assert_eq!(
            results[0],
            vec![Value::String("eng".to_string()), Value::Integer(2)]
        );
        assert_eq!(
            results[1],
            vec![Value::String("hr".to_string()), Value::Integer(1)]
        );
    }

    #[test]
    fn test_apply_aggregation_count_distinct() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(1))]),
            binding(&[("?v", Value::Integer(1))]), // duplicate
            binding(&[("?v", Value::Integer(2))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "count-distinct".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(2));
    }

    #[test]
    fn test_apply_aggregation_count_empty_no_grouping_vars() {
        // count with no grouping vars + zero bindings → [[0]]
        let find_specs = vec![FindSpec::Aggregate {
            func: "count".to_string(),
            var: "?e".to_string(),
        }];
        let results =
            apply_post_processing(vec![], &find_specs, &[], &FunctionRegistry::with_builtins())
                .unwrap();
        assert_eq!(results.len(), 1, "should return one row with 0");
        assert_eq!(results[0][0], Value::Integer(0));
    }

    #[test]
    fn test_apply_aggregation_count_empty_with_grouping_var() {
        // count with grouping var + zero bindings → empty result
        let find_specs = vec![
            FindSpec::Variable("?dept".to_string()),
            FindSpec::Aggregate {
                func: "count".to_string(),
                var: "?e".to_string(),
            },
        ];
        let results =
            apply_post_processing(vec![], &find_specs, &[], &FunctionRegistry::with_builtins())
                .unwrap();
        assert_eq!(results.len(), 0, "should return empty set");
    }

    #[test]
    fn test_apply_aggregation_sum_integers() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(10))]),
            binding(&[("?v", Value::Integer(20))]),
            binding(&[("?v", Value::Integer(30))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(60));
    }

    #[test]
    fn test_apply_aggregation_sum_widens_to_float() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(10))]),
            binding(&[("?v", Value::Float(0.5))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Float(10.5));
    }

    #[test]
    fn test_apply_aggregation_sum_distinct_deduplicates() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(5))]),
            binding(&[("?v", Value::Integer(5))]), // duplicate
            binding(&[("?v", Value::Integer(10))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum-distinct".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(15)); // 5 + 10, not 5 + 5 + 10
    }

    #[test]
    fn test_apply_aggregation_sum_type_error() {
        let bindings = vec![binding(&[("?v", Value::String("bad".to_string()))])];
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum".to_string(),
            var: "?v".to_string(),
        }];
        let result = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        );
        assert!(result.is_err(), "sum of string should fail");
    }

    #[test]
    fn test_apply_aggregation_min_integers() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(30))]),
            binding(&[("?v", Value::Integer(10))]),
            binding(&[("?v", Value::Integer(20))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "min".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(10));
    }

    #[test]
    fn test_apply_aggregation_max_strings() {
        let bindings = vec![
            binding(&[("?v", Value::String("apple".to_string()))]),
            binding(&[("?v", Value::String("zebra".to_string()))]),
            binding(&[("?v", Value::String("mango".to_string()))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "max".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::String("zebra".to_string()));
    }

    #[test]
    fn test_apply_aggregation_min_type_error_boolean() {
        let bindings = vec![binding(&[("?v", Value::Boolean(true))])];
        let find_specs = vec![FindSpec::Aggregate {
            func: "min".to_string(),
            var: "?v".to_string(),
        }];
        let result = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        );
        assert!(result.is_err(), "min of boolean should fail");
    }

    #[test]
    fn test_apply_aggregation_min_mixed_int_float_error() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(1))]),
            binding(&[("?v", Value::Float(2.0))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "min".to_string(),
            var: "?v".to_string(),
        }];
        let result = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        );
        assert!(result.is_err(), "min of mixed Integer/Float should fail");
    }

    #[test]
    fn test_apply_aggregation_skips_nulls_in_sum() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(10))]),
            binding(&[("?v", Value::Null)]),
            binding(&[("?v", Value::Integer(20))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(30));
    }

    #[test]
    fn test_apply_aggregation_skips_nulls_in_count() {
        let bindings = vec![
            binding(&[("?v", Value::Integer(1))]),
            binding(&[("?v", Value::Null)]),
            binding(&[("?v", Value::Integer(2))]),
        ];
        let find_specs = vec![FindSpec::Aggregate {
            func: "count".to_string(),
            var: "?v".to_string(),
        }];
        let results = apply_post_processing(
            bindings,
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results[0][0], Value::Integer(2)); // null not counted
    }

    #[test]
    fn test_apply_aggregation_sum_empty_bindings() {
        let find_specs = vec![FindSpec::Aggregate {
            func: "sum".to_string(),
            var: "?v".to_string(),
        }];
        let results =
            apply_post_processing(vec![], &find_specs, &[], &FunctionRegistry::with_builtins())
                .unwrap();
        assert_eq!(results.len(), 0, "sum on empty should return empty set");
    }

    #[test]
    fn test_apply_aggregation_with_var_grouping() {
        // :with ?e adds ?e to the group key. Two entities with same dept but different ?e
        // form separate groups.
        let bindings = vec![
            binding(&[
                ("?dept", Value::String("eng".to_string())),
                ("?salary", Value::Integer(50)),
                ("?e", Value::Integer(1)),
            ]),
            binding(&[
                ("?dept", Value::String("eng".to_string())),
                ("?salary", Value::Integer(50)),
                ("?e", Value::Integer(2)),
            ]),
        ];
        let find_specs = vec![
            FindSpec::Variable("?dept".to_string()),
            FindSpec::Aggregate {
                func: "sum".to_string(),
                var: "?salary".to_string(),
            },
        ];
        // Without :with: group key = ("eng",). Both bindings in one group → sum = 100.
        let results_no_with = apply_post_processing(
            bindings.clone(),
            &find_specs,
            &[],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results_no_with.len(), 1);
        assert_eq!(results_no_with[0][1], Value::Integer(100));
        // With :with ?e: group key = ("eng", e). Two separate groups → two rows, each sum = 50.
        let results_with = apply_post_processing(
            bindings,
            &find_specs,
            &["?e".to_string()],
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(results_with.len(), 2);
        assert_eq!(results_with[0][1], Value::Integer(50));
    }

    #[test]
    fn test_filter_facts_for_query_returns_net_asserted_slice() {
        // Setup: one fact asserted then retracted, one fact left standing.
        // After filter_facts_for_query, only the standing fact should appear.
        // The return type (Arc<[Fact]>) exposes .len() and index access [0].
        use uuid::Uuid;
        let storage = FactStorage::new();
        let alice = Uuid::new_v4();

        // tx 1: assert name
        storage
            .transact(
                vec![(
                    alice,
                    ":person/name".to_string(),
                    Value::String("Alice".to_string()),
                )],
                None,
            )
            .unwrap();

        // tx 2: retract name — net state for name is now gone
        storage
            .retract(vec![(
                alice,
                ":person/name".to_string(),
                Value::String("Alice".to_string()),
            )])
            .unwrap();

        // tx 3: assert age — this is the only net-asserted fact
        storage
            .transact(
                vec![(alice, ":person/age".to_string(), Value::Integer(30))],
                None,
            )
            .unwrap();

        let executor = DatalogExecutor::new(storage);
        let query = DatalogQuery {
            find: vec![],
            where_clauses: vec![],
            as_of: None,
            valid_at: Some(ValidAt::AnyValidTime),
            with_vars: vec![],
        };

        let facts = executor.filter_facts_for_query(&query).unwrap();
        assert_eq!(facts.len(), 1, "expected exactly 1 net-asserted fact");
        assert_eq!(facts[0].attribute, ":person/age");
    }

    #[test]
    fn test_filter_facts_for_query_valid_time_filter() {
        // Setup: one fact with a narrow valid-time window (1000..2000), one open-ended.
        // Query with valid_at inside the window → both facts visible.
        // Query with valid_at outside the window → only the open-ended fact visible.
        // filter_facts_for_query now returns Result<Arc<[Fact]>> (changed in Task 5).
        use crate::graph::types::TransactOptions;
        use uuid::Uuid;
        let storage = FactStorage::new();
        let alice = Uuid::new_v4();

        // Fact valid only during [1000, 2000)
        storage
            .transact(
                vec![(
                    alice,
                    ":employment/status".to_string(),
                    Value::String("active".to_string()),
                )],
                Some(TransactOptions::new(Some(1000_i64), Some(2000_i64))),
            )
            .unwrap();

        // Fact valid forever (open-ended): explicit valid_from=0 so it is visible at t=1500 and t=3000.
        // Passing None would set valid_from=tx_id_now() (current epoch ms ≈ 1.7T), which is
        // far beyond the test's query timestamps.
        storage
            .transact(
                vec![(
                    alice,
                    ":person/name".to_string(),
                    Value::String("Alice".to_string()),
                )],
                Some(TransactOptions::new(Some(0_i64), None)),
            )
            .unwrap();

        let executor = DatalogExecutor::new(storage);

        // Query inside the window: both facts should be visible
        let query_inside = DatalogQuery {
            find: vec![],
            where_clauses: vec![],
            as_of: None,
            valid_at: Some(ValidAt::Timestamp(1500_i64)),
            with_vars: vec![],
        };
        let facts_inside = executor.filter_facts_for_query(&query_inside).unwrap();
        assert_eq!(facts_inside.len(), 2, "both facts visible at t=1500");

        // Query outside the window: only the open-ended name fact should be visible
        let query_outside = DatalogQuery {
            find: vec![],
            where_clauses: vec![],
            as_of: None,
            valid_at: Some(ValidAt::Timestamp(3000_i64)),
            with_vars: vec![],
        };
        let facts_outside = executor.filter_facts_for_query(&query_outside).unwrap();
        assert_eq!(
            facts_outside.len(),
            1,
            "only open-ended fact visible at t=3000"
        );
        assert_eq!(facts_outside[0].attribute, ":person/name");
    }
}

#[cfg(test)]
mod expr_eval_tests {
    use super::*;
    use crate::graph::types::Value;
    use crate::query::datalog::parser::parse_datalog_command;
    use crate::query::datalog::types::{BinOp, Expr, UnaryOp, WhereClause};
    use std::collections::HashMap;
    use std::sync::Arc;
    use uuid::Uuid;

    fn b(pairs: &[(&str, Value)]) -> HashMap<String, Value> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.clone()))
            .collect()
    }

    #[test]
    fn test_eval_lit() {
        let e = Expr::Lit(Value::Integer(42));
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Integer(42)));
    }

    #[test]
    fn test_eval_var_bound() {
        let e = Expr::Var("?x".to_string());
        let binding = b(&[("?x", Value::Integer(10))]);
        assert_eq!(eval_expr(&e, &binding, None), Ok(Value::Integer(10)));
    }

    #[test]
    fn test_eval_var_unbound_is_err() {
        let e = Expr::Var("?x".to_string());
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Err(()));
    }

    #[test]
    fn test_eval_lt_true() {
        let e = Expr::BinOp(
            BinOp::Lt,
            Box::new(Expr::Var("?v".to_string())),
            Box::new(Expr::Lit(Value::Integer(100))),
        );
        let binding = b(&[("?v", Value::Integer(50))]);
        assert_eq!(eval_expr(&e, &binding, None), Ok(Value::Boolean(true)));
    }

    #[test]
    fn test_eval_lt_false() {
        let e = Expr::BinOp(
            BinOp::Lt,
            Box::new(Expr::Var("?v".to_string())),
            Box::new(Expr::Lit(Value::Integer(100))),
        );
        let binding = b(&[("?v", Value::Integer(150))]);
        assert_eq!(eval_expr(&e, &binding, None), Ok(Value::Boolean(false)));
    }

    #[test]
    fn test_eval_add_integers() {
        let e = Expr::BinOp(
            BinOp::Add,
            Box::new(Expr::Var("?a".to_string())),
            Box::new(Expr::Var("?b".to_string())),
        );
        let binding = b(&[("?a", Value::Integer(3)), ("?b", Value::Integer(4))]);
        assert_eq!(eval_expr(&e, &binding, None), Ok(Value::Integer(7)));
    }

    #[test]
    fn test_eval_add_int_float_promotes() {
        let e = Expr::BinOp(
            BinOp::Add,
            Box::new(Expr::Lit(Value::Integer(1))),
            Box::new(Expr::Lit(Value::Float(1.5))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Float(2.5)));
    }

    #[test]
    fn test_eval_div_integer_truncates() {
        let e = Expr::BinOp(
            BinOp::Div,
            Box::new(Expr::Lit(Value::Integer(5))),
            Box::new(Expr::Lit(Value::Integer(2))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Integer(2)));
    }

    #[test]
    fn test_eval_div_by_zero_is_err() {
        let e = Expr::BinOp(
            BinOp::Div,
            Box::new(Expr::Lit(Value::Integer(5))),
            Box::new(Expr::Lit(Value::Integer(0))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Err(()));
    }

    #[test]
    fn test_eval_eq_strings() {
        let e = Expr::BinOp(
            BinOp::Eq,
            Box::new(Expr::Lit(Value::String("Alice".to_string()))),
            Box::new(Expr::Lit(Value::String("Alice".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_eval_eq_int_float_false() {
        // Different Value variants → structural inequality
        let e = Expr::BinOp(
            BinOp::Eq,
            Box::new(Expr::Lit(Value::Integer(1))),
            Box::new(Expr::Lit(Value::Float(1.0))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(false))
        );
    }

    #[test]
    fn test_eval_type_mismatch_comparison_is_err() {
        let e = Expr::BinOp(
            BinOp::Lt,
            Box::new(Expr::Lit(Value::String("hello".to_string()))),
            Box::new(Expr::Lit(Value::Integer(100))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Err(()));
    }

    #[test]
    fn test_eval_string_q_true() {
        let e = Expr::UnaryOp(
            UnaryOp::StringQ,
            Box::new(Expr::Lit(Value::String("hi".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_eval_string_q_false() {
        let e = Expr::UnaryOp(UnaryOp::StringQ, Box::new(Expr::Lit(Value::Integer(1))));
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(false))
        );
    }

    #[test]
    fn test_eval_starts_with_true() {
        let e = Expr::BinOp(
            BinOp::StartsWith,
            Box::new(Expr::Lit(Value::String("foobar".to_string()))),
            Box::new(Expr::Lit(Value::String("foo".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_eval_ends_with_true() {
        let e = Expr::BinOp(
            BinOp::EndsWith,
            Box::new(Expr::Lit(Value::String("foobar".to_string()))),
            Box::new(Expr::Lit(Value::String("bar".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_eval_contains_true() {
        let e = Expr::BinOp(
            BinOp::Contains,
            Box::new(Expr::Lit(Value::String("engineer at co".to_string()))),
            Box::new(Expr::Lit(Value::String("engineer".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_eval_matches_true() {
        let re = regex_lite::Regex::new("^[^@]+@[^@]+$").unwrap();
        let e = Expr::BinOp(
            BinOp::Matches {
                regex: re,
                pattern: "^[^@]+@[^@]+$".to_string(),
            },
            Box::new(Expr::Lit(Value::String("test@example.com".to_string()))),
            Box::new(Expr::Lit(Value::String("^[^@]+@[^@]+$".to_string()))),
        );
        assert_eq!(
            eval_expr(&e, &HashMap::new(), None),
            Ok(Value::Boolean(true))
        );
    }

    #[test]
    fn test_is_truthy() {
        assert!(is_truthy(&Value::Boolean(true)));
        assert!(!is_truthy(&Value::Boolean(false)));
        assert!(is_truthy(&Value::Integer(1)));
        assert!(!is_truthy(&Value::Integer(0)));
        assert!(is_truthy(&Value::Float(0.1)));
        assert!(!is_truthy(&Value::Float(0.0)));
        assert!(!is_truthy(&Value::Null));
        assert!(!is_truthy(&Value::String("hi".to_string())));
    }

    #[test]
    fn test_apply_expr_filter_keeps_truthy() {
        // [(< ?v 100)] — keeps row where ?v < 100
        use crate::query::datalog::types::WhereClause;
        let expr = Expr::BinOp(
            BinOp::Lt,
            Box::new(Expr::Var("?v".to_string())),
            Box::new(Expr::Lit(Value::Integer(100))),
        );
        let clauses = vec![WhereClause::Expr {
            expr,
            binding: None,
        }];
        let bindings = vec![
            b(&[("?v", Value::Integer(50))]),
            b(&[("?v", Value::Integer(150))]),
        ];
        let result =
            apply_expr_clauses(bindings, &clauses, &FunctionRegistry::with_builtins()).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].get("?v"), Some(&Value::Integer(50)));
    }

    #[test]
    fn test_apply_expr_binding_extends_row() {
        // [(+ ?a ?b) ?sum] — binds ?sum
        use crate::query::datalog::types::WhereClause;
        let expr = Expr::BinOp(
            BinOp::Add,
            Box::new(Expr::Var("?a".to_string())),
            Box::new(Expr::Var("?b".to_string())),
        );
        let clauses = vec![WhereClause::Expr {
            expr,
            binding: Some("?sum".to_string()),
        }];
        let bindings = vec![b(&[("?a", Value::Integer(3)), ("?b", Value::Integer(4))])];
        let result =
            apply_expr_clauses(bindings, &clauses, &FunctionRegistry::with_builtins()).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].get("?sum"), Some(&Value::Integer(7)));
    }

    #[test]
    fn test_apply_expr_type_mismatch_drops_row() {
        // [(< ?v 100)] where ?v = "hello" — type mismatch silently drops row
        use crate::query::datalog::types::WhereClause;
        let expr = Expr::BinOp(
            BinOp::Lt,
            Box::new(Expr::Var("?v".to_string())),
            Box::new(Expr::Lit(Value::Integer(100))),
        );
        let clauses = vec![WhereClause::Expr {
            expr,
            binding: None,
        }];
        let bindings = vec![b(&[("?v", Value::String("hello".to_string()))])];
        let result =
            apply_expr_clauses(bindings, &clauses, &FunctionRegistry::with_builtins()).unwrap();
        assert_eq!(result.len(), 0);
    }

    #[test]
    fn test_execute_expr_filter_lt() {
        use crate::graph::storage::FactStorage;
        use crate::query::datalog::rules::RuleRegistry;
        use std::sync::{Arc, RwLock};

        let storage = FactStorage::new();
        let rules = Arc::new(RwLock::new(RuleRegistry::new()));
        let executor = DatalogExecutor::new_with_rules(storage.clone(), rules);

        // Transact two items with different prices
        executor
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(transact [[:item1 :item/price 50] [:item2 :item/price 150]])",
                )
                .unwrap(),
            )
            .unwrap();

        // Query: find items where price < 100
        let result = executor.execute(
            crate::query::datalog::parser::parse_datalog_command(
                "(query [:find ?e :where [?e :item/price ?p] [(< ?p 100)]])",
            )
            .unwrap(),
        );

        assert!(result.is_ok(), "expr filter query failed");
        match result.unwrap() {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "expected exactly one result");
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_apply_or_clauses_union_from_two_branches() {
        // e1 has :color :red, e2 has :color :blue.
        // or-only where clause: (or [?e :color :red] [?e :color :blue])
        // Without apply_or_clauses, get_patterns() returns [] → match_patterns returns
        // [{}] (one empty binding) → extract_variables finds no ?e binding → 0 results.
        // With apply_or_clauses, both entities are returned → 2 results.
        use uuid::Uuid;
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (e1, ":color".to_string(), Value::Keyword(":red".to_string())),
                    (
                        e2,
                        ":color".to_string(),
                        Value::Keyword(":blue".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();

        let executor = DatalogExecutor::new(storage.clone());
        let cmd = crate::query::datalog::parser::parse_datalog_command(
            r#"(query [:find ?e
                       :where (or [?e :color :red] [?e :color :blue])])"#,
        )
        .unwrap();
        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 2, "both entities should match via or");
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_apply_or_clauses_deduplication() {
        // e1 has :color :red AND :shape :circle.
        // or clause: (or [?e :color :red] [?e :shape :circle])
        // Without apply_or_clauses: or is skipped → 0 results (no non-or patterns).
        // With apply_or_clauses: e1 is returned by both branches → deduplicated to 1 result.
        use uuid::Uuid;
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (e1, ":color".to_string(), Value::Keyword(":red".to_string())),
                    (
                        e1,
                        ":shape".to_string(),
                        Value::Keyword(":circle".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();

        let executor = DatalogExecutor::new(storage.clone());
        let cmd = crate::query::datalog::parser::parse_datalog_command(
            r#"(query [:find ?e
                       :where (or [?e :color :red] [?e :shape :circle])])"#,
        )
        .unwrap();
        let result = executor.execute(cmd).unwrap();
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    1,
                    "one entity matched by both branches → deduplicated"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    // ── Stream 3: branches unreachable via the parser ─────────────────────────

    #[test]
    fn execute_transact_non_keyword_attribute_error() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        // Construct a transact with a String attribute (not a keyword)
        let cmd = DatalogCommand::Transact(Transaction {
            facts: vec![Pattern::new(
                EdnValue::Keyword(":e".to_string()),
                EdnValue::String("not-a-keyword".to_string()),
                EdnValue::String("value".to_string()),
            )],
            valid_from: None,
            valid_to: None,
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "non-keyword attribute in transact must fail");
    }

    #[test]
    fn execute_retract_non_keyword_attribute_error() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = DatalogCommand::Retract(Transaction {
            facts: vec![Pattern::new(
                EdnValue::Keyword(":e".to_string()),
                EdnValue::Integer(42),
                EdnValue::String("value".to_string()),
            )],
            valid_from: None,
            valid_to: None,
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "non-keyword attribute in retract must fail");
    }

    #[test]
    fn execute_transact_pseudo_attr_error() {
        // Exercises executor.rs line 103: Pseudo(_) arm in execute_transact
        use crate::query::datalog::types::{PseudoAttr, Transaction};
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = DatalogCommand::Transact(Transaction {
            facts: vec![Pattern::pseudo(
                EdnValue::Keyword(":e".to_string()),
                PseudoAttr::ValidFrom,
                EdnValue::Integer(0),
            )],
            valid_from: None,
            valid_to: None,
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "transacting a pseudo-attribute must fail");
    }

    #[test]
    fn execute_retract_pseudo_attr_error() {
        // Exercises executor.rs line 139: Pseudo(_) arm in execute_retract
        use crate::query::datalog::types::{PseudoAttr, Transaction};
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = DatalogCommand::Retract(Transaction {
            facts: vec![Pattern::pseudo(
                EdnValue::Keyword(":e".to_string()),
                PseudoAttr::TxCount,
                EdnValue::Integer(0),
            )],
            valid_from: None,
            valid_to: None,
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "retracting a pseudo-attribute must fail");
    }

    #[test]
    fn execute_rule_empty_head_error() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = DatalogCommand::Rule(Rule {
            head: vec![],
            body: vec![WhereClause::Pattern(Pattern::new(
                EdnValue::Symbol("?x".to_string()),
                EdnValue::Keyword(":a".to_string()),
                EdnValue::Symbol("?v".to_string()),
            ))],
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "rule with empty head must fail");
    }

    #[test]
    fn execute_rule_non_symbol_head_error() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = DatalogCommand::Rule(Rule {
            head: vec![EdnValue::Integer(99)], // not a Symbol
            body: vec![WhereClause::Pattern(Pattern::new(
                EdnValue::Symbol("?x".to_string()),
                EdnValue::Keyword(":a".to_string()),
                EdnValue::Symbol("?v".to_string()),
            ))],
        });
        let r = executor.execute(cmd);
        assert!(r.is_err(), "rule head starting with non-symbol must fail");
    }

    // ── Float arithmetic edge cases ──────────────────────────────────────────

    #[test]
    fn test_eval_float_div_by_zero_is_err() {
        // Line 1096: rf == 0.0 → Err(()) for float division
        let e = Expr::BinOp(
            BinOp::Div,
            Box::new(Expr::Lit(Value::Float(5.0))),
            Box::new(Expr::Lit(Value::Float(0.0))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Err(()));
    }

    #[test]
    fn test_eval_float_div_succeeds() {
        // Line 1096 false branch: rf != 0.0 → Ok(Float)
        let e = Expr::BinOp(
            BinOp::Div,
            Box::new(Expr::Lit(Value::Float(6.0))),
            Box::new(Expr::Lit(Value::Float(2.0))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Float(3.0)));
    }

    #[test]
    fn test_eval_float_sub() {
        // Line 1079-1085: float subtraction
        let e = Expr::BinOp(
            BinOp::Sub,
            Box::new(Expr::Lit(Value::Float(5.0))),
            Box::new(Expr::Lit(Value::Float(2.0))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Float(3.0)));
    }

    #[test]
    fn test_eval_float_mul() {
        // Line 1087-1093: float multiplication
        let e = Expr::BinOp(
            BinOp::Mul,
            Box::new(Expr::Lit(Value::Float(3.0))),
            Box::new(Expr::Lit(Value::Float(4.0))),
        );
        assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Float(12.0)));
    }

    // ── Aggregation edge cases ────────────────────────────────────────────────

    #[test]
    fn test_agg_count_empty_bindings_returns_zero() {
        // (count ?x) with no matching facts → zero bindings → special-case returns 0
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command("(query [:find (count ?x) :where [?x :no-such-attr _]])")
            .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "should return one row with count 0");
                assert_eq!(results[0][0], crate::graph::types::Value::Integer(0));
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_agg_sum_empty_no_grouping_returns_zero() {
        // (sum ?v) with no matching facts and no grouping vars
        // bindings is empty → `has_grouping_vars` is false but it's not count → returns []
        let storage = FactStorage::new();
        storage
            .transact(
                vec![(
                    Uuid::new_v4(),
                    ":item/price".to_string(),
                    crate::graph::types::Value::Integer(50),
                )],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // Query for non-existing attribute to produce empty bindings, then sum
        let cmd = parse_datalog_command("(query [:find (sum ?v) :where [?x :no-such-attr ?v]])")
            .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                // empty bindings with non-count agg and no grouping → returns []
                assert_eq!(results.len(), 0, "empty bindings with sum returns no rows");
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_agg_sum_distinct_float_values() {
        // sum-distinct on float values exercises the SumDistinct + has_float path
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":item/weight".to_string(),
                        crate::graph::types::Value::Float(1.5),
                    ),
                    (
                        e2,
                        ":item/weight".to_string(),
                        crate::graph::types::Value::Float(1.5),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        let cmd =
            parse_datalog_command("(query [:find (sum-distinct ?w) :where [?e :item/weight ?w]])")
                .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                // Both have weight 1.5 but sum-distinct deduplicates → result is 1.5
                assert_eq!(results.len(), 1, "expected one result row");
                assert_eq!(
                    results[0][0],
                    crate::graph::types::Value::Float(1.5),
                    "sum-distinct of [1.5, 1.5] should be 1.5"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_agg_min_max_on_all_null_group_skips_row() {
        // min/max on a group where all values are Null → row is skipped
        // We insert a fact with value Null and query for min
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        storage
            .transact(
                vec![(
                    e1,
                    ":item/score".to_string(),
                    crate::graph::types::Value::Null,
                )],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // min on only Null values → "no non-null values in group" → row skipped → 0 rows
        let cmd = parse_datalog_command("(query [:find (min ?s) :where [?e :item/score ?s]])")
            .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    0,
                    "min on all-null group should produce 0 rows"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_agg_min_on_strings() {
        // min on strings exercises the String comparison path in apply_agg_func
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":item/name".to_string(),
                        crate::graph::types::Value::String("banana".to_string()),
                    ),
                    (
                        e2,
                        ":item/name".to_string(),
                        crate::graph::types::Value::String("apple".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command("(query [:find (min ?n) :where [?e :item/name ?n]])")
            .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "expected one result");
                assert_eq!(
                    results[0][0],
                    crate::graph::types::Value::String("apple".to_string()),
                    "min of strings should return lexicographically smallest"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_agg_max_on_floats() {
        // max on floats exercises the Float comparison path
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":item/score".to_string(),
                        crate::graph::types::Value::Float(3.5),
                    ),
                    (
                        e2,
                        ":item/score".to_string(),
                        crate::graph::types::Value::Float(2.5),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command("(query [:find (max ?s) :where [?e :item/score ?s]])")
            .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "expected one result");
                assert_eq!(
                    results[0][0],
                    crate::graph::types::Value::Float(3.5),
                    "max of floats should return largest"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    // ── evaluate_branch / apply_or_clauses edge cases ────────────────────────

    #[test]
    fn test_evaluate_branch_with_timestamp_valid_at() {
        // Exercises executor.rs lines 930-931: evaluate_branch with Timestamp/AnyValidTime
        use crate::query::datalog::types::ValidAt;
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        storage
            .transact(
                vec![(
                    e1,
                    ":tag".to_string(),
                    crate::graph::types::Value::Integer(1),
                )],
                None,
            )
            .unwrap();
        let facts: Arc<[crate::graph::types::Fact]> =
            Arc::from(storage.get_asserted_facts().unwrap().as_slice());
        let rules = crate::query::datalog::rules::RuleRegistry::new();
        let branch = vec![WhereClause::Pattern(Pattern::new(
            EdnValue::Symbol("?e".to_string()),
            EdnValue::Keyword(":tag".to_string()),
            EdnValue::Symbol("?v".to_string()),
        ))];
        let mut initial = std::collections::HashMap::new();
        initial.insert("?seed".to_string(), crate::graph::types::Value::Integer(0));

        // Line 930: Some(ValidAt::Timestamp(t)) arm
        let ts_result = evaluate_branch(
            &branch,
            vec![initial.clone()],
            facts.clone(),
            &rules,
            None,
            Some(ValidAt::Timestamp(crate::graph::types::tx_id_now() as i64)),
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(ts_result.len(), 1, "timestamp valid_at should match");

        // Line 931: Some(ValidAt::AnyValidTime) arm
        let any_result = evaluate_branch(
            &branch,
            vec![initial],
            facts,
            &rules,
            None,
            Some(ValidAt::AnyValidTime),
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(any_result.len(), 1, "any_valid_time should match");
    }

    #[test]
    fn test_execute_query_with_rules_valid_at_timestamp() {
        // Exercises executor.rs lines 341-342 (valid_at_value in execute_query_with_rules
        // for Timestamp and AnyValidTime arms) and lines 348-350 (hard-error guard).
        use crate::query::datalog::parser::parse_datalog_command;
        use crate::query::datalog::types::ValidAt;
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Register a rule so the query routes through execute_query_with_rules
        let rule_cmd = parse_datalog_command(r#"(rule [(tagged ?e) [?e :item/tag ?v]])"#)
            .expect("rule parse failed");
        executor.execute(rule_cmd).expect("rule register failed");

        // Transact a fact
        executor
            .execute(
                parse_datalog_command(r#"(transact [[:item1 :item/tag "x"]])"#)
                    .expect("transact parse failed"),
            )
            .expect("transact failed");

        // Lines 341-342: call execute_query_with_rules directly with Timestamp and AnyValidTime.
        // The public execute() routing may bypass it if query_uses_rules returns false;
        // calling the private method directly guarantees coverage.
        let q_ts = crate::query::datalog::types::DatalogQuery {
            find: vec![crate::query::datalog::types::FindSpec::Variable(
                "?e".to_string(),
            )],
            where_clauses: vec![],
            as_of: None,
            valid_at: Some(ValidAt::Timestamp(946684800000)), // 2000-01-01
            with_vars: vec![],
        };
        let r_ts = executor.execute_query_with_rules(q_ts);
        assert!(
            r_ts.is_ok(),
            "execute_query_with_rules with Timestamp must not error"
        );

        let q_any = crate::query::datalog::types::DatalogQuery {
            find: vec![crate::query::datalog::types::FindSpec::Variable(
                "?e".to_string(),
            )],
            where_clauses: vec![],
            as_of: None,
            valid_at: Some(ValidAt::AnyValidTime),
            with_vars: vec![],
        };
        let r_any = executor.execute_query_with_rules(q_any);
        assert!(
            r_any.is_ok(),
            "execute_query_with_rules with AnyValidTime must not error"
        );

        // Lines 348-350: hard-error guard in execute_query_with_rules
        // Per-fact pseudo-attr without :any-valid-time in a rules query
        let err_cmd = parse_datalog_command(
            "(query [:find ?e ?vf :where (tagged ?e) [?e :db/valid-from ?vf]])",
        )
        .expect("err query parse failed");
        let err_result = executor.execute(err_cmd);
        assert!(
            err_result.is_err(),
            "per-fact pseudo-attr without :any-valid-time in rules query must fail"
        );
    }

    #[test]
    fn test_evaluate_branch_empty_incoming_returns_empty() {
        // evaluate_branch with empty incoming bindings → returns [] immediately (line 842)
        let storage = FactStorage::new();
        storage
            .transact(
                vec![(
                    Uuid::new_v4(),
                    ":a".to_string(),
                    crate::graph::types::Value::Integer(1),
                )],
                None,
            )
            .unwrap();
        let facts: Arc<[crate::graph::types::Fact]> =
            Arc::from(storage.get_asserted_facts().unwrap().as_slice());
        let rules = crate::query::datalog::rules::RuleRegistry::new();
        let branch = vec![WhereClause::Pattern(Pattern::new(
            EdnValue::Symbol("?x".to_string()),
            EdnValue::Keyword(":a".to_string()),
            EdnValue::Symbol("?v".to_string()),
        ))];
        let result = evaluate_branch(
            &branch,
            vec![],
            facts,
            &rules,
            None,
            None,
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(result.len(), 0, "empty incoming should return empty");
    }

    #[test]
    fn test_evaluate_branch_no_match_patterns_empty_bindings_returns_empty() {
        // evaluate_branch: patterns exist but match nothing → bindings is empty (line 865)
        let storage = FactStorage::new();
        let facts: Arc<[crate::graph::types::Fact]> =
            Arc::from(storage.get_asserted_facts().unwrap().as_slice());
        let rules = crate::query::datalog::rules::RuleRegistry::new();
        // Branch has a pattern that won't match empty storage
        let branch = vec![WhereClause::Pattern(Pattern::new(
            EdnValue::Symbol("?x".to_string()),
            EdnValue::Keyword(":no-such-attr".to_string()),
            EdnValue::Symbol("?v".to_string()),
        ))];
        // Seed with one binding so the branch has something to work with
        let mut initial = std::collections::HashMap::new();
        initial.insert("?init".to_string(), crate::graph::types::Value::Integer(1));
        let result = evaluate_branch(
            &branch,
            vec![initial],
            facts,
            &rules,
            None,
            None,
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(
            result.len(),
            0,
            "no matching facts should return empty bindings"
        );
    }

    #[test]
    fn test_evaluate_branch_not_filter_excludes_matching() {
        // evaluate_branch with Not clause: entities that match the not body are excluded (line 909)
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":color".to_string(),
                        crate::graph::types::Value::Keyword(":red".to_string()),
                    ),
                    (
                        e2,
                        ":color".to_string(),
                        crate::graph::types::Value::Keyword(":blue".to_string()),
                    ),
                    (
                        e1,
                        ":flagged".to_string(),
                        crate::graph::types::Value::Boolean(true),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // Query: find entities with :color but not :flagged
        // Uses the not_body_matches path in not-post-filter (line 909)
        let cmd = parse_datalog_command(
            "(query [:find ?e :where [?e :color ?c] (not-join [?e] [?e :flagged ?fv])])",
        )
        .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "only e2 (non-flagged) should match");
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_or_join_deduplication() {
        // or-join where both branches bind ?e → duplicate bindings are deduplicated
        // (line 991: if !result.contains(&b) { result.push(b); })
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":color".to_string(),
                        crate::graph::types::Value::Keyword(":red".to_string()),
                    ),
                    (
                        e1,
                        ":shape".to_string(),
                        crate::graph::types::Value::Keyword(":circle".to_string()),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // or-join [?e] with two branches that both match e1 → deduplicated to 1
        // ?e must be bound by an earlier clause; use :color as the primary clause
        let cmd = parse_datalog_command(
            "(query [:find ?e :where [?e :color ?c] (or-join [?e] [?e :color ?c2] [?e :shape ?s])])",
        )
        .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    1,
                    "e1 should appear once despite two matching branches"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_transact_with_tx_level_valid_time() {
        // Exercises the tx_opts = Some(...) path when valid_from/valid_to set at tx level (line 66)
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command(
            r#"(transact {:valid-from "2020-01-01T00:00:00Z" :valid-to "2025-01-01T00:00:00Z"} [[:alice :person/name "Alice"]])"#,
        )
        .expect("parse with tx-level valid-time should succeed");
        let result = executor.execute(cmd);
        assert!(
            result.is_ok(),
            "transact with tx-level valid-time should succeed"
        );
    }

    #[test]
    fn test_transact_with_per_fact_valid_time() {
        // Exercises the per_fact_opts = Some(...) path when valid_from/valid_to set per fact (line 87)
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command(
            r#"(transact [[:alice :person/name "Alice" {:valid-from "2020-01-01T00:00:00Z"}]])"#,
        )
        .expect("parse with per-fact valid-time should succeed");
        let result = executor.execute(cmd);
        assert!(
            result.is_ok(),
            "transact with per-fact valid-time should succeed"
        );
    }

    #[test]
    fn test_transact_with_valid_to_only_at_tx_level() {
        // Exercises the `|| tx.valid_to.is_some()` branch (line 66 col 53)
        // when valid_from is None but valid_to is Some
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command(
            r#"(transact {:valid-to "2025-01-01T00:00:00Z"} [[:alice :person/name "Alice"]])"#,
        )
        .expect("parse with tx-level valid-to only should succeed");
        let result = executor.execute(cmd);
        assert!(
            result.is_ok(),
            "transact with valid-to only at tx level should succeed"
        );
    }

    #[test]
    fn test_transact_with_valid_to_only_per_fact() {
        // Exercises the `|| pattern.valid_to.is_some()` branch (line 87 col 68)
        // when per-fact valid_from is None but valid_to is Some
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);
        let cmd = parse_datalog_command(
            r#"(transact [[:alice :person/name "Alice" {:valid-to "2025-01-01T00:00:00Z"}]])"#,
        )
        .expect("parse with per-fact valid-to only should succeed");
        let result = executor.execute(cmd);
        assert!(
            result.is_ok(),
            "transact with valid-to only per fact should succeed"
        );
    }

    #[test]
    fn test_evaluate_branch_empty_patterns_passes_incoming_through() {
        // Line 859: patterns.is_empty() = true → bindings = incoming (pass through)
        // Achieved when branch contains only Not/Expr clauses, no Pattern/RuleInvocation
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        storage
            .transact(
                vec![(
                    e1,
                    ":a".to_string(),
                    crate::graph::types::Value::Integer(10),
                )],
                None,
            )
            .unwrap();
        let facts: Arc<[crate::graph::types::Fact]> =
            Arc::from(storage.get_asserted_facts().unwrap().as_slice());
        let rules = crate::query::datalog::rules::RuleRegistry::new();

        // Branch with only an Expr clause (no patterns) — patterns.is_empty() = true
        let branch = vec![WhereClause::Expr {
            expr: crate::query::datalog::types::Expr::Lit(crate::graph::types::Value::Boolean(
                true,
            )),
            binding: None,
        }];
        // Incoming with one binding
        let mut initial = std::collections::HashMap::new();
        initial.insert("?x".to_string(), crate::graph::types::Value::Integer(42));
        let result = evaluate_branch(
            &branch,
            vec![initial],
            facts,
            &rules,
            None,
            None,
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        // The expr is truthy so the binding passes through
        assert_eq!(
            result.len(),
            1,
            "expr-only branch should pass binding through"
        );
    }

    #[test]
    fn test_evaluate_branch_or_clause_produces_empty_bindings() {
        // Line 879: bindings empty after apply_or_clauses → return Ok([])
        // This happens when an Or clause produces no results
        let storage = FactStorage::new();
        // Empty storage → no facts
        let facts: Arc<[crate::graph::types::Fact]> =
            Arc::from(storage.get_asserted_facts().unwrap().as_slice());
        let rules = crate::query::datalog::rules::RuleRegistry::new();

        // Branch with an Or clause that matches nothing
        let or_branch = vec![WhereClause::Pattern(Pattern::new(
            EdnValue::Symbol("?x".to_string()),
            EdnValue::Keyword(":no-attr".to_string()),
            EdnValue::Symbol("?v".to_string()),
        ))];
        let branch = vec![WhereClause::Or(vec![or_branch])];

        let mut initial = std::collections::HashMap::new();
        initial.insert("?seed".to_string(), crate::graph::types::Value::Integer(1));
        let result = evaluate_branch(
            &branch,
            vec![initial],
            facts,
            &rules,
            None,
            None,
            &FunctionRegistry::with_builtins(),
        )
        .unwrap();
        assert_eq!(
            result.len(),
            0,
            "or clause with no matches should yield empty"
        );
    }

    #[test]
    fn test_evaluate_branch_not_join_excludes_matching() {
        // Line 914: evaluate_not_join returns true inside evaluate_branch → exclude
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":status".to_string(),
                        crate::graph::types::Value::Keyword(":active".to_string()),
                    ),
                    (
                        e2,
                        ":status".to_string(),
                        crate::graph::types::Value::Keyword(":inactive".to_string()),
                    ),
                    (
                        e2,
                        ":blocked".to_string(),
                        crate::graph::types::Value::Boolean(true),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // not-join: exclude entities that have :blocked = true
        let cmd = parse_datalog_command(
            "(query [:find ?e :where [?e :status ?s] (not-join [?e] [?e :blocked ?b])])",
        )
        .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    1,
                    "only non-blocked entity should be returned"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn test_not_body_expr_only_filters_binding() {
        // Exercises not_body_matches with patterns.is_empty() (Expr-only not body) at line 561
        let storage = FactStorage::new();
        let e1 = Uuid::new_v4();
        let e2 = Uuid::new_v4();
        storage
            .transact(
                vec![
                    (
                        e1,
                        ":item/price".to_string(),
                        crate::graph::types::Value::Integer(200),
                    ),
                    (
                        e2,
                        ":item/price".to_string(),
                        crate::graph::types::Value::Integer(50),
                    ),
                ],
                None,
            )
            .unwrap();
        let executor = DatalogExecutor::new(storage);
        // not with expr-only body: exclude items where price > 100
        let cmd = parse_datalog_command(
            "(query [:find ?e :where [?e :item/price ?p] (not [(> ?p 100)])])",
        )
        .expect("parse failed");
        let result = executor.execute(cmd).expect("query failed");
        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(
                    results.len(),
                    1,
                    "only the item with price 50 should survive"
                );
            }
            _ => panic!("expected QueryResults"),
        }
    }

    #[test]
    fn window_sum_resets_per_partition() {
        use super::super::functions::FunctionRegistry;
        use super::super::types::{Order, WindowFunc, WindowSpec};

        let mut bindings: Vec<std::collections::HashMap<String, Value>> = vec![
            [
                ("dept".into(), Value::String("A".into())),
                ("salary".into(), Value::Integer(10)),
            ]
            .into_iter()
            .collect(),
            [
                ("dept".into(), Value::String("A".into())),
                ("salary".into(), Value::Integer(20)),
            ]
            .into_iter()
            .collect(),
            [
                ("dept".into(), Value::String("B".into())),
                ("salary".into(), Value::Integer(100)),
            ]
            .into_iter()
            .collect(),
        ];
        let find_specs = vec![
            FindSpec::Variable("dept".into()),
            FindSpec::Variable("salary".into()),
            FindSpec::Window(WindowSpec {
                func: WindowFunc::Sum,
                var: Some("salary".into()),
                partition_by: Some("dept".into()),
                order_by: "salary".into(),
                order: Order::Asc,
            }),
        ];
        let registry = FunctionRegistry::with_builtins();
        apply_window_functions(&mut bindings, &find_specs, &registry).expect("window");

        // Partition A: 10 → sum=10, 20 → sum=30
        let row_a10 = bindings
            .iter()
            .find(|b| b.get("salary") == Some(&Value::Integer(10)))
            .unwrap();
        assert_eq!(row_a10.get("__win_2"), Some(&Value::Integer(10)));

        let row_a20 = bindings
            .iter()
            .find(|b| b.get("salary") == Some(&Value::Integer(20)))
            .unwrap();
        assert_eq!(row_a20.get("__win_2"), Some(&Value::Integer(30)));

        // Partition B: 100 → sum=100 (accumulator reset, NOT 130)
        let row_b100 = bindings
            .iter()
            .find(|b| b.get("salary") == Some(&Value::Integer(100)))
            .unwrap();
        assert_eq!(row_b100.get("__win_2"), Some(&Value::Integer(100)));
    }
}

#[cfg(test)]
mod selective_lookup_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;

    fn make_db_with_entities(n: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(50) {
            let batch_end = (batch_start + 50).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!(r#"[:e{i} :name "entity{i}"]"#, i = i));
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        exec
    }

    #[test]
    fn entity_bound_query_returns_correct_results() {
        let exec = make_db_with_entities(100);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    r#"(query [:find ?n :where [:e5 :name ?n]])"#,
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), 1, "expected exactly 1 result for entity :e5");
            assert_eq!(
                results[0][0],
                crate::graph::types::Value::String("entity5".to_string())
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn attribute_bound_query_returns_correct_results() {
        let exec = make_db_with_entities(100);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e ?v :where [?e :val ?v]])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(
                results.len(),
                100,
                "expected 100 results for :val attribute scan"
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn as_of_query_still_works_after_change() {
        let exec = make_db_with_entities(10);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?n :where [?e :name ?n] :as-of 1])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert!(!results.is_empty(), "expected results from as-of 1 query");
        } else {
            panic!("expected QueryResults");
        }
    }
}

#[cfg(test)]
mod not_hash_join_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;

    fn make_not_db(n: usize, excluded: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        for batch_start in (0..excluded).step_by(100) {
            let batch_end = (batch_start + 100).min(excluded);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :banned true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        exec
    }

    #[test]
    fn not_filter_returns_correct_count() {
        let n = 1_000;
        let excluded = n / 10; // 100 banned
        let exec = make_not_db(n, excluded);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] (not [?e :banned true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(
                results.len(),
                n - excluded,
                "expected {} results after not-filter",
                n - excluded
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn not_join_filter_returns_correct_count() {
        let n = 1_000;
        let excluded = n / 10;
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        exec.execute(
            crate::query::datalog::parser::parse_datalog_command(
                "(transact [[:d-bad :status :bad]])",
            )
            .unwrap(),
        )
        .unwrap();
        for batch_start in (0..excluded).step_by(100) {
            let batch_end = (batch_start + 100).min(excluded);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :dep :d-bad]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (not-join [?e] [?e :dep ?d] [?d :status :bad])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(
                results.len(),
                n - excluded,
                "expected {} results after not-join-filter",
                n - excluded
            );
        } else {
            panic!("expected QueryResults");
        }
    }
}

#[cfg(test)]
mod or_hash_join_tests {
    use crate::graph::FactStorage;
    use crate::query::datalog::executor::DatalogExecutor;

    fn make_or_db(n: usize, a_count: usize, b_count: usize) -> DatalogExecutor {
        let storage = FactStorage::new();
        let exec = DatalogExecutor::new(storage);
        for batch_start in (0..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :val {i}]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        for batch_start in (0..a_count).step_by(100) {
            let batch_end = (batch_start + 100).min(a_count);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :tag-a true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        let b_start = n.saturating_sub(b_count);
        for batch_start in (b_start..n).step_by(100) {
            let batch_end = (batch_start + 100).min(n);
            let mut cmd = String::from("(transact [");
            for i in batch_start..batch_end {
                cmd.push_str(&format!("[:e{i} :tag-b true]", i = i));
            }
            cmd.push_str("])");
            exec.execute(crate::query::datalog::parser::parse_datalog_command(&cmd).unwrap())
                .unwrap();
        }
        exec
    }

    #[test]
    fn or_clause_returns_correct_count() {
        // 1000 entities, first 250 have :tag-a, last 250 have :tag-b, no overlap
        let n = 1_000;
        let a = n / 4;
        let b = n / 4;
        let exec = make_or_db(n, a, b);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(
                results.len(),
                a + b,
                "expected {} results from or (a={} b={} no overlap)",
                a + b,
                a,
                b
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn or_join_clause_returns_correct_count() {
        let n = 1_000;
        let a = n / 4;
        let b = n / 4;
        let exec = make_or_db(n, a, b);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or-join [?e] [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(
                results.len(),
                a + b,
                "expected {} results from or-join",
                a + b
            );
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn or_clause_with_overlap_deduplicates() {
        // 100 entities, all have both :tag-a and :tag-b → result should be 100, not 200
        let n = 100;
        let exec = make_or_db(n, n, n);
        let result = exec
            .execute(
                crate::query::datalog::parser::parse_datalog_command(
                    "(query [:find ?e :where [?e :val ?v] \
                     (or [?e :tag-a true] [?e :tag-b true])])",
                )
                .unwrap(),
            )
            .unwrap();
        if let crate::query::datalog::executor::QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), n, "expected {} deduplicated results", n);
        } else {
            panic!("expected QueryResults");
        }
    }
}

#[cfg(test)]
mod pushdown_tests {
    use crate::graph::FactStorage;
    use crate::graph::types::Value;
    use crate::query::datalog::executor::{DatalogExecutor, QueryResult};
    use crate::query::datalog::parser::parse_datalog_command;

    #[test]
    fn test_expr_pushdown_preserves_query_results() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());
        executor
            .execute(
                parse_datalog_command("(transact [[:e1 :val 10] [:e2 :val 20] [:e3 :val 30]])")
                    .unwrap(),
            )
            .unwrap();
        let result = executor
            .execute(
                parse_datalog_command("(query [:find ?e ?v :where [?e :val ?v] [(> ?v 15)]])")
                    .unwrap(),
            )
            .unwrap();
        if let QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), 2, "only :e2 and :e3 have :val > 15");
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn test_expr_pushdown_multi_pattern_preserves_results() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());
        executor
            .execute(
                parse_datalog_command(
                    r#"(transact [[:e1 :val 5] [:e1 :name "a"] [:e2 :val 20] [:e2 :name "b"]])"#,
                )
                .unwrap(),
            )
            .unwrap();
        let result = executor
            .execute(
                parse_datalog_command(
                    r#"(query [:find ?e ?n :where [?e :val ?v] [?e :name ?n] [(> ?v 10)]])"#,
                )
                .unwrap(),
            )
            .unwrap();
        if let QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), 1, "only :e2 passes the predicate");
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn test_expr_binding_form_preserves_results() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());
        executor
            .execute(parse_datalog_command("(transact [[:e1 :val 5] [:e2 :val 10]])").unwrap())
            .unwrap();
        let result = executor
            .execute(
                parse_datalog_command(
                    "(query [:find ?e ?doubled :where [?e :val ?v] [(* ?v 2) ?doubled]])",
                )
                .unwrap(),
            )
            .unwrap();
        if let QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), 2, "both entities must appear");
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn test_expr_pushdown_with_rules_preserves_results() {
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage.clone());
        executor
            .execute(
                parse_datalog_command("(transact [[:e1 :val 5] [:e2 :val 20] [:e3 :val 30]])")
                    .unwrap(),
            )
            .unwrap();
        executor
            .execute(parse_datalog_command("(rule [(high ?e) [?e :val ?v] [(> ?v 15)]])").unwrap())
            .unwrap();
        let result = executor
            .execute(parse_datalog_command("(query [:find ?e :where (high ?e)])").unwrap())
            .unwrap();
        if let QueryResult::QueryResults { results, .. } = result {
            assert_eq!(results.len(), 2, "only :e2 and :e3 qualify");
        } else {
            panic!("expected QueryResults");
        }
    }

    #[test]
    fn test_not_clause_ordering_correctness() {
        // Two `not` clauses given in expensive-first source order; results must be
        // identical regardless of which order the optimizer chooses to evaluate them.
        // This is a correctness regression guard — semantics must not change.
        let storage = FactStorage::new();
        let executor = DatalogExecutor::new(storage);

        // Transact 3 items: widget, gadget, doohickey
        executor
            .execute(
                parse_datalog_command(
                    r#"(transact [[:item1 :item/name "widget"]
                                  [:item2 :item/name "gadget"]
                                  [:item3 :item/name "doohickey"]])"#,
                )
                .unwrap(),
            )
            .unwrap();

        // Query: find items that are NOT "widget" AND NOT "gadget"
        // Clauses are given in expensive-first order to exercise the cost-based sort.
        let result = executor
            .execute(
                parse_datalog_command(
                    r#"(query [:find ?name
                               :where [?e :item/name ?name]
                                      (not [?e :item/name "gadget"])
                                      (not [?e :item/name "widget"])])"#,
                )
                .unwrap(),
            )
            .unwrap();

        match result {
            QueryResult::QueryResults { results, .. } => {
                assert_eq!(results.len(), 1, "only doohickey should pass");
                if let Value::String(ref s) = results[0][0] {
                    assert_eq!(s.as_str(), "doohickey", "result should be doohickey");
                } else {
                    panic!("expected a String value for the result");
                }
            }
            _ => panic!("expected QueryResults"),
        }
    }
}