lunaris_retrieve/operators/

graph.rs

//! `Graph::anchored(entity_ids, hops)` — graph-anchored retrieval (RETRIEVE-03).
//!
//! Per blueprint §8 + ROADMAP Phase 3 success criterion #3, the operator starts
//! from a list of pre-resolved [`EntityId`]s (extracted from the query text by
//! the planner per D-13), traverses the graph via Cypher BFS for `hops` steps,
//! and returns reachable entity ids as `RawHit { source_op: SourceOp::Graph }`.
//!
//! ## Capability-driven Cypher dialect dispatch (Wave 4 amendment)
//!
//! A parallel-agent probe found that Moon does not accept the full Wave-4
//! template. The operator picks the Cypher template at `retrieve()`-time
//! from [`StorageCapabilities::cypher_dialect`](lunaris_core::CypherDialect):
//!
//! | Tier | Backends | Template shape |
//! |------|----------|----------------|
//! | [`CypherDialect::Legacy`] | Moon | `MATCH (n)-[*1..N]-(m) RETURN id/name/type` |
//! | [`CypherDialect::Full`] | (forward-compat) | `MATCH p = ...` + `length(p)` + `source_entity_id` + `reduce(...) AS edge_weight_product` |
//!
//! Primary-source rejection that motivated the split: **Moon** rejects
//! `MATCH p = (n)-[*1..N]-(m)` outside `shortestPath()`
//! (`vendor/moon/src/graph/cypher/parser/pattern.rs:172-216`) and its
//! function table omits `length()`, `reduce()`, `relationships()`
//! (`executor/eval.rs:116-227`).
//!
//! 0.7.0 dropped the intermediate `PathMetrics` tier along with the
//! Postgres backend (Apache AGE 1.5) that was its only producer.
//!
//! W-7 fix: the property name MUST be `id_hex` because Plan 03-03's ingest
//! fan-out writes `WriteOp::GraphNode { props: { "id_hex": format!("{}",
//! entity_id), ... } }`. Earlier drafts used `id`, which silently returned
//! zero rows. Both templates use `id_hex`.
//!
//! Score formula:
//!   `score = (edge_weight_product / (1.0 + path_length)) * anchor_confidence`
//!
//! `anchor_confidence` is NEVER a Cypher column — the operator synthesizes
//! it post-Cypher by joining the (optional) `source_entity_id` column
//! against the planner's per-seed confidence map (built from
//! `Graph::anchored`'s `Vec<(EntityId, f32)>` argument).
//!
//! Back-compat: when a column the dialect promises is absent from the
//! response (e.g., a degraded backend that answers with the legacy
//! three-column shape), the header-keyed parser falls
//! back to defaults (`path_length = i`, `edge_weight_product = 1.0`,
//! `anchor_confidence = 1.0`) — the score reduces to the Wave-3 legacy
//! `1.0 / (1.0 + i)`. The `graph_score_back_compat_when_no_new_headers`
//! test plus the `dispatch_*_preserves_back_compat_score*` tests pin this
//! property across every tier.
//!
//! `<hops>` is spliced into the cypher string at construction time — the
//! openCypher spec requires variable-length pattern bounds (`[*lo..hi]`) to
//! be literals; both Moon GRAPH.QUERY and Apache AGE refuse `[*1..$hops]`.
//! `$ids` and `$k` are typed parameters so EntityIds NEVER reach the cypher
//! string text — defends T-03-02-01 (Cypher injection) by mirroring the
//! Plan 02-02 plainto_tsquery parameter-binding pattern.
//!
//! ## Defense layers (T-03-02-02 DoS mitigation)
//!
//! - `MAX_GRAPH_HOPS = 5` — hard cap on hops per D-14 (caller-supplied `hops
//!   greater than 5` clamps to 5; `hops == 0` clamps up to 1 so the BFS
//!   always runs at least one edge step).
//! - [`super::clamp_k`] caps the result count at `MAX_K = 1000` (Plan 02-02
//!   convention).
//! - Phase 4 OPS-05 will wire `tower::timeout` around the Retriever for
//!   callers that need a wall-clock cap — bounded fan-out per query.

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use lunaris_core::storage::types::GraphDecay;
use lunaris_core::{CypherDialect, CypherQuery, LunarisError};
use lunaris_extract::EntityId;

