sqry_core/query/executor/

graph_eval.rs

//! Graph-native predicate evaluation.
//!
//! This module provides graph-native predicate evaluation for queries against
//! `CodeGraph`, bypassing the legacy index path.
//!
//! # Design (v6 - CodeGraph-Native Query Executor)
//!
//! Key semantics preserved from the legacy index path:
//! - `name:` uses `segments_match` for qualified name suffix matching
//! - `kind:`, `lang:`, `visibility:`, `scope.*` are **case-sensitive**
//! - `imports:` per-node — matches a node iff its own outgoing `Imports`
//!   edge target/alias/wildcard matches, or it is an `Import` node whose
//!   text matches. Aligned with `sqry-db::queries::ImportsQuery` per the
//!   Phase N "Unified Surface Contract" (planner-IR is canonical, all
//!   transports mirror it). The previous file-scoped semantic was retired
//!   in DB15 to remove the cross-engine divergence flagged by Codex's
//!   DB14 review.
//! - `references:` includes `References` + `Calls` + `Imports` + `FfiCall` edges
//! - `callers:` checks OUTGOING edges (find nodes that call X)
//! - `callees:` checks INCOMING edges (find nodes called by X)
//! - Relation predicates use `segments_match` for qualified names
//!
//! # Limitations (v1)
//!
//! The following predicates are **NOT SUPPORTED** in graph backend v1:
//! - `returns:` - requires signature parsing/metadata not in `NodeEntry`
//! - Plugin fields - requires metadata `HashMap` not in `NodeEntry`
//! - Numeric operators - requires metadata values
//!
//! **Supported boolean predicates**: `async:true`, `static:true`

use crate::graph::unified::FileId;
use crate::graph::unified::concurrent::CodeGraph;
use crate::graph::unified::edge::{EdgeKind, StoreEdgeRef};
use crate::graph::unified::node::{NodeId, NodeKind};
use crate::graph::unified::resolution::{
    canonicalize_graph_qualified_name, display_graph_qualified_name,
};
use crate::graph::unified::storage::arena::NodeEntry;
use crate::plugin::PluginManager;
use crate::query::name_matching::segments_match;
use crate::query::regex_cache::{CompiledRegex, get_or_compile_regex};
use crate::query::types::{Condition, Expr, JoinEdgeKind, JoinExpr, Operator, Value};
use anyhow::{Result, anyhow};
use std::collections::{HashMap, HashSet};
use std::path::Path;

/// Try `re.is_match(text)`, logging a warning and returning `false` if the
/// regex engine hits its backtrack limit.  This prevents silent failures
/// while keeping the predicate-evaluation chain infallible.
fn regex_is_match(re: &CompiledRegex, text: &str) -> bool {
    match re.is_match(text) {
        Ok(b) => b,
        Err(e) => {
            log::warn!("regex match aborted (backtrack limit?): {e}");
            false
        }
    }
}
use std::sync::Arc;

/// Cache of precomputed subquery result sets, keyed by `(span.start, span.end)`.
///
/// Subquery expressions from the AST are evaluated once against all graph nodes
/// and their results cached here. During per-node evaluation, matchers look up
/// the cache instead of re-evaluating the full graph. This reduces subquery cost
/// from O(N^2) to O(N) where N is the number of graph nodes.
type SubqueryCache = HashMap<(usize, usize), Arc<HashSet<NodeId>>>;

/// Context for graph-native predicate evaluation.
///
/// Encapsulates all dependencies needed for evaluating predicates against
/// a `CodeGraph`. The context includes precomputed caches for performance.
pub struct GraphEvalContext<'a> {
    /// Reference to the code graph
    pub graph: &'a CodeGraph,
    /// Plugin manager for detecting plugin fields
    pub plugin_manager: &'a PluginManager,
    /// Workspace root for relative path resolution in `path:` predicates
    pub workspace_root: Option<&'a Path>,
    /// If true, disable parallel execution
    pub disable_parallel: bool,
    /// Precomputed subquery result sets, keyed by `(span.start, span.end)`.
    /// Populated by `precompute_subqueries()` before the per-node evaluation loop.
    pub subquery_cache: SubqueryCache,
}

impl<'a> GraphEvalContext<'a> {
    /// Creates a new evaluation context.
    #[must_use]
    pub fn new(graph: &'a CodeGraph, plugin_manager: &'a PluginManager) -> Self {
        Self {
            graph,
            plugin_manager,
            workspace_root: None,
            disable_parallel: false,
            subquery_cache: HashMap::new(),
        }
    }

    /// Sets the workspace root for relative path resolution.
    #[must_use]
    pub fn with_workspace_root(mut self, root: &'a Path) -> Self {
        self.workspace_root = Some(root);
        self
    }

    /// Disables parallel execution.
    #[must_use]
    pub fn with_parallel_disabled(mut self, disabled: bool) -> Self {
        self.disable_parallel = disabled;
        self
    }

    /// Precomputes all subquery result sets from the expression tree.
    ///
    /// This must be called before `evaluate_all` to avoid O(N^2) behavior
    /// where each subquery is re-evaluated for every candidate node.
    ///
    /// # Errors
    ///
    /// Returns an error if subquery evaluation fails.
    pub fn precompute_subqueries(&mut self, expr: &Expr) -> Result<()> {
        let mut subquery_exprs = Vec::new();
        collect_subquery_exprs(expr, &mut subquery_exprs);

        for (span_key, inner_expr) in subquery_exprs {
            if !self.subquery_cache.contains_key(&span_key) {
                let result_set = evaluate_subquery(self, inner_expr)?;
                self.subquery_cache.insert(span_key, Arc::new(result_set));
            }
        }
        Ok(())
    }
}

/// Collects all `Value::Subquery(inner)` expressions from the AST for precomputation.
///
/// Returns `(span_key, &Expr)` pairs where `span_key` is `(span.start, span.end)`.
///
/// Uses **post-order** traversal: nested (inner) subqueries appear before their
/// enclosing (outer) subqueries. This ensures `precompute_subqueries()` evaluates
/// dependencies first, so outer subquery evaluation can find inner results in the cache.
fn collect_subquery_exprs<'a>(expr: &'a Expr, out: &mut Vec<((usize, usize), &'a Expr)>) {
    match expr {
        Expr::Condition(cond) => {
            if let Value::Subquery(inner) = &cond.value {
                // Post-order: recurse into nested subqueries FIRST
                collect_subquery_exprs(inner, out);
                // Then record this (outer) subquery
                out.push(((cond.span.start, cond.span.end), inner));
            }
        }
        Expr::And(operands) | Expr::Or(operands) => {
            for op in operands {
                collect_subquery_exprs(op, out);
            }
        }
        Expr::Not(inner) => collect_subquery_exprs(inner, out),
        Expr::Join(join) => {
            collect_subquery_exprs(&join.left, out);
            collect_subquery_exprs(&join.right, out);
        }
    }
}