use super::{QueryContext, Retriever, clamp_k};
use crate::types::{RawHit, SourceOp};

/// Default hops per ROADMAP Phase 3 success criterion #3.
pub const DEFAULT_GRAPH_HOPS: usize = 2;
/// Hard cap on hops to prevent runaway BFS (D-14). Caller-supplied
/// `hops greater than 5` clamps to 5; `hops == 0` clamps up to 1 so the BFS
/// always runs at least one edge step (zero-hop traversal would only return
/// the anchor entities themselves, which the caller already has).
pub const MAX_GRAPH_HOPS: usize = 5;
/// Default candidate count for graph traversal (matches Vector/Keyword default of 30).
pub const DEFAULT_GRAPH_K: usize = 30;
/// Graph name Lunaris traverses — the `<graph>` argument of Moon's
/// `GRAPH.QUERY <graph> "..."`. Held as a constant (rather than inlined at
/// each call site) so a future backend can be pointed at the same graph
/// identifier and keep the Cypher template portable; `with_graph()`
/// overrides it per tenant.
pub const LUNARIS_GRAPH_NAME: &str = "lunaris_graph";

/// Graph-anchored retrieval operator.
///
/// Construction is fluent — wire at `recall()`-build time, the network call
/// only happens at `.retrieve()` time. The `entity_ids` come from the
/// RETRIEVE-13 planner stub (extracted from query text) per D-13. Empty
/// `entity_ids` short-circuits to an empty result set without touching
/// storage (the planner returns no entities when the query has no entity
/// mentions; treat that as "graph branch contributes nothing" rather than an
/// error).
#[derive(Clone, Debug)]
#[must_use = "Graph is a query node — pass it to RetrievalBuilder::with_root() or chain via .and/.or/.fuse_rrf, otherwise it never executes"]
pub struct Graph {
    /// Anchor seeds with per-seed confidence in `[0.0, 1.0]`. The operator
    /// snapshots a `HashMap<EntityId, f32>` from this vector at construction
    /// for O(1) confidence lookup after Cypher returns.
    pub seeds: Vec<(EntityId, f32)>,
    pub hops: usize,
    pub k: usize,
    pub graph: String,
    /// Per-seed confidence keyed by `EntityId` for O(1) post-Cypher lookup
    /// against the `source_entity_id` column. Built from `seeds` at
    /// construction time. On duplicate seed keys the MAX confidence wins
    /// (safest default for caller mistakes — a low-confidence duplicate
    /// must never demote a high-confidence anchor). Values are clamped to
    /// `[0.0, 1.0]` to prevent a caller-passed out-of-range value from
    /// poisoning the score downstream.
    pub(crate) confidence_by_seed: HashMap<EntityId, f32>,
    /// Optional recency decay (ADD task `ft-navigate-recall`): threads a λ
    /// through `StoragePort::graph_traverse_decayed` so edge cost becomes
    /// `|weight| + λ·w·age_seconds` on backends with native decay (Moon).
    /// `None` keeps the exact pre-decay traversal (delegation guarantee of
    /// the graph-decay-recency contract).
    pub(crate) decay: Option<GraphDecay>,
}