/// Evaluates query against all nodes, returning matching `NodeIds`.
///
/// # Errors
///
/// Returns an error if predicate evaluation fails (e.g., unsupported predicates).
pub fn evaluate_all(ctx: &mut GraphEvalContext, expr: &Expr) -> Result<Vec<NodeId>> {
    // Precompute all subquery result sets before the per-node evaluation loop.
    // This is critical for performance: without precomputation, each subquery
    // matcher would re-evaluate the full graph for every candidate node (O(N^2)).
    ctx.precompute_subqueries(expr)?;

    let arena = ctx.graph.nodes();

    // Create recursion guard
    let recursion_limits = crate::config::RecursionLimits::load_or_default()?;
    let expr_depth = recursion_limits.effective_expr_depth()?;
    let mut guard = crate::query::security::RecursionGuard::new(expr_depth)?;

    if ctx.disable_parallel {
        // Sequential evaluation
        let mut matches = Vec::new();
        for (id, entry) in arena.iter() {
            // Skip Phase 4c-prime unified-away losers — they remain in
            // the arena as inert duplicates so CSR row_ptr sizing stays
            // stable, but publish-visible query evaluation must not
            // surface them (Gate 0d iter-1 blocker). See
            // `NodeEntry::is_unified_loser`.
            if entry.is_unified_loser() {
                continue;
            }
            if evaluate_node(ctx, id, expr, &mut guard)? {
                matches.push(id);
            }
        }
        Ok(matches)
    } else {
        // Parallel evaluation - each thread needs its own guard
        use rayon::prelude::*;

        let node_ids: Vec<_> = arena
            .iter()
            .filter(|(_id, entry)| !entry.is_unified_loser())
            .map(|(id, _)| id)
            .collect();
        let results: Vec<Result<Option<NodeId>>> = node_ids
            .into_par_iter()
            .map(|id| {
                let mut thread_guard = crate::query::security::RecursionGuard::new(expr_depth)?;
                evaluate_node(ctx, id, expr, &mut thread_guard)
                    .map(|m| if m { Some(id) } else { None })
            })
            .collect();

        // Check for any errors and collect matches
        let mut matches = Vec::new();
        for result in results {
            if let Some(id) = result? {
                matches.push(id);
            }
        }
        Ok(matches)
    }
}

/// Evaluates a single node against an expression.
///
/// # Errors
///
/// Returns an error for unsupported predicates or if recursion depth exceeds the guard's limit.
pub fn evaluate_node(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    expr: &Expr,
    guard: &mut crate::query::security::RecursionGuard,
) -> Result<bool> {
    guard.enter()?;

    let result = match expr {
        Expr::Condition(cond) => evaluate_condition(ctx, node_id, cond),
        Expr::And(operands) => {
            for operand in operands {
                if !evaluate_node(ctx, node_id, operand, guard)? {
                    guard.exit();
                    return Ok(false);
                }
            }
            Ok(true)
        }
        Expr::Or(operands) => {
            for operand in operands {
                if evaluate_node(ctx, node_id, operand, guard)? {
                    guard.exit();
                    return Ok(true);
                }
            }
            Ok(false)
        }
        Expr::Not(inner) => Ok(!evaluate_node(ctx, node_id, inner, guard)?),
        Expr::Join(_) => {
            // Join expressions are evaluated at a higher level (execute_join),
            // not per-node. If we reach here, it's a programming error.
            Err(anyhow::anyhow!(
                "Join expressions cannot be evaluated per-node; use execute_join instead"
            ))
        }
    };

    guard.exit();
    result
}

fn evaluate_condition(ctx: &GraphEvalContext, node_id: NodeId, cond: &Condition) -> Result<bool> {
    let Some(entry) = ctx.graph.nodes().get(node_id) else {
        return Ok(false);
    };

    match cond.field.as_str() {
        "kind" => Ok(match_kind(ctx, entry, &cond.operator, &cond.value)),
        "name" => Ok(match_name(ctx, entry, &cond.operator, &cond.value)),
        "path" => Ok(match_path(ctx, entry, &cond.operator, &cond.value)),
        "lang" | "language" => Ok(match_lang(ctx, entry, &cond.operator, &cond.value)),
        "visibility" => Ok(match_visibility(ctx, entry, &cond.operator, &cond.value)),
        "async" => Ok(match_async(entry, &cond.operator, &cond.value)),
        "static" => Ok(match_static(entry, &cond.operator, &cond.value)),
        "callers" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_callers_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_callers(ctx, node_id, &cond.value))
            }
        }
        "callees" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_callees_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_callees(ctx, node_id, &cond.value))
            }
        }
        "imports" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_imports_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_imports(ctx, node_id, &cond.value))
            }
        }
        "exports" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_exports_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_exports(ctx, node_id, &cond.value))
            }
        }
        "references" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_references_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_references(ctx, node_id, &cond.operator, &cond.value))
            }
        }
        "impl" | "implements" => {
            if matches!(cond.value, Value::Subquery(_)) {
                let key = (cond.span.start, cond.span.end);
                let cached = ctx.subquery_cache.get(&key).cloned();
                match_implements_subquery(ctx, node_id, cached.as_deref())
            } else {
                Ok(match_implements(ctx, node_id, &cond.value))
            }
        }
        field if field.starts_with("scope.") => Ok(match_scope(
            ctx,
            node_id,
            field,
            &cond.operator,
            &cond.value,
        )),
        "returns" => Ok(match_returns(ctx, entry, &cond.operator, &cond.value)),
        field if is_plugin_field(ctx, field) => Err(anyhow!(
            "Plugin field '{field}' requires metadata not available in graph backend"
        )),
        _ => Ok(false), // Unknown field
    }
}

/// Checks if a field is a plugin-specific field.
fn is_plugin_field(ctx: &GraphEvalContext, field: &str) -> bool {
    // Check plugin registry for field descriptors
    let is_registered_field = ctx
        .plugin_manager
        .plugins()
        .iter()
        .flat_map(|plugin| plugin.fields().iter())
        .any(|descriptor| descriptor.name == field);

    if is_registered_field {
        return true;
    }

    // Fallback static list (for when registry not fully wired)
    // Note: "async" and "static" are now handled natively via `NodeEntry` flags
    matches!(
        field,
        "abstract" | "final" | "generic" | "parameters" | "arity"
    )
}

// ============================================================================
// Kind predicate (with regex + synonyms)
// ============================================================================

/// Maps synonyms to canonical kind names (case-sensitive).
///
/// Per v6 spec, kind: comparisons are case-sensitive.
/// Only exact matches for known synonyms are normalized.
fn normalize_kind(kind: &str) -> &str {
    match kind {
        // Rust/Go synonyms (case-sensitive)
        "trait" => "interface", // Rust trait = interface
        "impl" => "implementation",
        // Property synonyms
        "field" => "property",
        // Module synonyms
        "namespace" => "module",
        // Component synonyms
        "element" => "component",
        // CSS/Style synonyms
        "style" => "style_rule",
        "at_rule" => "style_at_rule",
        "css_var" | "custom_property" => "style_variable",
        // No lowercasing - return as-is for case-sensitive comparison
        _ => kind,
    }
}