impl Graph {
    /// `Graph::anchored(seeds, hops)` — primary constructor per D-13.
    ///
    /// `seeds` carries `(EntityId, confidence)` pairs where `confidence` is
    /// the planner's belief that the entity is actually relevant to the
    /// query. Values are clamped to `[0.0, 1.0]`. Duplicates collapse to
    /// MAX (safe default — never demote a high-confidence anchor by passing
    /// a low-confidence duplicate).
    ///
    /// `hops` is clamped to `[1, MAX_GRAPH_HOPS]` (D-14):
    /// - `hops` greater than 5 — clamped to 5 (DoS defense; bounded BFS fan-out)
    /// - `hops == 0` — clamped up to 1 (zero-hop traversal would only return
    ///   the anchor entities, which the caller already has)
    ///
    /// `k` defaults to [`DEFAULT_GRAPH_K`] (matches Vector/Keyword default of
    /// 30) and is clamped via [`super::clamp_k`].
    pub fn anchored(seeds: Vec<(EntityId, f32)>, hops: usize) -> Self {
        let mut confidence_by_seed: HashMap<EntityId, f32> = HashMap::with_capacity(seeds.len());
        for (id, conf) in &seeds {
            let clamped = conf.clamp(0.0, 1.0);
            confidence_by_seed
                .entry(*id)
                .and_modify(|prev| {
                    if clamped > *prev {
                        *prev = clamped;
                    }
                })
                .or_insert(clamped);
        }
        Self {
            seeds,
            hops: hops.clamp(1, MAX_GRAPH_HOPS),
            k: clamp_k(DEFAULT_GRAPH_K),
            graph: LUNARIS_GRAPH_NAME.into(),
            confidence_by_seed,
            decay: None,
        }
    }

    /// Apply recency decay to the traversal (ADD task `ft-navigate-recall`).
    /// The λ rides `StoragePort::graph_traverse_decayed`; backends without
    /// `capabilities().graph_decay_native` surface `NotSupported` — gate on
    /// the capability before composing decay into a recall.
    pub fn with_decay(mut self, decay: GraphDecay) -> Self {
        self.decay = Some(decay);
        self
    }

    /// Convenience: list of seed [`EntityId`]s (without confidence). Mirrors
    /// the pre-Wave-4 `entity_ids` field for callers that only need the keys.
    pub fn entity_ids(&self) -> Vec<EntityId> {
        self.seeds.iter().map(|(id, _)| *id).collect()
    }

    /// Override the candidate count. Clamped at [`super::MAX_K`].
    pub fn with_k(mut self, k: usize) -> Self {
        self.k = clamp_k(k);
        self
    }

    /// Override the graph name (defaults to [`LUNARIS_GRAPH_NAME`]). Useful
    /// for tenant-isolated deployments that scope each tenant to its own
    /// AGE / Moon graph.
    pub fn with_graph(mut self, graph: impl Into<String>) -> Self {
        self.graph = graph.into();
        self
    }

    // Mirror vector.rs lines 37-67 exactly — same chain method shape so the
    // canonical compose example
    // `Vector::new(...).and(Graph::anchored(...)).fuse_rrf(60).top(5)` works
    // regardless of operator order.