fn match_kind(
    _ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    let actual = entry.kind.as_str();

    match (operator, value) {
        (Operator::Equal, Value::String(expected)) => {
            let normalized_expected = normalize_kind(expected);
            let normalized_actual = normalize_kind(actual);
            normalized_actual == normalized_expected
        }
        (Operator::Regex, Value::Regex(regex_val)) => get_or_compile_regex(
            &regex_val.pattern,
            regex_val.flags.case_insensitive,
            regex_val.flags.multiline,
            regex_val.flags.dot_all,
        )
        .map(|re| regex_is_match(&re, actual))
        .unwrap_or(false),
        _ => false,
    }
}

// ============================================================================
// Name predicate (EXACT match for equality, regex supported)
// ============================================================================

fn match_name(
    ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    match (operator, value) {
        // Use segments_match for qualified name matching:
        // - "connect" matches "database::connect" (suffix match)
        // - "foo::bar" matches "baz::foo::bar" (suffix match)
        // - "database::connect" matches "database::connect" (exact match)
        (Operator::Equal, Value::String(expected)) => {
            entry_query_texts(ctx.graph, entry).iter().any(|candidate| {
                language_aware_segments_match(ctx.graph, entry.file, candidate, expected)
            })
        }
        (Operator::Regex, Value::Regex(regex_val)) => get_or_compile_regex(
            &regex_val.pattern,
            regex_val.flags.case_insensitive,
            regex_val.flags.multiline,
            regex_val.flags.dot_all,
        )
        .map(|re| {
            entry_query_texts(ctx.graph, entry)
                .iter()
                .any(|candidate| regex_is_match(&re, candidate))
        })
        .unwrap_or(false),
        _ => false,
    }
}

// ============================================================================
// Path predicate (glob + regex)
// ============================================================================

/// Check if a pattern is relative (doesn't start with `/`).
fn is_relative_pattern(pattern: &str) -> bool {
    !pattern.starts_with('/')
}

fn match_path(
    ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    let Some(file_path) = ctx.graph.files().resolve(entry.file) else {
        return false;
    };

    match (operator, value) {
        (Operator::Equal, Value::String(pattern)) => {
            // Only strip workspace_root for RELATIVE patterns (parity with legacy index)
            let match_path = if is_relative_pattern(pattern) {
                if let Some(root) = ctx.workspace_root {
                    file_path
                        .strip_prefix(root)
                        .map_or_else(|_| file_path.to_path_buf(), std::path::Path::to_path_buf)
                } else {
                    file_path.to_path_buf()
                }
            } else {
                // Absolute pattern: match against full path
                file_path.to_path_buf()
            };
            globset::Glob::new(pattern)
                .map(|g| g.compile_matcher().is_match(&match_path))
                .unwrap_or(false)
        }
        (Operator::Regex, Value::Regex(regex_val)) => {
            // For regex, always use full path (matches legacy index behavior)
            get_or_compile_regex(
                &regex_val.pattern,
                regex_val.flags.case_insensitive,
                regex_val.flags.multiline,
                regex_val.flags.dot_all,
            )
            .map(|re| regex_is_match(&re, file_path.to_string_lossy().as_ref()))
            .unwrap_or(false)
        }
        _ => false,
    }
}

// ============================================================================
// Language predicate (from FILE REGISTRY, not NodeEntry)
// ============================================================================

/// Convert Language enum to canonical string for parity with legacy index.
///
/// Legacy index uses canonical names like "javascript", "typescript", etc.
/// `Language::Display` uses short forms like "js", "ts".
/// This function provides the canonical mapping for query parity.
fn language_to_canonical(lang: crate::graph::node::Language) -> &'static str {
    use crate::graph::node::Language;
    match lang {
        Language::C => "c",
        Language::Cpp => "cpp",
        Language::CSharp => "csharp",
        Language::Css => "css",
        Language::JavaScript => "javascript",
        Language::Python => "python",
        Language::TypeScript => "typescript",
        Language::Rust => "rust",
        Language::Go => "go",
        Language::Java => "java",
        Language::Ruby => "ruby",
        Language::Php => "php",
        Language::Swift => "swift",
        Language::Kotlin => "kotlin",
        Language::Scala => "scala",
        Language::Sql => "sql",
        Language::Dart => "dart",
        Language::Lua => "lua",
        Language::Perl => "perl",
        Language::Shell => "shell",
        Language::Groovy => "groovy",
        Language::Elixir => "elixir",
        Language::R => "r",
        Language::Haskell => "haskell",
        Language::Html => "html",
        Language::Svelte => "svelte",
        Language::Vue => "vue",
        Language::Zig => "zig",
        Language::Terraform => "terraform",
        Language::Puppet => "puppet",
        Language::Pulumi => "pulumi",
        Language::Http => "http",
        Language::Plsql => "plsql",
        Language::Apex => "apex",
        Language::Abap => "abap",
        Language::ServiceNow => "servicenow",
        Language::Json => "json",
    }
}

fn match_lang(
    ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    // Get language from file registry - NodeEntry has no language field
    let Some(lang) = ctx.graph.files().language_for_file(entry.file) else {
        return false;
    };

    // Use canonical language names for parity with legacy index
    let actual = language_to_canonical(lang);

    // Support both equality and regex operators
    match (operator, value) {
        (Operator::Equal, Value::String(expected)) => actual == expected,
        (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
            &rv.pattern,
            rv.flags.case_insensitive,
            rv.flags.multiline,
            rv.flags.dot_all,
        )
        .map(|re| regex_is_match(&re, actual))
        .unwrap_or(false),
        _ => false,
    }
}

// ============================================================================
// Visibility predicate
// ============================================================================

fn match_visibility(
    ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    let Some(expected) = value.as_string() else {
        return false;
    };

    let normalized_expected = if expected == "pub" {
        "public"
    } else {
        expected
    };

    let Some(vis_id) = entry.visibility else {
        // No visibility = private by default
        return match operator {
            Operator::Equal => normalized_expected == "private",
            _ => false,
        };
    };

    let Some(actual) = ctx.graph.strings().resolve(vis_id) else {
        return false;
    };
    let normalized_actual = if actual.as_ref().starts_with("pub") {
        "public"
    } else {
        actual.as_ref()
    };

    // Case-sensitive comparison
    match operator {
        Operator::Equal => normalized_actual == normalized_expected,
        _ => false,
    }
}

// ============================================================================
// Returns predicate
// ============================================================================

/// Match `returns:TypeName` predicate.
///
/// Checks the `signature` field on `NodeEntry` for return type matching.
/// Uses substring matching to support both:
/// - Exact: `returns:Optional<User>` matches signature containing "Optional<User>"
/// - Partial: `returns:Optional` matches any signature containing "Optional"
fn match_returns(
    ctx: &GraphEvalContext,
    entry: &NodeEntry,
    operator: &Operator,
    value: &Value,
) -> bool {
    let Some(expected) = value.as_string() else {
        return false;
    };

    // Only functions and methods have return types
    if !matches!(entry.kind, NodeKind::Function | NodeKind::Method) {
        return false;
    }

    let Some(sig_id) = entry.signature else {
        // No signature means no return type info
        return false;
    };

    let Some(signature) = ctx.graph.strings().resolve(sig_id) else {
        return false;
    };

    // The signature typically contains the return type.
    // Use contains for substring matching to support partial matches.
    match operator {
        Operator::Equal => signature.contains(expected),
        _ => false,
    }
}

// ============================================================================
// Boolean predicates (async, static)
// ============================================================================

/// Match `async:true` or `async:false` predicate.
///
/// Checks the `is_async` flag on `NodeEntry`.
/// Handles both Boolean values (from parser) and String values ("true"/"false").
fn match_async(entry: &NodeEntry, operator: &Operator, value: &Value) -> bool {
    let expected = value_to_bool(value);
    let Some(expected) = expected else {
        return false;
    };

    match operator {
        Operator::Equal => entry.is_async == expected,
        _ => false,
    }
}

/// Match `static:true` or `static:false` predicate.
///
/// Checks the `is_static` flag on `NodeEntry`.
/// Handles both Boolean values (from parser) and String values ("true"/"false").
fn match_static(entry: &NodeEntry, operator: &Operator, value: &Value) -> bool {
    let expected = value_to_bool(value);
    let Some(expected) = expected else {
        return false;
    };

    match operator {
        Operator::Equal => entry.is_static == expected,
        _ => false,
    }
}

/// Convert a Value to a boolean.
///
/// Handles:
/// - `Value::Boolean(b)` → `Some(b)`
/// - `Value::String("true"|"yes"|"1")` → `Some(true)`
/// - `Value::String("false"|"no"|"0")` → `Some(false)`
/// - Other values → `None`
fn value_to_bool(value: &Value) -> Option<bool> {
    match value {
        Value::Boolean(b) => Some(*b),
        Value::String(s) => match s.to_lowercase().as_str() {
            "true" | "yes" | "1" => Some(true),
            "false" | "no" | "0" => Some(false),
            _ => None,
        },
        _ => None,
    }
}

// ============================================================================
// Relation predicates (CORRECTED DIRECTION)
// ============================================================================

/// `callers:X` - find symbols that CALL X
///
/// When evaluating node Y: does Y call X? → Check Y's OUTGOING edges
///
/// For dynamically-typed languages (Lua, Python, Ruby, etc.), method calls like
/// `target:method()` create edges to `receiver::method` where `receiver` is the
/// variable name, not the class name. To handle this, when querying for a qualified
/// name like `Class::method`, we also match call edges to `X::method` where X is
/// any receiver, as long as a symbol `Class::method` exists in the graph.
fn match_callers(ctx: &GraphEvalContext, node_id: NodeId, value: &Value) -> bool {
    let Some(target_name) = value.as_string() else {
        return false;
    };

    // Extract the method name part for fallback matching in dynamic languages
    // e.g., "Player::takeDamage" -> "takeDamage"
    let method_part = extract_method_name(target_name);

    // Check OUTGOING Calls edges from this node
    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Calls { .. } = &edge.kind
            && let Some(target_entry) = ctx.graph.nodes().get(edge.target)
        {
            let callee_names = entry_query_texts(ctx.graph, target_entry);

            if callee_names.iter().any(|callee_name| {
                language_aware_segments_match(
                    ctx.graph,
                    target_entry.file,
                    callee_name,
                    target_name,
                )
            }) {
                return true;
            }

            // For dynamic languages: if target_name is qualified (e.g., "Player::takeDamage")
            // and callee is also qualified (e.g., "target::takeDamage"), match on method part
            // This handles cases where the receiver type isn't known at call time
            if let Some(method) = &method_part
                && callee_names
                    .iter()
                    .filter_map(|callee_name| extract_method_name(callee_name))
                    .any(|callee_method| method == &callee_method)
            {
                return true;
            }
        }
    }
    false
}

/// Extract the method name from a qualified name.
/// e.g., "`Player::takeDamage`" -> Some("takeDamage")
/// e.g., "takeDamage" -> None (no separator, not qualified)
#[must_use]
pub fn extract_method_name(qualified: &str) -> Option<String> {
    // Look for separator from the end
    for sep in ["::", ".", "#", ":", "/"] {
        if let Some(pos) = qualified.rfind(sep) {
            let method = &qualified[pos + sep.len()..];
            if !method.is_empty() {
                return Some(method.to_string());
            }
        }
    }
    None
}

/// `callees:X` - find symbols that ARE CALLED BY X
///
/// When evaluating node Y: does X call Y? → Check Y's INCOMING edges for source=X
fn match_callees(ctx: &GraphEvalContext, node_id: NodeId, value: &Value) -> bool {
    let Some(caller_name) = value.as_string() else {
        return false;
    };

    // Check INCOMING Calls edges to this node
    for edge in ctx.graph.edges().edges_to(node_id) {
        if let EdgeKind::Calls { .. } = &edge.kind
            && let Some(source_entry) = ctx.graph.nodes().get(edge.source)
            && entry_query_texts(ctx.graph, source_entry)
                .iter()
                .any(|source_name| {
                    language_aware_segments_match(
                        ctx.graph,
                        source_entry.file,
                        source_name,
                        caller_name,
                    )
                })
        {
            return true;
        }
    }
    false
}

/// `imports:X` — per-node match.
///
/// A node matches iff:
/// 1. It is itself an `Import` node whose name (or alias text) matches `X`, or
/// 2. It has at least one outgoing `Imports` edge whose target name, alias,
///    or wildcard flag matches `X` (see [`import_edge_matches`]).
///
/// This is the planner-canonical semantic per
/// `docs/development/phase-n-structural-semantics/02_DESIGN.md` §6 — every
/// transport (planner, MCP, CLI, LSP) shares it. The previous file-scoped
/// behavior ("every node in a file that imports X matches") was retired in
/// DB15 to remove the cross-engine divergence flagged by Codex's DB14
/// review. Matches the set returned by
/// [`sqry_db::queries::ImportsQuery`](https://docs.rs/sqry-db) for the same
/// key.
fn match_imports(ctx: &GraphEvalContext, node_id: NodeId, value: &Value) -> bool {
    let Some(target_module) = value.as_string() else {
        return false;
    };

    let Some(entry) = ctx.graph.nodes().get(node_id) else {
        return false;
    };

    if entry.kind == NodeKind::Import && import_entry_matches(ctx.graph, entry, target_module) {
        return true;
    }

    for edge in ctx.graph.edges().edges_from(node_id) {
        if import_edge_matches(ctx.graph, &edge, target_module) {
            return true;
        }
    }
    false
}

/// Returns `true` when the given `Imports` edge imports something matching
/// `target_module`, checking the target node text, the edge's alias, and the
/// wildcard flag. Shared with [`crate::graph::unified::concurrent::GraphSnapshot`]
/// consumers (including `sqry-db::queries::relation`) that need the same
/// semantics as the graph-native `imports:` predicate.
#[must_use]
pub fn import_edge_matches<G: crate::graph::unified::concurrent::GraphAccess>(
    graph: &G,
    edge: &StoreEdgeRef,
    target_module: &str,
) -> bool {
    let EdgeKind::Imports { alias, is_wildcard } = &edge.kind else {
        return false;
    };

    // Check target node text across simple, canonical, and native-display forms.
    let target_match = graph
        .nodes()
        .get(edge.target)
        .is_some_and(|entry| import_entry_matches(graph, entry, target_module));

    // Check alias
    let alias_match = alias
        .and_then(|sid| graph.strings().resolve(sid))
        .is_some_and(|alias_str| {
            graph.nodes().get(edge.source).is_some_and(|entry| {
                import_text_matches(graph, entry.file, alias_str.as_ref(), target_module)
            })
        });

    // Check wildcard
    let wildcard_match = *is_wildcard && target_module == "*";

    target_match || alias_match || wildcard_match
}