    /// Concurrent fan-out — runs THIS graph branch alongside `other` and
    /// concatenates results (per-source ranking preserved via
    /// [`SourceOp`]). Downstream `fuse_rrf` groups by `source_op` to fold
    /// per-branch rankings.
    pub fn and<R: Retriever + 'static>(self, other: R) -> super::combinators::AndRetriever {
        super::combinators::AndRetriever::new(Box::new(self), Box::new(other))
    }
    /// Concurrent fan-out — runs THIS graph branch alongside `other` and
    /// unions results by id (max score wins on duplicate id).
    pub fn or<R: Retriever + 'static>(self, other: R) -> super::combinators::OrRetriever {
        super::combinators::OrRetriever::new(Box::new(self), Box::new(other))
    }
    /// Sequential narrow — runs THIS graph branch first, then passes its hit
    /// ids as a `Filter::Or(Filter::Eq{id, …})` to `other`.
    pub fn then<R: Retriever + 'static>(self, other: R) -> super::combinators::ThenRetriever {
        super::combinators::ThenRetriever::new(Box::new(self), Box::new(other))
    }
    /// Cap the final result set after the graph traversal resolves.
    pub fn top(self, n: usize) -> super::modifiers::TopRetriever {
        super::modifiers::TopRetriever::new(Box::new(self), n)
    }

    /// Wrap with a cross-encoder rerank pass (Plan 02-03).
    pub fn rerank(
        self,
        reranker: Arc<dyn lunaris_rerank::Reranker>,
    ) -> super::rerank::RerankRetriever {
        super::rerank::RerankRetriever::new(Box::new(self), reranker)
    }

    /// Wrap with a fallback retriever — if THIS graph path errors (e.g., a
    /// backend without graph support returns `NotSupported`, or Moon
    /// GRAPH.QUERY hits a transient disconnect), switch to `fallback` and
    /// tag returned hits with `degraded: true` (Plan 02-03).
    pub fn degraded_fallback<R: Retriever + 'static>(
        self,
        fallback: R,
    ) -> super::degraded::DegradedFallbackRetriever {
        super::degraded::DegradedFallbackRetriever::new(Box::new(self), Box::new(fallback))
    }

    /// Build the parameterized [`CypherQuery`] for this operator using the
    /// requested [`CypherDialect`] tier.
    ///
    /// `hops` is spliced into the cypher string literal because openCypher
    /// requires variable-length pattern bounds to be literals (both Moon and
    /// AGE refuse `[*1..$hops]`). The literal is bounded by `MAX_GRAPH_HOPS`
    /// at construction time so a caller cannot inject `[*1..1000000]`.
    ///
    /// `$ids` carries the EntityIds as hex strings — they are NEVER spliced
    /// into the cypher text (T-03-02-01 mitigation). `$k` carries the row
    /// limit. This pattern mirrors the Plan 02-02 `plainto_tsquery($1)`
    /// parameter-binding mitigation for tsquery DSL injection.
    ///
    /// Wave 4 amendment — dialect tiers:
    /// - [`CypherDialect::Legacy`] — Moon-compatible. No path binding, no
    ///   `length()`, no `reduce()`, no `source_entity_id`. The header-keyed
    ///   parser falls back to the Wave-3 `1/(1+i)` score formula.
    /// - [`CypherDialect::Full`] — forward-compat. `MATCH p = ...` +
    ///   `length(p) AS path_length` + `n.id_hex AS source_entity_id` +
    ///   `reduce(w=1.0, r in relationships(p) | ...) AS
    ///   edge_weight_product`. No current backend accepts this.
    ///
    /// W-7 fix: MATCH and RETURN MUST use `id_hex` to align with Plan 03-03's
    /// GraphNode props which writes `id_hex`. Using `id` would silently
    /// return zero rows because the node property is `id_hex` not `id`.
    fn build_cypher(&self, dialect: CypherDialect) -> CypherQuery {
        let cypher = match dialect {
            CypherDialect::Legacy => format!(
                "UNWIND $ids AS sid \
                 MATCH (n)-[*1..{hops}]-(m) WHERE n.id_hex = sid \
                 RETURN \
                   m.id_hex AS id, \
                   m.name AS name, \
                   m.type AS type \
                 LIMIT $k",
                hops = self.hops,
            ),
            CypherDialect::Full => format!(
                "UNWIND $ids AS sid \
                 MATCH p = (n)-[*1..{hops}]-(m) WHERE n.id_hex = sid \
                 RETURN \
                   m.id_hex AS id, \
                   m.name AS name, \
                   m.type AS type, \
                   length(p) AS path_length, \
                   reduce(w=1.0, r in relationships(p) | w * coalesce(r.weight, 1.0)) AS edge_weight_product, \
                   n.id_hex AS source_entity_id \
                 LIMIT $k",
                hops = self.hops,
            ),
        };
        let id_strs: Vec<serde_json::Value> = self
            .seeds
            .iter()
            // EntityId Display = hex (16 bytes → 32 hex chars). Stable across
            // calls — see lunaris_extract::types::EntityId::Display impl.
            .map(|(id, _)| serde_json::Value::String(format!("{}", id)))
            .collect();
        let mut params = serde_json::Map::new();
        params.insert("ids".into(), serde_json::Value::Array(id_strs));
        params.insert("k".into(), serde_json::Value::Number(self.k.into()));
        CypherQuery { graph: self.graph.clone(), cypher, params }
    }
}