/// Substring-based text match for `imports:` semantics, with language-aware
/// canonicalization fallback when the file's language maps the input module
/// path into graph-internal `::` form.
#[must_use]
pub fn import_text_matches<G: crate::graph::unified::concurrent::GraphAccess>(
    graph: &G,
    file_id: FileId,
    candidate: &str,
    target_module: &str,
) -> bool {
    if candidate.contains(target_module) {
        return true;
    }

    graph
        .files()
        .language_for_file(file_id)
        .is_some_and(|language| {
            let canonical_target = canonicalize_graph_qualified_name(language, target_module);
            canonical_target != target_module && candidate.contains(&canonical_target)
        })
}

/// Matches an Import/candidate node entry against a target module using
/// the shared `imports:` substring + canonicalization semantics.
#[must_use]
pub fn import_entry_matches<G: crate::graph::unified::concurrent::GraphAccess>(
    graph: &G,
    entry: &NodeEntry,
    target_module: &str,
) -> bool {
    entry_query_texts(graph, entry)
        .iter()
        .any(|candidate| import_text_matches(graph, entry.file, candidate, target_module))
}

/// Segment-aware name equality with a language-specific canonicalization
/// fallback. First tries a direct [`segments_match`]; if that fails, the
/// file's language (if any) is consulted to canonicalize `expected` into
/// graph-internal `::` form before retrying.
#[must_use]
pub fn language_aware_segments_match<G: crate::graph::unified::concurrent::GraphAccess>(
    graph: &G,
    file_id: FileId,
    candidate: &str,
    expected: &str,
) -> bool {
    if segments_match(candidate, expected) {
        return true;
    }

    graph
        .files()
        .language_for_file(file_id)
        .is_some_and(|language| {
            let canonical_expected = canonicalize_graph_qualified_name(language, expected);
            canonical_expected != expected && segments_match(candidate, &canonical_expected)
        })
}

fn push_unique_query_text(texts: &mut Vec<String>, candidate: impl Into<String>) {
    let candidate = candidate.into();
    if !texts.iter().any(|existing| existing == &candidate) {
        texts.push(candidate);
    }
}

/// Collects every string form that can satisfy a name query for the given
/// node entry: the interned name, the qualified name, and (when a language
/// is recorded) the display-qualified name produced by
/// [`display_graph_qualified_name`]. Duplicates are dropped so that relation
/// matchers do not re-check equivalent forms.
#[must_use]
pub fn entry_query_texts<G: crate::graph::unified::concurrent::GraphAccess>(
    graph: &G,
    entry: &NodeEntry,
) -> Vec<String> {
    let mut texts = Vec::with_capacity(3);

    if let Some(name) = graph.strings().resolve(entry.name) {
        push_unique_query_text(&mut texts, name.to_string());
    }

    if let Some(qualified) = entry
        .qualified_name
        .and_then(|qualified_name_id| graph.strings().resolve(qualified_name_id))
    {
        push_unique_query_text(&mut texts, qualified.to_string());

        if let Some(language) = graph.files().language_for_file(entry.file) {
            push_unique_query_text(
                &mut texts,
                display_graph_qualified_name(
                    language,
                    qualified.as_ref(),
                    entry.kind,
                    entry.is_static,
                ),
            );
        }
    }

    texts
}

/// `exports:X` - find symbols that export X
///
/// File scope filtering: Only considers export edges where both source and
/// target are in the same file as the node being evaluated. This prevents
/// re-export edges from causing symbols in other files to match.
fn match_exports(ctx: &GraphEvalContext, node_id: NodeId, value: &Value) -> bool {
    let Some(target_name) = value.as_string() else {
        return false;
    };

    let Some(entry) = ctx.graph.nodes().get(node_id) else {
        return false;
    };
    let node_file = entry.file;

    if !entry_query_texts(ctx.graph, entry).iter().any(|candidate| {
        language_aware_segments_match(ctx.graph, entry.file, candidate, target_name)
    }) {
        return false;
    }

    let edges = ctx.graph.edges();

    // Check OUTGOING export edges: this node exports something
    for edge in edges.edges_from(node_id) {
        if let EdgeKind::Exports { .. } = &edge.kind {
            // File scope filtering: export edge must be in same file
            if let Some(target_entry) = ctx.graph.nodes().get(edge.target)
                && target_entry.file == node_file
            {
                return true;
            }
        }
    }

    // Check INCOMING export edges: something exports this node
    for edge in edges.edges_to(node_id) {
        if let EdgeKind::Exports { .. } = &edge.kind {
            // File scope filtering: export edge source must be in same file
            if let Some(source_entry) = ctx.graph.nodes().get(edge.source)
                && source_entry.file == node_file
            {
                return true;
            }
        }
    }

    false
}

/// `references:X` - find symbols named X that have references TO them.
///
/// Current behavior: references include `Calls`, `Imports`, `FfiCall`, and `References` edges.
fn match_references(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    operator: &Operator,
    value: &Value,
) -> bool {
    // First check if the symbol name matches the target
    let Some(entry) = ctx.graph.nodes().get(node_id) else {
        return false;
    };

    let name_matches = match (operator, value) {
        (Operator::Equal, Value::String(target)) => entry_query_texts(ctx.graph, entry)
            .iter()
            .any(|candidate| candidate == target || candidate.ends_with(&format!("::{target}"))),
        (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
            &rv.pattern,
            rv.flags.case_insensitive,
            rv.flags.multiline,
            rv.flags.dot_all,
        )
        .map(|re| {
            entry_query_texts(ctx.graph, entry)
                .iter()
                .any(|candidate| regex_is_match(&re, candidate))
        })
        .unwrap_or(false),
        _ => false,
    };

    if !name_matches {
        return false;
    }

    // Check if the symbol has references TO it (incoming edges)
    // Current behavior: References, Calls, Imports, AND FfiCall all count as references
    for edge in ctx.graph.edges().edges_to(node_id) {
        let is_reference = matches!(
            &edge.kind,
            EdgeKind::References
                | EdgeKind::Calls { .. }
                | EdgeKind::Imports { .. }
                | EdgeKind::FfiCall { .. }
        );
        if is_reference {
            return true;
        }
    }

    false
}

/// `implements:X` - find symbols that implement interface/trait X
fn match_implements(ctx: &GraphEvalContext, node_id: NodeId, value: &Value) -> bool {
    let Some(trait_name) = value.as_string() else {
        return false;
    };

    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Implements = &edge.kind
            && let Some(target_entry) = ctx.graph.nodes().get(edge.target)
            && entry_query_texts(ctx.graph, target_entry)
                .iter()
                .any(|name| {
                    language_aware_segments_match(ctx.graph, target_entry.file, name, trait_name)
                })
        {
            return true;
        }
    }
    false
}