#[async_trait]
impl Retriever for Graph {
    async fn retrieve(&self, ctx: &QueryContext) -> Result<Vec<RawHit>, LunarisError> {
        if self.seeds.is_empty() {
            // No anchor entities → empty result. The planner returns no
            // entity_ids when the query has no entity mentions; treat that
            // as "graph branch contributes nothing" rather than an error.
            // Importantly, we do NOT call ctx.storage.graph_traverse() — the
            // empty case must NOT trigger a backend round trip.
            return Ok(Vec::new());
        }
        // Wave 4 amendment: read the backend's declared dialect tier and
        // build the matching Cypher template. Moon stays at Legacy (its
        // parser rejects `MATCH p = ...` outside shortestPath()); `Full`
        // is reserved for a future backend that accepts path binding +
        // reduce(). The capability is read ONCE here, not threaded through
        // every operator method, so the dispatch point is single and
        // testable. See `dispatch_*` integration tests in graph_anchored.rs.
        let dialect = ctx.storage.capabilities().cypher_dialect;
        let q = self.build_cypher(dialect);
        // Wave 2.5C: use ctx.scope — plumbed from RetrievalBuilder::with_scope
        // (set by ScopedLunaris::recall/dsl) so graph_traverse is scope-isolated
        // at the storage layer. Bare Lunaris::recall() uses Scope::dev().
        // ft-navigate-recall: route through graph_traverse_decayed — decay
        // None delegates byte-for-byte to graph_traverse (graph-decay-recency
        // contract), so the no-decay path is behaviorally unchanged.
        let result = ctx
            .storage
            .graph_traverse_decayed(&ctx.scope, &q, ctx.query.as_of, self.decay.as_ref())
            .await?;

        // P0 #4 Wave 3 + Wave 4 — Real graph scoring.
        //
        // Score formula:
        //     score = (edge_weight_product / (1.0 + path_length)) * anchor_confidence
        //
        // - `path_length` and `edge_weight_product` are read by HEADER NAME
        //   from GraphResult (additive: backends that haven't been upgraded
        //   to Wave 4 omit the headers; the operator falls back to
        //   `i / 1.0` so the score reduces to the pre-Wave-3 legacy
        //   `1.0 / (1.0 + i)`).
        // - `source_entity_id` (Wave 4) is also header-keyed. When present,
        //   the operator decodes it back to an [`EntityId`] and looks up
        //   `confidence_by_seed` for the seed-supplied confidence (Wave 4
        //   piece A — `EntityId` confidence plumbing). When absent OR when
        //   the decoded id is not in the seed map, anchor_confidence
        //   defaults to `1.0` — preserving the back-compat property.
        //
        // The first three columns (`id`/`name`/`type`) stay positional —
        // 30+ existing test fixtures and the `canned_graph_with` helper build
        // rows positionally; switching them to header lookup would break the
        // whole fixture surface.
        //
        // Defensive parsing (T-03-02-04 mitigation against malformed
        // GraphResult): row.first().and_then(...).unwrap_or("") + hex::decode
        // .unwrap_or_default() — a malformed row produces an empty-id RawHit
        // rather than panicking. Hit count never exceeds result.rows.len().
        let path_len_idx = result.headers.iter().position(|h| h == "path_length");
        let edge_w_idx = result.headers.iter().position(|h| h == "edge_weight_product");
        let source_id_idx = result.headers.iter().position(|h| h == "source_entity_id");

        let hits: Vec<RawHit> = result
            .rows
            .into_iter()
            .enumerate()
            .map(|(i, row)| {
                let id_hex = row.first().and_then(|v| v.as_str()).unwrap_or("").to_string();
                let id_bytes = hex::decode(&id_hex).unwrap_or_default();
                let name = row.get(1).cloned().unwrap_or(serde_json::Value::Null);
                let typ = row.get(2).cloned().unwrap_or(serde_json::Value::Null);

                // Header-keyed optional columns. f64 inputs are clamped at the
                // f32 conversion boundary; non-numeric / out-of-range cells
                // fall back to defaults rather than poisoning the score.
                let path_length = path_len_idx
                    .and_then(|idx| row.get(idx))
                    .and_then(|v| v.as_f64())
                    .map(|x| x as f32)
                    .unwrap_or(i as f32);
                let edge_weight_product = edge_w_idx
                    .and_then(|idx| row.get(idx))
                    .and_then(|v| v.as_f64())
                    .map(|x| x as f32)
                    .unwrap_or(1.0);

                // Wave 4: synthesize anchor_confidence by joining
                // `source_entity_id` against the per-seed confidence map.
                // Missing column → default 1.0 (back-compat). Decode failure
                // OR seed-not-in-map → default 1.0 (defensive; a row whose
                // anchor we can't identify is treated as full-confidence
                // rather than zero-confidence, which would silently zero out
                // a legitimate hit).
                let anchor_confidence = source_id_idx
                    .and_then(|idx| row.get(idx))
                    .and_then(|v| v.as_str())
                    .and_then(EntityId::from_hex)
                    .and_then(|seed| self.confidence_by_seed.get(&seed).copied())
                    .unwrap_or(1.0);

                let score = (edge_weight_product / (1.0 + path_length)) * anchor_confidence;

                RawHit {
                    id: id_bytes,
                    score,
                    rerank_applied: false,
                    degraded: false,
                    metadata: serde_json::json!({"name": name, "type": typ}),
                    source_op: SourceOp::Graph,
                }
            })
            .collect();

        Ok(hits)
    }