// ============================================================================
// Scope predicates (with regex support)
// ============================================================================

/// Convert `NodeKind` to scope type string for predicate matching.
///
/// This mapping provides parity with legacy index scope predicates:
/// - Trait → interface
/// - Test → function
/// - Service → class
/// - etc.
fn node_kind_to_scope_type(kind: NodeKind) -> &'static str {
    match kind {
        NodeKind::Function | NodeKind::Test => "function",
        NodeKind::Method => "method",
        NodeKind::Class | NodeKind::Service => "class",
        NodeKind::Interface | NodeKind::Trait => "interface",
        NodeKind::Struct => "struct",
        NodeKind::Enum => "enum",
        NodeKind::Module => "module",
        NodeKind::Macro => "macro",
        NodeKind::Component => "component",
        NodeKind::Resource | NodeKind::Endpoint => "resource",
        // Non-container types
        NodeKind::Variable => "variable",
        NodeKind::Constant => "constant",
        NodeKind::Type => "type",
        NodeKind::EnumVariant => "enumvariant",
        NodeKind::Import => "import",
        NodeKind::Export => "export",
        NodeKind::CallSite => "callsite",
        NodeKind::Parameter => "parameter",
        NodeKind::Property => "property",
        NodeKind::StyleRule => "style_rule",
        NodeKind::StyleAtRule => "style_at_rule",
        NodeKind::StyleVariable => "style_variable",
        NodeKind::Lifetime => "lifetime",
        NodeKind::TypeParameter => "type_parameter",
        NodeKind::Annotation => "annotation",
        NodeKind::AnnotationValue => "annotation_value",
        NodeKind::LambdaTarget => "lambda_target",
        NodeKind::JavaModule => "java_module",
        NodeKind::EnumConstant => "enum_constant",
        NodeKind::Other => "other",
    }
}

fn match_scope(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    field: &str,
    operator: &Operator,
    value: &Value,
) -> bool {
    let scope_part = field.strip_prefix("scope.").unwrap_or("");
    match scope_part {
        "type" => match_scope_type(ctx, node_id, operator, value),
        "name" => match_scope_name(ctx, node_id, operator, value),
        "parent" => match_scope_parent_name(ctx, node_id, operator, value),
        "ancestor" => match_scope_ancestor_name(ctx, node_id, operator, value),
        _ => false,
    }
}

fn match_scope_type(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    operator: &Operator,
    value: &Value,
) -> bool {
    for edge in ctx.graph.edges().edges_to(node_id) {
        if let EdgeKind::Contains = &edge.kind
            && let Some(parent) = ctx.graph.nodes().get(edge.source)
        {
            // Use mapped scope type for parity with legacy index
            let scope_type = node_kind_to_scope_type(parent.kind);
            return match (operator, value) {
                // Case-sensitive comparison
                (Operator::Equal, Value::String(exp)) => scope_type == exp,
                (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
                    &rv.pattern,
                    rv.flags.case_insensitive,
                    rv.flags.multiline,
                    rv.flags.dot_all,
                )
                .map(|re| regex_is_match(&re, scope_type))
                .unwrap_or(false),
                _ => false,
            };
        }
    }
    false
}

fn match_scope_name(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    operator: &Operator,
    value: &Value,
) -> bool {
    for edge in ctx.graph.edges().edges_to(node_id) {
        if let EdgeKind::Contains = &edge.kind
            && let Some(parent) = ctx.graph.nodes().get(edge.source)
            && let Some(name) = ctx.graph.strings().resolve(parent.name)
        {
            return match (operator, value) {
                // Use segments_match for qualified name suffix matching
                (Operator::Equal, Value::String(exp)) => segments_match(&name, exp),
                (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
                    &rv.pattern,
                    rv.flags.case_insensitive,
                    rv.flags.multiline,
                    rv.flags.dot_all,
                )
                .map(|re| regex_is_match(&re, &name))
                .unwrap_or(false),
                _ => false,
            };
        }
    }
    false
}

/// `scope.parent:X` - find nodes with immediate parent NAMED X.
///
/// Uses `segments_match` for qualified name suffix matching (same as name: predicate).
fn match_scope_parent_name(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    operator: &Operator,
    value: &Value,
) -> bool {
    for edge in ctx.graph.edges().edges_to(node_id) {
        if let EdgeKind::Contains = &edge.kind
            && let Some(parent) = ctx.graph.nodes().get(edge.source)
            && let Some(name) = ctx.graph.strings().resolve(parent.name)
        {
            return match (operator, value) {
                // Use segments_match for qualified name suffix matching
                (Operator::Equal, Value::String(exp)) => segments_match(&name, exp),
                (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
                    &rv.pattern,
                    rv.flags.case_insensitive,
                    rv.flags.multiline,
                    rv.flags.dot_all,
                )
                .map(|re| regex_is_match(&re, &name))
                .unwrap_or(false),
                _ => false,
            };
        }
    }
    false
}

/// `scope.ancestor:X` - find nodes with any ancestor NAMED X.
///
/// CYCLE PROTECTION: Uses visited set to prevent infinite loops.
/// Uses `segments_match` for qualified name suffix matching.
fn match_scope_ancestor_name(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    operator: &Operator,
    value: &Value,
) -> bool {
    let mut current = node_id;
    let mut visited = HashSet::new();
    visited.insert(node_id);

    loop {
        let mut found_parent = false;
        for edge in ctx.graph.edges().edges_to(current) {
            if let EdgeKind::Contains = &edge.kind {
                // CYCLE PROTECTION: Skip if we've already visited this node
                if visited.contains(&edge.source) {
                    continue;
                }
                visited.insert(edge.source);

                found_parent = true;
                current = edge.source;
                if let Some(parent) = ctx.graph.nodes().get(current)
                    && let Some(name) = ctx.graph.strings().resolve(parent.name)
                {
                    let matches = match (operator, value) {
                        // Use segments_match for qualified name suffix matching
                        (Operator::Equal, Value::String(exp)) => segments_match(&name, exp),
                        (Operator::Regex, Value::Regex(rv)) => get_or_compile_regex(
                            &rv.pattern,
                            rv.flags.case_insensitive,
                            rv.flags.multiline,
                            rv.flags.dot_all,
                        )
                        .map(|re| regex_is_match(&re, &name))
                        .unwrap_or(false),
                        _ => false,
                    };
                    if matches {
                        return true;
                    }
                }
                break;
            }
        }
        if !found_parent {
            break;
        }
    }
    false
}

// ============================================================================
// Subquery evaluation
// ============================================================================

/// Evaluate an expression against all nodes and return the set of matching node IDs.
///
/// Used for subquery evaluation: `callers:(kind:function AND async:true)`.
///
/// # Errors
///
/// Returns an error if predicate evaluation fails.
pub fn evaluate_subquery(ctx: &GraphEvalContext, expr: &Expr) -> Result<HashSet<NodeId>> {
    let recursion_limits = crate::config::RecursionLimits::load_or_default()?;
    let expr_depth = recursion_limits.effective_expr_depth()?;
    let mut guard = crate::query::security::RecursionGuard::new(expr_depth)?;

    let arena = ctx.graph.nodes();
    let mut matches = HashSet::new();
    for (id, _) in arena.iter() {
        if evaluate_node(ctx, id, expr, &mut guard)? {
            matches.insert(id);
        }
    }
    Ok(matches)
}

/// Match callers using a precomputed subquery result set.
///
/// Checks if any of this node's outgoing Calls edges target a node in the
/// precomputed set. Returns an error if the subquery was not precomputed
/// (indicates a bug in `precompute_subqueries`).
fn match_callers_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Calls { .. } = &edge.kind
            && matches.contains(&edge.target)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Match callees using a precomputed subquery result set.
fn match_callees_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_to(node_id) {
        if let EdgeKind::Calls { .. } = &edge.kind
            && matches.contains(&edge.source)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Match imports using a precomputed subquery result set.
///
/// Per-node semantic (DB15): returns true iff this specific node has an
/// outgoing `Imports` edge whose target is in the precomputed subquery set.
/// Aligned with [`match_imports`] and `sqry_db::queries::ImportsQuery`.
fn match_imports_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Imports { .. } = &edge.kind
            && matches.contains(&edge.target)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Match exports using a precomputed subquery result set.
fn match_exports_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Exports { .. } = &edge.kind
            && matches.contains(&edge.target)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Match implements using a precomputed subquery result set.
fn match_implements_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_from(node_id) {
        if let EdgeKind::Implements = &edge.kind
            && matches.contains(&edge.target)
        {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Match references using a precomputed subquery result set.
fn match_references_subquery(
    ctx: &GraphEvalContext,
    node_id: NodeId,
    subquery_matches: Option<&HashSet<NodeId>>,
) -> Result<bool> {
    let Some(matches) = subquery_matches else {
        return Err(anyhow!(
            "subquery cache miss: precompute_subqueries did not populate cache for this relation predicate"
        ));
    };
    for edge in ctx.graph.edges().edges_to(node_id) {
        let is_reference = matches!(
            &edge.kind,
            EdgeKind::References
                | EdgeKind::Calls { .. }
                | EdgeKind::Imports { .. }
                | EdgeKind::FfiCall { .. }
        );
        if is_reference && matches.contains(&edge.source) {
            return Ok(true);
        }
    }
    Ok(false)
}

// ============================================================================
// Join evaluation
// ============================================================================

/// Evaluate a join expression, returning matched (left, right) node pairs.
///
/// For `(kind:function AND lang:rust) CALLS (kind:function AND lang:python)`:
/// 1. Evaluate LHS query → set of matching nodes
/// 2. Evaluate RHS query → set of matching nodes
/// 3. For each LHS node, check edges of the specified kind
/// 4. If edge target is in RHS set → add (lhs, rhs) pair
///
/// # Errors
///
/// Returns an error if subquery evaluation or edge traversal fails.
pub fn evaluate_join(
    ctx: &GraphEvalContext,
    join: &JoinExpr,
    max_results: Option<usize>,
) -> Result<JoinEvalResult> {
    let lhs_matches = evaluate_subquery(ctx, &join.left)?;
    let rhs_matches = evaluate_subquery(ctx, &join.right)?;
    let cap = max_results.unwrap_or(DEFAULT_JOIN_RESULT_CAP);

    let mut pairs = Vec::new();
    let mut truncated = false;
    'outer: for &lhs_id in &lhs_matches {
        for edge in ctx.graph.edges().edges_from(lhs_id) {
            if edge_matches_join_kind(&edge.kind, &join.edge) && rhs_matches.contains(&edge.target)
            {
                pairs.push((lhs_id, edge.target));
                if pairs.len() >= cap {
                    truncated = true;
                    break 'outer;
                }
            }
        }
    }
    Ok(JoinEvalResult { pairs, truncated })
}

/// Result of a join evaluation including truncation metadata.
pub struct JoinEvalResult {
    /// The matched (source, target) node ID pairs.
    pub pairs: Vec<(NodeId, NodeId)>,
    /// Whether the result set was truncated by the result cap.
    pub truncated: bool,
}

/// Default result cap for join queries.
///
/// Prevents unbounded memory growth when a join produces a large number of matching pairs.
const DEFAULT_JOIN_RESULT_CAP: usize = 10_000;

/// Check if an edge kind matches a join edge kind.
fn edge_matches_join_kind(edge_kind: &EdgeKind, join_kind: &JoinEdgeKind) -> bool {
    match join_kind {
        JoinEdgeKind::Calls => matches!(edge_kind, EdgeKind::Calls { .. }),
        JoinEdgeKind::Imports => matches!(edge_kind, EdgeKind::Imports { .. }),
        JoinEdgeKind::Inherits => matches!(edge_kind, EdgeKind::Inherits),
        JoinEdgeKind::Implements => matches!(edge_kind, EdgeKind::Implements),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graph::node::Language;
    use crate::query::types::{Condition, Field, Span};
    use std::path::Path;

    #[test]
    fn test_import_text_matches_canonicalized_qualified_imports() {
        let mut graph = CodeGraph::new();
        let file_id = graph
            .files_mut()
            .register(Path::new("src/FileProcessor.cs"))
            .unwrap();
        assert!(graph.files_mut().set_language(file_id, Language::CSharp));

        assert!(import_text_matches(
            &graph,
            file_id,
            "System::IO",
            "System.IO"
        ));
        assert!(import_text_matches(
            &graph,
            file_id,
            "System::Collections::Generic",
            "System.Collections.Generic"
        ));
        assert!(!import_text_matches(
            &graph,
            file_id,
            "System::Text",
            "System.IO"
        ));
    }

    #[test]
    fn test_language_aware_segments_match_supports_ruby_method_separators() {
        let mut graph = CodeGraph::new();
        let file_id = graph
            .files_mut()
            .register(Path::new("app/models/user.rb"))
            .unwrap();
        assert!(graph.files_mut().set_language(file_id, Language::Ruby));

        assert!(language_aware_segments_match(
            &graph,
            file_id,
            "Admin::Users::Controller::show",
            "Admin::Users::Controller#show"
        ));
        assert!(language_aware_segments_match(
            &graph,
            file_id,
            "Admin::Users::Controller::show",
            "show"
        ));
        assert!(!language_aware_segments_match(
            &graph,
            file_id,
            "Admin::Users::Controller::index",
            "Admin::Users::Controller#show"
        ));
    }

    #[test]
    fn test_normalize_kind() {
        // Case-sensitive synonyms
        assert_eq!(normalize_kind("trait"), "interface");
        assert_eq!(normalize_kind("TRAIT"), "TRAIT"); // Case-sensitive: not a synonym
        assert_eq!(normalize_kind("field"), "property");
        assert_eq!(normalize_kind("namespace"), "module");
        assert_eq!(normalize_kind("function"), "function"); // unchanged
    }

    #[test]
    fn test_graph_eval_context_builder() {
        let graph = CodeGraph::new();
        let pm = PluginManager::new();
        let ctx = GraphEvalContext::new(&graph, &pm)
            .with_workspace_root(Path::new("/test"))
            .with_parallel_disabled(true);

        assert!(ctx.disable_parallel);
        assert_eq!(ctx.workspace_root, Some(Path::new("/test")));
    }

    // ================================================================
    // collect_subquery_exprs tests
    // ================================================================

    /// Helper: build `field:(inner_expr)` as a Condition with `Value::Subquery`.
    fn subquery_condition(field: &str, inner: Expr, start: usize, end: usize) -> Expr {
        Expr::Condition(Condition {
            field: Field(field.to_string()),
            operator: Operator::Equal,
            value: Value::Subquery(Box::new(inner)),
            span: Span::with_position(start, end, 1, start + 1),
        })
    }

    /// Helper: build a simple `kind:function` condition.
    fn kind_condition(kind: &str) -> Expr {
        Expr::Condition(Condition {
            field: Field("kind".to_string()),
            operator: Operator::Equal,
            value: Value::String(kind.to_string()),
            span: Span::default(),
        })
    }

    #[test]
    fn test_collect_subquery_exprs_post_order_depth_2() {
        // Build: callers:(callees:(kind:function))
        // Inner subquery: callees:(kind:function) at span (20, 40)
        // Outer subquery: callers:(...) at span (0, 50)
        let inner_subquery = subquery_condition("callees", kind_condition("function"), 20, 40);
        let outer_subquery = subquery_condition("callers", inner_subquery, 0, 50);

        let mut out = Vec::new();
        collect_subquery_exprs(&outer_subquery, &mut out);

        // Post-order: inner appears before outer
        assert_eq!(
            out.len(),
            2,
            "should collect both inner and outer subqueries"
        );
        assert_eq!(out[0].0, (20, 40), "inner subquery span should come first");
        assert_eq!(out[1].0, (0, 50), "outer subquery span should come second");
    }

    #[test]
    fn test_collect_subquery_exprs_post_order_depth_3() {
        // Build: callers:(callees:(imports:(kind:function)))
        let innermost = subquery_condition("imports", kind_condition("function"), 30, 50);
        let middle = subquery_condition("callees", innermost, 15, 55);
        let outer = subquery_condition("callers", middle, 0, 60);

        let mut out = Vec::new();
        collect_subquery_exprs(&outer, &mut out);

        assert_eq!(out.len(), 3, "should collect all three nested subqueries");
        assert_eq!(out[0].0, (30, 50), "innermost should come first");
        assert_eq!(out[1].0, (15, 55), "middle should come second");
        assert_eq!(out[2].0, (0, 60), "outer should come last");
    }

    #[test]
    fn test_collect_subquery_exprs_and_or_branches() {
        // Build: callers:(kind:function) AND callees:(kind:method)
        let left = subquery_condition("callers", kind_condition("function"), 0, 25);
        let right = subquery_condition("callees", kind_condition("method"), 30, 55);
        let expr = Expr::And(vec![left, right]);

        let mut out = Vec::new();
        collect_subquery_exprs(&expr, &mut out);

        assert_eq!(out.len(), 2, "should collect subqueries from both branches");
        assert_eq!(out[0].0, (0, 25), "left branch subquery");
        assert_eq!(out[1].0, (30, 55), "right branch subquery");
    }

    #[test]
    fn test_collect_subquery_exprs_no_subqueries() {
        // Simple condition with no subqueries: kind:function
        let expr = kind_condition("function");

        let mut out = Vec::new();
        collect_subquery_exprs(&expr, &mut out);

        assert!(
            out.is_empty(),
            "should collect nothing for plain conditions"
        );
    }

    // ================================================================
    // FfiCall edge in references/referenced_by predicates
    // ================================================================

    use crate::graph::unified::edge::{BidirectionalEdgeStore, FfiConvention};
    use crate::graph::unified::storage::{
        AuxiliaryIndices, FileRegistry, NodeArena, StringInterner,
    };

    /// Build a `CodeGraph` with: `caller --FfiCall(C)--> target`
    fn build_ffi_graph() -> (CodeGraph, NodeId, NodeId) {
        let mut arena = NodeArena::new();
        let edges = BidirectionalEdgeStore::new();
        let mut strings = StringInterner::new();
        let mut files = FileRegistry::new();
        let mut indices = AuxiliaryIndices::new();

        let caller_name = strings.intern("caller_fn").unwrap();
        let target_name = strings.intern("ffi_target").unwrap();
        let file_id = files.register(Path::new("test.r")).unwrap();

        let caller_id = arena
            .alloc(NodeEntry {
                kind: NodeKind::Function,
                name: caller_name,
                file: file_id,
                start_byte: 0,
                end_byte: 100,
                start_line: 1,
                start_column: 0,
                end_line: 5,
                end_column: 0,
                signature: None,
                doc: None,
                qualified_name: None,
                visibility: None,
                is_async: false,
                is_static: false,
                is_unsafe: false,
                body_hash: None,
            })
            .unwrap();

        let target_id = arena
            .alloc(NodeEntry {
                kind: NodeKind::Function,
                name: target_name,
                file: file_id,
                start_byte: 200,
                end_byte: 300,
                start_line: 10,
                start_column: 0,
                end_line: 15,
                end_column: 0,
                signature: None,
                doc: None,
                qualified_name: None,
                visibility: None,
                is_async: false,
                is_static: false,
                is_unsafe: false,
                body_hash: None,
            })
            .unwrap();

        indices.add(caller_id, NodeKind::Function, caller_name, None, file_id);
        indices.add(target_id, NodeKind::Function, target_name, None, file_id);

        edges.add_edge(
            caller_id,
            target_id,
            EdgeKind::FfiCall {
                convention: FfiConvention::C,
            },
            file_id,
        );

        let graph = CodeGraph::from_components(
            arena,
            edges,
            strings,
            files,
            indices,
            crate::graph::unified::NodeMetadataStore::new(),
        );
        (graph, caller_id, target_id)
    }

    #[test]
    fn test_ffi_call_edge_in_references_predicate() {
        let (graph, _caller_id, target_id) = build_ffi_graph();
        let pm = PluginManager::new();
        let ctx = GraphEvalContext::new(&graph, &pm);

        // `references:ffi_target` should match because there is an FfiCall edge to it
        let result = match_references(
            &ctx,
            target_id,
            &Operator::Equal,
            &Value::String("ffi_target".to_string()),
        );
        assert!(result, "references: predicate should match FfiCall edges");
    }

    #[test]
    fn test_ffi_call_edge_in_references_subquery() {
        let (graph, caller_id, target_id) = build_ffi_graph();
        let pm = PluginManager::new();
        let ctx = GraphEvalContext::new(&graph, &pm);

        // Simulate a subquery that matched the caller node
        let mut subquery_results = HashSet::new();
        subquery_results.insert(caller_id);

        // match_references_subquery checks incoming edges to target_id
        // and verifies the source is in the subquery result set
        let result = match_references_subquery(&ctx, target_id, Some(&subquery_results)).unwrap();
        assert!(
            result,
            "references subquery should match FfiCall edge sources"
        );
    }
}