    fn as_any(&self) -> &dyn Any {
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Shorthand for the empty-seeds case in tests. Avoids `Vec::<(EntityId, f32)>::new()`
    /// boilerplate when the test only cares about non-seed constructor behavior
    /// (hops clamp, k default, graph default).
    fn no_seeds() -> Vec<(EntityId, f32)> {
        Vec::new()
    }

    #[test]
    fn anchored_clamps_hops_at_5() {
        // D-14: hops > MAX_GRAPH_HOPS (5) clamps down to 5.
        let g = Graph::anchored(no_seeds(), 100);
        assert_eq!(g.hops, MAX_GRAPH_HOPS);
        let g = Graph::anchored(no_seeds(), 6);
        assert_eq!(g.hops, MAX_GRAPH_HOPS);
        let g = Graph::anchored(no_seeds(), 5);
        assert_eq!(g.hops, MAX_GRAPH_HOPS);
    }

    #[test]
    fn anchored_clamps_hops_at_least_1() {
        // hops == 0 clamps UP to 1 — zero-hop traversal would only return the
        // anchor entities themselves (which the caller already has).
        let g = Graph::anchored(no_seeds(), 0);
        assert_eq!(g.hops, 1);
    }

    #[test]
    fn anchored_default_k_is_30() {
        // Matches Vector / Keyword default of 30.
        let g = Graph::anchored(no_seeds(), 2);
        assert_eq!(g.k, DEFAULT_GRAPH_K);
    }

    #[test]
    fn with_k_clamps_at_max() {
        // T-02-02-03 — clamp_k caps at super::MAX_K to defend against
        // k = 1_000_000 DoS.
        let g = Graph::anchored(no_seeds(), 2).with_k(usize::MAX);
        assert_eq!(g.k, super::super::MAX_K);
    }

    #[test]
    fn with_graph_overrides_default() {
        let g = Graph::anchored(no_seeds(), 2).with_graph("tenant_42_graph");
        assert_eq!(g.graph, "tenant_42_graph");
    }

    #[test]
    fn anchored_default_graph_is_lunaris_graph() {
        let g = Graph::anchored(no_seeds(), 2);
        assert_eq!(g.graph, LUNARIS_GRAPH_NAME);
        assert_eq!(g.graph, "lunaris_graph");
    }

    #[test]
    fn anchored_clamps_confidence_into_unit_interval() {
        // Out-of-range confidence values must NOT poison the score. Clamp to
        // [0.0, 1.0] at the constructor boundary.
        let id_high = EntityId::from_name_and_type("High", "Person");
        let id_neg = EntityId::from_name_and_type("Neg", "Person");
        let g = Graph::anchored(vec![(id_high, 100.0), (id_neg, -5.0)], 2);
        assert_eq!(g.confidence_by_seed.get(&id_high), Some(&1.0));
        assert_eq!(g.confidence_by_seed.get(&id_neg), Some(&0.0));
    }

    #[test]
    fn anchored_duplicate_seeds_collapse_to_max_confidence() {
        // Duplicate seeds in the input vector must collapse to the MAX
        // confidence — a caller-supplied low-confidence duplicate MUST NOT
        // demote a high-confidence anchor.
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 0.2), (id, 0.9), (id, 0.5)], 2);
        assert_eq!(g.confidence_by_seed.get(&id), Some(&0.9));
    }

    #[test]
    fn build_cypher_splices_hops_literal_and_parameterizes_ids_and_k() {
        // Property holds across ALL dialect tiers: hops is a literal,
        // EntityIds are parameterized, k is a parameter.
        let id1 = EntityId([1u8; 16]);
        let id2 = EntityId([2u8; 16]);
        let g = Graph::anchored(vec![(id1, 1.0), (id2, 1.0)], 3);
        let id1_hex = format!("{}", id1);
        let id2_hex = format!("{}", id2);
        for dialect in [CypherDialect::Legacy, CypherDialect::Full] {
            let q = g.build_cypher(dialect);
            assert!(
                q.cypher.contains("[*1..3]"),
                "[{dialect:?}] hops literal must be spliced: {}",
                q.cypher
            );
            assert!(
                !q.cypher.contains(&id1_hex),
                "[{dialect:?}] entity id 1 must NOT be in cypher text: {}",
                q.cypher
            );
            assert!(
                !q.cypher.contains(&id2_hex),
                "[{dialect:?}] entity id 2 must NOT be in cypher text: {}",
                q.cypher
            );
            let ids = q.params.get("ids").expect("ids param present").as_array().unwrap();
            assert_eq!(ids.len(), 2);
            assert_eq!(ids[0], serde_json::Value::String(id1_hex.clone()));
            assert_eq!(ids[1], serde_json::Value::String(id2_hex.clone()));
            assert!(q.params.contains_key("k"));
            assert_eq!(q.params.get("k").unwrap().as_u64(), Some(DEFAULT_GRAPH_K as u64));
            assert!(!q.params.contains_key("hops"), "hops MUST NOT be a parameter");
        }
    }

    #[test]
    fn build_cypher_uses_id_hex_property_name_not_id() {
        // W-7 fix: MATCH and RETURN MUST use the `id_hex` property (Plan
        // 03-03's GraphNode writes `id_hex`; using `id` would silently return
        // zero rows).
        //
        // Inspector-UAT fix (2026-06-16): the anchor filter MUST be a `WHERE`
        // clause, NOT the inline-property form `(n {id_hex: sid})`. Moon's
        // Cypher executor SILENTLY IGNORES inline-property filters on a plain
        // MATCH (live-confirmed) — it matches every node as an anchor, so the
        // traversal returns the whole connected component and neither `root`
        // nor `hops` constrains. See `feedback_moon_cypher_inline_filter`.
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 1.0)], 2);
        for dialect in [CypherDialect::Legacy, CypherDialect::Full] {
            let q = g.build_cypher(dialect);
            assert!(
                q.cypher.contains("WHERE n.id_hex = sid"),
                "[{dialect:?}] anchor MUST filter via `WHERE n.id_hex = sid` (Moon ignores \
                 inline-property filters): {}",
                q.cypher
            );
            assert!(
                !q.cypher.contains("{id_hex: sid}"),
                "[{dialect:?}] inline-property filter `(n {{id_hex: sid}})` is silently ignored \
                 by Moon — must not be used: {}",
                q.cypher
            );
            assert!(
                q.cypher.contains("m.id_hex AS id"),
                "[{dialect:?}] RETURN must select m.id_hex AS id: {}",
                q.cypher
            );
        }
    }

    #[test]
    fn build_cypher_legacy_dialect_omits_path_metrics_and_reduce() {
        // Legacy tier (Moon's ceiling): id/name/type only. No
        // path binding, no length(), no reduce(), no source_entity_id.
        // This is the universally-supported template.
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 1.0)], 2);
        let q = g.build_cypher(CypherDialect::Legacy);
        assert!(
            !q.cypher.contains("MATCH p ="),
            "Legacy MUST NOT bind a path variable (Moon rejects it): {}",
            q.cypher
        );
        assert!(
            !q.cypher.contains("length("),
            "Legacy MUST NOT call length() (Moon function table omits it): {}",
            q.cypher
        );
        assert!(
            !q.cypher.contains("reduce("),
            "Legacy MUST NOT call reduce() (Moon function table omits it): {}",
            q.cypher
        );
        assert!(
            !q.cypher.contains("source_entity_id"),
            "Legacy MUST NOT emit source_entity_id alias: {}",
            q.cypher
        );
    }

    #[test]
    fn build_cypher_full_dialect_emits_full_wave4_columns() {
        // Full tier (forward-compat — no backend supports this yet): path
        // binding + length(p) + source_entity_id + reduce(...) AS
        // edge_weight_product.
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 1.0)], 2);
        let q = g.build_cypher(CypherDialect::Full);
        assert!(q.cypher.contains("MATCH p ="), "Full MUST bind path: {}", q.cypher);
        assert!(
            q.cypher.contains("length(p) AS path_length"),
            "Full MUST emit length(p) AS path_length: {}",
            q.cypher
        );
        assert!(q.cypher.contains("reduce("), "Full MUST emit reduce(...): {}", q.cypher);
        assert!(
            q.cypher.contains("edge_weight_product"),
            "Full MUST emit edge_weight_product alias: {}",
            q.cypher
        );
        assert!(
            q.cypher.contains("n.id_hex AS source_entity_id"),
            "Full MUST emit source_entity_id alias: {}",
            q.cypher
        );
    }

    #[test]
    fn build_cypher_targets_lunaris_graph_by_default() {
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 1.0)], 2);
        for dialect in [CypherDialect::Legacy, CypherDialect::Full] {
            let q = g.build_cypher(dialect);
            assert_eq!(q.graph, "lunaris_graph", "[{dialect:?}] default graph name");
        }
    }

    #[test]
    fn build_cypher_with_custom_graph_name() {
        // with_graph() lets tenant-isolated deployments scope each tenant to
        // its own AGE / Moon graph (property holds per-dialect).
        let id = EntityId::from_name_and_type("Alice", "Person");
        let g = Graph::anchored(vec![(id, 1.0)], 2).with_graph("tenant_42_graph");
        for dialect in [CypherDialect::Legacy, CypherDialect::Full] {
            let q = g.build_cypher(dialect);
            assert_eq!(q.graph, "tenant_42_graph", "[{dialect:?}] custom graph name");
        }
    }

    #[test]
    fn graph_is_dyn_compatible() {
        // Compile-time proof Graph constructs as Box<dyn Retriever>; required
        // for the AndRetriever / FuseRrfRetriever composition path.
        let _: Box<dyn Retriever> = Box::new(Graph::anchored(no_seeds(), 2));
    }
}