kglite 0.17.2

//! Structural-validator rule procedures exposed via Cypher `CALL`.
//!
//! Each walks the graph directly via [`GraphRead`] and emits [`ResultRow`]s
//! with node bindings — no recursion through the executor, no `g.cypher()`
//! round-trip per rule — so the rows flow into the surrounding Cypher
//! (`WHERE` / `ORDER BY` / `LIMIT` / `RETURN`) the same way
//! [`super::call_clause::CypherExecutor::centrality_to_rows`] does for
//! `pagerank`.
//!
//! This module owns the family dispatcher; `call_clause::execute_call`
//! remains the shared CALL validation and budget choke point.

use std::collections::HashMap;

use petgraph::graph::NodeIndex;
use petgraph::Direction;

use super::super::ast::YieldItem;
use super::super::result::ResultRow;
use crate::datatypes::values::Value;
use crate::graph::dir_graph::DirGraph;
use crate::graph::schema::InternedKey;
use crate::graph::storage::GraphRead;

/// The procedure names `execute_call` routes to [`execute_rule_procedure`],
/// beside the dispatcher that answers them so wiring a new rule procedure
/// touches this file and the registry rather than the CALL clause. Every
/// name here must be answered below: `ontology_audit` and
/// the node/edge property procedures return from the early declaration-driven
/// branches, the rest from the final match, whose `unreachable!()` is the
/// consequence of adding a name here and nowhere else.
pub(super) const RULE_PROCEDURES: &[&str] = &[
    "orphan_node",
    "self_loop",
    "cycle_2step",
    "missing_required_edge",
    "missing_inbound_edge",
    "duplicate_title",
    "duplicate_id",
    "outline",
    "null_property",
    "inverse_violation",
    "transitivity_violation",
    "cardinality_violation",
    "type_domain_violation",
    "type_range_violation",
    "parallel_edges",
    "edge_property_violation",
    "node_property_violation",
    "ontology_audit",
    "kg_knn",
];

/// Dispatch a structural-rule procedure after the CALL clause has completed
/// shared name/YIELD/parameter validation.
pub(super) fn execute_rule_procedure(
    proc_name: &str,
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    // With an ontology declared, a no-argument call runs the declaration-
    // driven form (ontology_procedures.rs) instead of erroring on the
    // missing parameter.
    if params.is_empty() {
        if let Some(result) =
            super::ontology_procedures::no_arg_declaration_rows(proc_name, graph, yield_items)
        {
            return result;
        }
    }
    if proc_name == "ontology_audit" {
        return super::ontology_procedures::execute_ontology_audit(graph, params, yield_items);
    }
    // Declaration-driven like the no-arg forms above, but with no
    // parameter-driven twin to fall back to (`required_properties` /
    // `property_types` are declaration-only checks), so it dispatches here
    // rather than through `proc_check`.
    if proc_name == "node_property_violation" {
        return super::node_ontology::execute_node_property_violation(graph, params, yield_items);
    }
    if proc_name == "edge_property_violation" {
        return super::ontology_procedures::execute_edge_property_violation(
            graph,
            params,
            yield_items,
        );
    }
    let mut rows = match proc_name {
        "orphan_node" => execute_orphan_node(graph, params, yield_items),
        "self_loop" => execute_self_loop(graph, params, yield_items),
        "cycle_2step" => execute_cycle_2step(graph, params, yield_items),
        "missing_required_edge" => execute_missing_required_edge(graph, params, yield_items),
        "missing_inbound_edge" => execute_missing_inbound_edge(graph, params, yield_items),
        "duplicate_title" => execute_duplicate_title(graph, params, yield_items),
        "duplicate_id" => execute_duplicate_id(graph, params, yield_items),
        "outline" => execute_outline(graph, params, yield_items),
        "null_property" => execute_null_property(graph, params, yield_items),
        "inverse_violation" => execute_inverse_violation(graph, params, yield_items),
        "transitivity_violation" => execute_transitivity_violation(graph, params, yield_items),
        "cardinality_violation" => execute_cardinality_violation(graph, params, yield_items),
        "type_domain_violation" => execute_type_domain_violation(graph, params, yield_items),
        "type_range_violation" => execute_type_range_violation(graph, params, yield_items),
        "parallel_edges" => execute_parallel_edges(graph, params, yield_items),
        "kg_knn" => execute_kg_knn(graph, params, yield_items),
        _ => unreachable!("non-rule procedure routed to rule dispatcher: {proc_name}"),
    }?;
    // The declaration-backed procs carry a `rule` column (the no-arg form
    // stamps the declaration name); explicit calls stamp their own subject
    // so a `YIELD rule` never hits a missing binding.
    if matches!(
        proc_name,
        "missing_required_edge"
            | "inverse_violation"
            | "transitivity_violation"
            | "cardinality_violation"
            | "type_domain_violation"
            | "type_range_violation"
    ) {
        let subject = ["edge", "rel", "rel_a", "type"]
            .iter()
            .find_map(|k| match params.get(*k) {
                Some(Value::String(s)) => Some(s.clone()),
                _ => None,
            })
            .unwrap_or_default();
        super::ontology_procedures::stamp_rule(&mut rows, yield_items, &subject);
    }
    Ok(rows)
}

/// `CALL orphan_node({type, link_type?, direction?}) YIELD node`
///
/// Yields one row per node of `type` that has zero matching edges.
///
/// - **Default** (no `link_type`, no `direction`) — node has zero edges
///   in any direction (the historical "graph-theoretically isolated"
///   semantics).
/// - **`link_type: 'CALLS'`** — restrict the edge-existence check to
///   that connection type only. Other edges (e.g. `DEFINES`, `HAS_METHOD`)
///   don't disqualify the node.
/// - **`direction: 'in' | 'out' | 'both'`** — which side(s) to check
///   (default `'both'`). `'in'` finds nodes with no inbound matching
///   edge; symmetric for `'out'`.
///
/// The combination is the natural "find functions never called" query:
/// `CALL orphan_node({type: 'Function', link_type: 'CALLS', direction: 'in'})`.
///
/// @procedure: orphan_node
pub(super) fn execute_orphan_node(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "orphan_node")?;
    let link_type = optional_string_param(params, "link_type", "orphan_node")?;
    let direction = optional_string_param(params, "direction", "orphan_node")?
        .unwrap_or_else(|| "both".to_string());
    let check_out = matches!(direction.as_str(), "out" | "both");
    let check_in = matches!(direction.as_str(), "in" | "both");
    if !check_out && !check_in {
        return Err(format!(
            "CALL orphan_node: invalid direction '{direction}'. Use 'in', 'out', or 'both'."
        ));
    }
    let edge_key = link_type.as_deref().map(InternedKey::from_str);
    let yield_var = require_node_yield(yield_items, "orphan_node", "node")?;
    let nodes = type_indices(graph, &node_type)?;

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let has_match = |dir: Direction| -> bool {
            let mut iter = graph.graph.edges_directed(nidx, dir);
            match edge_key {
                Some(key) => iter.any(|er| er.weight().connection_type == key),
                None => iter.next().is_some(),
            }
        };
        if check_out && has_match(Direction::Outgoing) {
            continue;
        }
        if check_in && has_match(Direction::Incoming) {
            continue;
        }
        rows.push(make_node_row(&yield_var, nidx));
    }
    Ok(rows)
}

/// `CALL self_loop({type: 'Person', edge: 'KNOWS'}) YIELD node`
///
/// Yields one row per node of `type` that has an outgoing `edge` whose
/// target is itself. Always a data error in tree-shaped hierarchies;
/// occasionally legitimate for self-referential domain edges.
///
/// @procedure: self_loop
pub(super) fn execute_self_loop(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "self_loop")?;
    let edge_type = require_string_param(params, "edge", "self_loop")?;
    let yield_var = require_node_yield(yield_items, "self_loop", "node")?;
    let nodes = type_indices(graph, &node_type)?;
    let edge_key = InternedKey::from_str(&edge_type);

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let hit = graph
            .graph
            .edges_directed(nidx, Direction::Outgoing)
            .any(|er| er.target() == nidx && er.weight().connection_type == edge_key);
        if hit {
            rows.push(make_node_row(&yield_var, nidx));
        }
    }
    Ok(rows)
}

/// `CALL cycle_2step({type: 'Person', edge: 'KNOWS'}) YIELD node_a, node_b`
///
/// Yields one row per `(a, b)` pair where `a -[edge]-> b -[edge]-> a`,
/// `a` and `b` are both of `type`, and `a < b` (deduplicates the pair).
/// Variables are named `node_a` / `node_b` to avoid clashing with the
/// Cypher reserved word `END` (CASE expressions).
///
/// @procedure: cycle_2step
pub(super) fn execute_cycle_2step(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "cycle_2step")?;
    let edge_type = require_string_param(params, "edge", "cycle_2step")?;
    let (start_var, end_var) =
        require_two_node_yields(yield_items, "cycle_2step", "node_a", "node_b")?;
    let nodes = type_indices(graph, &node_type)?;
    let node_set: std::collections::HashSet<NodeIndex> = nodes.iter().collect();
    let edge_key = InternedKey::from_str(&edge_type);

    let mut rows = Vec::new();
    for a in nodes.iter() {
        for er_a in graph.graph.edges_directed(a, Direction::Outgoing) {
            if er_a.weight().connection_type != edge_key {
                continue;
            }
            let b = er_a.target();
            if a >= b {
                continue;
            }
            if !node_set.contains(&b) {
                continue;
            }
            let returns = graph
                .graph
                .edges_directed(b, Direction::Outgoing)
                .any(|er_b| er_b.target() == a && er_b.weight().connection_type == edge_key);
            if returns {
                let mut row = ResultRow::new();
                row.node_bindings.insert(start_var.clone(), a);
                row.node_bindings.insert(end_var.clone(), b);
                rows.push(row);
            }
        }
    }
    Ok(rows)
}

/// `CALL missing_required_edge({type: 'Wellbore', edge: 'IN_LICENCE'}) YIELD node`
///
/// Yields one row per node of `type` that has **no outgoing** edge of
/// `edge`. Refuses to execute when the graph's actual schema for `edge`
/// makes `type` the target rather than the source — flips that case to a
/// `DirectionMismatch` error suggesting `missing_inbound_edge`.
///
/// @procedure: missing_required_edge
pub(super) fn execute_missing_required_edge(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "missing_required_edge")?;
    let edge_type = require_string_param(params, "edge", "missing_required_edge")?;
    let yield_var = require_node_yield(yield_items, "missing_required_edge", "node")?;
    validate_direction(
        graph,
        &node_type,
        &edge_type,
        OUTBOUND,
        "missing_required_edge",
    )?;
    let nodes = type_indices(graph, &node_type)?;
    let edge_key = InternedKey::from_str(&edge_type);

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let has_edge = has_edge_of_type(graph, nidx, Direction::Outgoing, &edge_type, edge_key);
        if !has_edge {
            rows.push(make_node_row(&yield_var, nidx));
        }
    }
    Ok(rows)
}

/// `CALL missing_inbound_edge({type: 'Discovery', edge: 'IN_DISCOVERY'}) YIELD node`
///
/// Mirror of [`execute_missing_required_edge`]: yields nodes of `type`
/// with no **incoming** edge of `edge`.
///
/// @procedure: missing_inbound_edge
pub(super) fn execute_missing_inbound_edge(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "missing_inbound_edge")?;
    let edge_type = require_string_param(params, "edge", "missing_inbound_edge")?;
    let yield_var = require_node_yield(yield_items, "missing_inbound_edge", "node")?;
    validate_direction(
        graph,
        &node_type,
        &edge_type,
        INBOUND,
        "missing_inbound_edge",
    )?;
    let nodes = type_indices(graph, &node_type)?;
    let edge_key = InternedKey::from_str(&edge_type);

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let has_edge = has_edge_of_type(graph, nidx, Direction::Incoming, &edge_type, edge_key);
        if !has_edge {
            rows.push(make_node_row(&yield_var, nidx));
        }
    }
    Ok(rows)
}

/// `CALL duplicate_title({type: 'Prospect'}) YIELD node`
///
/// Yields one row per node of `type` whose title is shared with at least
/// one other node of the same type. Agents can aggregate downstream:
///
/// ```cypher
/// CALL duplicate_title({type: 'Prospect'}) YIELD node
/// WITH node.title AS t, collect(node) AS dups
/// WHERE size(dups) > 1
/// RETURN t, size(dups) AS count
/// ORDER BY count DESC LIMIT 20
/// ```
///
/// @procedure: duplicate_title
pub(super) fn execute_duplicate_title(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "duplicate_title")?;
    let yield_var = require_node_yield(yield_items, "duplicate_title", "node")?;
    let nodes = type_indices(graph, &node_type)?;

    let mut counts: HashMap<String, u32> = HashMap::with_capacity(nodes.len());
    for nidx in nodes.iter() {
        if let Some(title) = title_of(graph, nidx) {
            *counts.entry(title).or_insert(0) += 1;
        }
    }
    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        if let Some(title) = title_of(graph, nidx) {
            if counts.get(&title).copied().unwrap_or(0) > 1 {
                rows.push(make_node_row(&yield_var, nidx));
            }
        }
    }
    Ok(rows)
}

/// `CALL duplicate_id({type: 'Artifact'}) YIELD node`
///
/// Yields one row per node of `type` whose id is shared with at least one
/// other node of the same type. The sibling of `duplicate_title` for the
/// identity column — useful after bulk writes, since a `CREATE` fanned out
/// over a multi-row `MATCH` (standard Cypher: one create per row) can mint
/// several nodes with the same id without complaint. Aggregate downstream as
/// in [`execute_duplicate_title`].
///
/// @procedure: duplicate_id
pub(super) fn execute_duplicate_id(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "duplicate_id")?;
    let yield_var = require_node_yield(yield_items, "duplicate_id", "node")?;
    let nodes = type_indices(graph, &node_type)?;

    let mut counts: HashMap<String, u32> = HashMap::with_capacity(nodes.len());
    for nidx in nodes.iter() {
        if let Some(id) = id_of(graph, nidx) {
            *counts.entry(id).or_insert(0) += 1;
        }
    }
    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        if let Some(id) = id_of(graph, nidx) {
            if counts.get(&id).copied().unwrap_or(0) > 1 {
                rows.push(make_node_row(&yield_var, nidx));
            }
        }
    }
    Ok(rows)
}

/// `CALL inverse_violation({rel_a: 'parent_of', rel_b: 'child_of'}) YIELD a, b`
///
/// Yields one row per `(a, b)` pair where `(a)-[rel_a]->(b)` exists but
/// the inverse `(b)-[rel_b]->(a)` does not — the unidirectional cases where
/// two relations are declared logical inverses (parent_of/child_of).
///
/// @procedure: inverse_violation
pub(super) fn execute_inverse_violation(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let rel_a = require_string_param(params, "rel_a", "inverse_violation")?;
    let rel_b = require_string_param(params, "rel_b", "inverse_violation")?;
    let (a_var, b_var) = require_two_node_yields(yield_items, "inverse_violation", "a", "b")?;
    let key_a = InternedKey::from_str(&rel_a);
    let key_b = InternedKey::from_str(&rel_b);

    let mut rows = Vec::new();
    for a in graph.graph.node_indices() {
        for er in graph.graph.edges_directed(a, Direction::Outgoing) {
            if er.weight().connection_type != key_a {
                continue;
            }
            let b = er.target();
            let has_inverse = graph
                .graph
                .edges_directed(b, Direction::Outgoing)
                .any(|er2| er2.target() == a && er2.weight().connection_type == key_b);
            if !has_inverse {
                let mut row = ResultRow::new();
                row.node_bindings.insert(a_var.clone(), a);
                row.node_bindings.insert(b_var.clone(), b);
                rows.push(row);
            }
        }
    }
    Ok(rows)
}

/// `CALL transitivity_violation({rel: 'subClassOf'}) YIELD a, b, c`
///
/// For every `(a)-[rel]->(b)-[rel]->(c)` chain, yields the triple when no
/// direct `(a)-[rel]->(c)` edge exists. Pairs with `cycle_2step` for the
/// cycle case.
///
/// @procedure: transitivity_violation
pub(super) fn execute_transitivity_violation(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let rel = require_string_param(params, "rel", "transitivity_violation")?;
    let a_var = require_node_yield(yield_items, "transitivity_violation", "a")?;
    let b_var = require_node_yield(yield_items, "transitivity_violation", "b")?;
    let c_var = require_node_yield(yield_items, "transitivity_violation", "c")?;
    let key = InternedKey::from_str(&rel);

    let mut rows = Vec::new();
    for a in graph.graph.node_indices() {
        let direct_a: std::collections::HashSet<NodeIndex> = graph
            .graph
            .edges_directed(a, Direction::Outgoing)
            .filter(|er| er.weight().connection_type == key)
            .map(|er| er.target())
            .collect();
        for &b in &direct_a {
            for er_b in graph.graph.edges_directed(b, Direction::Outgoing) {
                if er_b.weight().connection_type != key {
                    continue;
                }
                let c = er_b.target();
                if c == a || c == b {
                    continue;
                }
                if !direct_a.contains(&c) {
                    let mut row = ResultRow::new();
                    row.node_bindings.insert(a_var.clone(), a);
                    row.node_bindings.insert(b_var.clone(), b);
                    row.node_bindings.insert(c_var.clone(), c);
                    rows.push(row);
                }
            }
        }
    }
    Ok(rows)
}

/// `CALL cardinality_violation({type: 'Wellbore', edge: 'IN_LICENCE', min: 1, max: 1}) YIELD node, count`
///
/// Yields nodes of `type` whose outgoing edge count of `edge` falls
/// outside `[min, max]`. Either bound is optional (`min` defaults to 0,
/// `max` defaults to no upper limit). Setting `max: 1` catches functional-
/// property violations; setting `min: 1` catches missing-required-edge.
///
/// @procedure: cardinality_violation
pub(super) fn execute_cardinality_violation(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "cardinality_violation")?;
    let edge_type = require_string_param(params, "edge", "cardinality_violation")?;
    let min_count = call_param_i64(params, "min", 0).max(0) as usize;
    let max_count = call_param_opt_i64(params, "max").map(|v| v.max(0) as usize);
    let node_var = require_node_yield(yield_items, "cardinality_violation", "node")?;
    let count_var = require_scalar_yield(yield_items, "cardinality_violation", "count")?;
    let nodes = type_indices(graph, &node_type)?;
    let edge_key = InternedKey::from_str(&edge_type);

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let count = graph
            .graph
            .edges_directed(nidx, Direction::Outgoing)
            .filter(|er| er.weight().connection_type == edge_key)
            .count();
        let too_few = count < min_count;
        let too_many = max_count.is_some_and(|m| count > m);
        if too_few || too_many {
            let mut row = ResultRow::new();
            row.node_bindings.insert(node_var.clone(), nidx);
            row.projected
                .insert(count_var.clone(), Value::Int64(count as i64));
            rows.push(row);
        }
    }
    Ok(rows)
}

/// `CALL type_domain_violation({edge: 'CITES', expected_source: 'Case'}) YIELD source, target`
///
/// Yields edges of `edge` whose source node is not of `expected_source`
/// type.
///
/// @procedure: type_domain_violation
pub(super) fn execute_type_domain_violation(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let edge_type = require_string_param(params, "edge", "type_domain_violation")?;
    let expected = require_string_param(params, "expected_source", "type_domain_violation")?;
    let src_var = require_node_yield(yield_items, "type_domain_violation", "source")?;
    let tgt_var = require_node_yield(yield_items, "type_domain_violation", "target")?;
    let accepted = [expected];
    let pairs =
        super::ontology_procedures::scan_endpoint_mismatch(graph, &edge_type, &accepted, true);
    let mut rows = Vec::new();
    for (src, tgt) in pairs {
        let mut row = ResultRow::new();
        row.node_bindings.insert(src_var.clone(), src);
        row.node_bindings.insert(tgt_var.clone(), tgt);
        rows.push(row);
    }
    Ok(rows)
}

/// `CALL type_range_violation({edge: 'CITES', expected_target: 'Case'}) YIELD source, target`
///
/// Mirror of [`execute_type_domain_violation`]: yields edges whose
/// target node is not of `expected_target` type.
///
/// @procedure: type_range_violation
pub(super) fn execute_type_range_violation(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let edge_type = require_string_param(params, "edge", "type_range_violation")?;
    let expected = require_string_param(params, "expected_target", "type_range_violation")?;
    let src_var = require_node_yield(yield_items, "type_range_violation", "source")?;
    let tgt_var = require_node_yield(yield_items, "type_range_violation", "target")?;
    let accepted = [expected];
    let pairs =
        super::ontology_procedures::scan_endpoint_mismatch(graph, &edge_type, &accepted, false);
    let mut rows = Vec::new();
    for (src, tgt) in pairs {
        let mut row = ResultRow::new();
        row.node_bindings.insert(src_var.clone(), src);
        row.node_bindings.insert(tgt_var.clone(), tgt);
        rows.push(row);
    }
    Ok(rows)
}

/// `CALL parallel_edges({edge: 'CITES'}) YIELD a, b, count`
///
/// Yields one row per `(a, b)` pair connected by more than one edge of
/// the same `edge` type. Almost always a load-time bug — duplicate rows
/// in the source CSV, or an upsert path that didn't dedupe.
///
/// @procedure: parallel_edges
pub(super) fn execute_parallel_edges(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let edge_type = require_string_param(params, "edge", "parallel_edges")?;
    let (a_var, b_var) = require_two_node_yields(yield_items, "parallel_edges", "a", "b")?;
    let count_var = require_scalar_yield(yield_items, "parallel_edges", "count")?;
    let key = InternedKey::from_str(&edge_type);

    let mut counts: HashMap<(NodeIndex, NodeIndex), u32> = HashMap::new();
    for er in graph.graph.edge_references() {
        if er.weight().connection_type == key {
            *counts.entry((er.source(), er.target())).or_insert(0) += 1;
        }
    }
    let mut rows = Vec::new();
    for ((a, b), c) in counts {
        if c > 1 {
            let mut row = ResultRow::new();
            row.node_bindings.insert(a_var.clone(), a);
            row.node_bindings.insert(b_var.clone(), b);
            row.projected
                .insert(count_var.clone(), Value::Int64(c as i64));
            rows.push(row);
        }
    }
    Ok(rows)
}

/// `CALL null_property({type: 'Person', property: 'email'}) YIELD node`
///
/// Yields nodes of `type` where `property` is absent, null, or the empty
/// string.
///
/// @procedure: null_property
pub(super) fn execute_null_property(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let node_type = require_string_param(params, "type", "null_property")?;
    let property = require_string_param(params, "property", "null_property")?;
    let yield_var = require_node_yield(yield_items, "null_property", "node")?;
    let nodes = type_indices(graph, &node_type)?;

    let mut rows = Vec::new();
    for nidx in nodes.iter() {
        let node = match graph.graph.node_view(nidx) {
            Some(n) => n,
            None => continue,
        };
        let val = node.get_property_value(&property);
        let is_null = match val {
            None => true,
            Some(Value::Null) => true,
            Some(Value::String(ref s)) if s.is_empty() => true,
            _ => false,
        };
        if is_null {
            rows.push(make_node_row(&yield_var, nidx));
        }
    }
    Ok(rows)
}

/// `CALL kg_knn({lat: 60.4, lon: 5.3, target_type: 'Field', k: 5}) YIELD node, distance_m`
///
/// Returns the *k* nodes of `target_type` closest (geodesic) to a given
/// `(lat, lon)` coordinate, by scanning every node of the type and keeping
/// the k smallest distances.
///
/// Nodes without spatial config (no `location`/`geometry` mapping) are
/// skipped silently. Distance is point-to-point from the configured
/// `location`; when that is missing, it is measured to the *centroid* of the
/// `geometry` WKT, not to its nearest edge.
///
/// @procedure: kg_knn
pub(super) fn execute_kg_knn(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    let lat = call_param_f64_required(params, "lat", "kg_knn")?;
    let lon = call_param_f64_required(params, "lon", "kg_knn")?;
    let target_type = require_string_param(params, "target_type", "kg_knn")?;
    let k = call_param_i64(params, "k", 10).max(1) as usize;
    let node_var = require_node_yield(yield_items, "kg_knn", "node")?;
    let dist_var = require_scalar_yield(yield_items, "kg_knn", "distance_m")?;
    let nodes = type_indices(graph, &target_type)?;

    // Min-heap keyed on distance — emit the K closest.
    use std::cmp::Ordering;
    use std::collections::BinaryHeap;
    #[derive(PartialEq)]
    struct Entry(f64, NodeIndex);
    impl Eq for Entry {}
    impl Ord for Entry {
        fn cmp(&self, other: &Self) -> Ordering {
            // BinaryHeap is a max-heap; negate for min-heap-by-distance.
            other
                .0
                .partial_cmp(&self.0)
                .unwrap_or(Ordering::Equal)
                .then_with(|| self.1.index().cmp(&other.1.index()))
        }
    }
    impl PartialOrd for Entry {
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
            Some(self.cmp(other))
        }
    }

    let mut heap: BinaryHeap<Entry> = BinaryHeap::with_capacity(nodes.len());
    let config = graph.get_spatial_config(&target_type);
    for nidx in nodes.iter() {
        let node = match graph.graph.node_view(nidx) {
            Some(n) => n,
            None => continue,
        };
        // Compute geodesic distance: prefer location (point-to-point),
        // fall back to centroid of geometry.
        let mut dist = f64::INFINITY;
        if let Some(cfg) = config {
            if let Some((lat_f, lon_f)) = &cfg.location {
                if let (Some(plat), Some(plon)) = (
                    node.get_property(lat_f)
                        .as_deref()
                        .and_then(crate::graph::core::value_operations::value_to_f64),
                    node.get_property(lon_f)
                        .as_deref()
                        .and_then(crate::graph::core::value_operations::value_to_f64),
                ) {
                    dist = crate::graph::features::spatial::geodesic_distance(lat, lon, plat, plon);
                }
            }
            if dist == f64::INFINITY {
                if let Some(geom_f) = &cfg.geometry {
                    if let Some(Value::String(wkt)) = node.get_property(geom_f).as_deref() {
                        if let Ok(g) = crate::graph::features::spatial::parse_wkt(wkt) {
                            if let Ok((clat, clon)) =
                                crate::graph::features::spatial::wkt_centroid(wkt)
                            {
                                dist = crate::graph::features::spatial::geodesic_distance(
                                    lat, lon, clat, clon,
                                );
                            }
                            let _ = g; // silence unused if no centroid
                        }
                    }
                }
            }
        }
        if dist.is_finite() {
            heap.push(Entry(dist, nidx));
        }
    }

    let mut rows: Vec<ResultRow> = Vec::with_capacity(k);
    while rows.len() < k {
        match heap.pop() {
            Some(Entry(d, idx)) => {
                let mut row = ResultRow::new();
                row.node_bindings.insert(node_var.clone(), idx);
                row.projected.insert(dist_var.clone(), Value::Float64(d));
                rows.push(row);
            }
            None => break,
        }
    }
    Ok(rows)
}

fn call_param_f64_required(
    params: &HashMap<String, Value>,
    key: &str,
    proc: &str,
) -> Result<f64, String> {
    match params.get(key) {
        Some(Value::Int64(v)) => Ok(*v as f64),
        Some(Value::Float64(v)) => Ok(*v),
        Some(other) => Err(format!(
            "CALL {proc}: parameter '{key}' must be numeric, got {other:?}"
        )),
        None => Err(format!(
            "CALL {proc}: missing required parameter '{key}'.{}",
            param_schema_hint(proc)
        )),
    }
}

/// Per-procedure parameter schema. Used to enrich the "missing required
/// parameter" / "must be a string" error messages so first-time users
/// don't have to brute-force-guess parameter names. Each entry is
/// `(proc_name, &[(param_name, required)])`.
///
/// Keep alphabetised by proc_name so additions land in a stable place.
const RULE_PARAM_SCHEMAS: &[(&str, &[(&str, bool)])] = &[
    (
        "cardinality_violation",
        &[
            ("type", true),
            ("edge", true),
            ("min", false),
            ("max", false),
        ],
    ),
    ("cycle_2step", &[("type", true), ("edge", true)]),
    ("duplicate_title", &[("type", true)]),
    ("duplicate_id", &[("type", true)]),
    (
        "outline",
        &[
            ("root", true),
            ("root_type", false),
            ("edge", true),
            ("max_depth", false),
        ],
    ),
    ("inverse_violation", &[("rel_a", true), ("rel_b", true)]),
    (
        "kg_knn",
        &[
            ("lat", true),
            ("lon", true),
            ("target_type", true),
            ("k", false),
        ],
    ),
    ("missing_inbound_edge", &[("type", true), ("edge", true)]),
    ("missing_required_edge", &[("type", true), ("edge", true)]),
    ("null_property", &[("type", true), ("property", true)]),
    (
        "orphan_node",
        &[("type", true), ("link_type", false), ("direction", false)],
    ),
    ("parallel_edges", &[("edge", true)]),
    ("self_loop", &[("type", true), ("edge", true)]),
    ("transitivity_violation", &[("rel", true)]),
    (
        "type_domain_violation",
        &[("edge", true), ("expected_source", true)],
    ),
    (
        "type_range_violation",
        &[("edge", true), ("expected_target", true)],
    ),
];

/// Format a parameter-schema hint to append to error messages. Returns
/// `""` when the procedure has no registered schema (so older procedures
/// keep their existing error format).
fn param_schema_hint(proc: &str) -> String {
    let Some((_, schema)) = RULE_PARAM_SCHEMAS.iter().find(|(name, _)| *name == proc) else {
        return String::new();
    };
    let required: Vec<&str> = schema
        .iter()
        .filter_map(|(name, req)| if *req { Some(*name) } else { None })
        .collect();
    let optional: Vec<&str> = schema
        .iter()
        .filter_map(|(name, req)| if !*req { Some(*name) } else { None })
        .collect();
    let mut parts = Vec::new();
    if !required.is_empty() {
        parts.push(format!("required: [{}]", required.join(", ")));
    }
    if !optional.is_empty() {
        parts.push(format!("optional: [{}]", optional.join(", ")));
    }
    if parts.is_empty() {
        String::new()
    } else {
        format!(" Accepted parameters — {}.", parts.join(", "))
    }
}

/// Per-edge-type presence check. The disk backend's
/// `edges_directed_filtered` may use an inverted-index fast path that
/// can miss specific (node, edge_type) combinations on sparsely-built
/// disk graphs — iterating the unfiltered edge stream and matching
/// against the interned key is correct in all storage modes and the
/// cost is identical for nodes with low fan-out.
fn has_edge_of_type(
    graph: &DirGraph,
    nidx: NodeIndex,
    dir: Direction,
    _edge_type_name: &str,
    edge_type_key: InternedKey,
) -> bool {
    graph
        .graph
        .edges_directed(nidx, dir)
        .any(|er| er.weight().connection_type == edge_type_key)
}

/// `CALL outline({root, root_type?, edge, max_depth?})`
///
/// BFS from the node whose id is `root`, following outgoing `edge`-typed edges,
/// yielding the spanning tree — each reachable node once, at its first-discovery
/// `depth` (0 = root). The appended type columns describe ids without relying
/// on Python equality, while `node_token` / `parent_token` carry physical slots
/// that are reconstruction keys only within this result, not persistent ids.
/// `root_type` selects between nodes of different primary types whose ids
/// compare equal; an ambiguous root is refused. This is the tree **structure**; render it as a nested outline /
/// markdown document in the binding layer (`g.outline(...)`) — the engine stays
/// a query engine. `max_depth` (optional) is a non-negative descent bound.
///
/// @procedure: outline
pub(super) fn execute_outline(
    graph: &DirGraph,
    params: &HashMap<String, Value>,
    yield_items: &[YieldItem],
) -> Result<Vec<ResultRow>, String> {
    use std::collections::{HashSet, VecDeque};

    let edge_type = require_string_param(params, "edge", "outline")?;
    let root_id = params
        .get("root")
        .ok_or_else(|| "CALL outline: missing required parameter 'root'".to_string())?;
    let root_type = optional_string_param(params, "root_type", "outline")?;
    let max_depth = outline_max_depth(params)?;
    let edge_key = InternedKey::from_str(&edge_type);
    let root = resolve_outline_root(graph, root_id, root_type.as_deref())?;

    let mut rows = Vec::new();
    let mut visited: HashSet<NodeIndex> = HashSet::new();
    let mut queue: VecDeque<(NodeIndex, usize, Option<NodeIndex>)> = VecDeque::new();
    visited.insert(root);
    queue.push_back((root, 0, None));
    while let Some((nidx, depth, parent)) = queue.pop_front() {
        rows.push(outline_result_row(graph, yield_items, nidx, depth, parent)?);
        if depth < max_depth {
            for er in graph.graph.edges_directed(nidx, Direction::Outgoing) {
                if er.weight().connection_type == edge_key {
                    let child = er.target();
                    if visited.insert(child) {
                        queue.push_back((child, depth + 1, Some(nidx)));
                    }
                }
            }
        }
    }
    Ok(rows)
}

fn outline_max_depth(params: &HashMap<String, Value>) -> Result<usize, String> {
    match params.get("max_depth") {
        Some(Value::Int64(n)) if *n >= 0 => Ok(*n as usize),
        Some(Value::Int64(_)) => Err("CALL outline: 'max_depth' must be non-negative".to_string()),
        Some(other) => Err(format!(
            "CALL outline: 'max_depth' must be an integer, got {}",
            other.type_name()
        )),
        None => Ok(usize::MAX),
    }
}

fn resolve_outline_root(
    graph: &DirGraph,
    root_id: &Value,
    root_type: Option<&str>,
) -> Result<NodeIndex, String> {
    let types: Vec<&str> = match root_type {
        Some(node_type) => vec![node_type],
        None => graph.type_indices.keys().collect(),
    };
    let mut roots = Vec::new();
    for node_type in types {
        match graph.outline_id_candidates(node_type, root_id) {
            Some(mut candidates) => roots.append(&mut candidates),
            None => {
                let Some(members) = graph.type_indices.get(node_type) else {
                    continue;
                };
                roots.extend(members.iter().filter(|&index| {
                    graph
                        .graph
                        .node_view(index)
                        .is_some_and(|node| outline_root_matches(node.id().as_ref(), root_id))
                }));
            }
        }
        if roots.len() > 1 {
            break;
        }
    }
    if roots.is_empty() {
        return Err(format!(
            "CALL outline: no node with id {}{}",
            crate::datatypes::values::raw_string(root_id),
            root_type
                .map(|kind| format!(" and root_type '{kind}'"))
                .unwrap_or_default()
        ));
    }
    if roots.len() > 1 {
        return Err(if let Some(kind) = root_type {
            format!(
                "CALL outline: root {} with root_type '{kind}' is ambiguous",
                crate::datatypes::values::raw_string(root_id),
            )
        } else {
            format!(
                "CALL outline: root {} is ambiguous; pass root_type to select one node",
                crate::datatypes::values::raw_string(root_id),
            )
        });
    }
    Ok(roots[0])
}

struct OutlineIdentity {
    id: Option<Value>,
    node_type: Option<Value>,
    id_type: Option<Value>,
}

fn outline_identity(
    graph: &DirGraph,
    nidx: NodeIndex,
    role: &str,
    need_id: bool,
    need_node_type: bool,
    need_id_type: bool,
) -> Result<OutlineIdentity, String> {
    let node = graph
        .graph
        .node_view(nidx)
        .ok_or_else(|| format!("CALL outline: {role} disappeared during traversal"))?;
    let id = node.id();
    Ok(OutlineIdentity {
        id_type: need_id_type.then(|| Value::String(id.type_name().to_string())),
        id: need_id.then(|| id.into_owned()),
        node_type: need_node_type
            .then(|| Value::String(node.node_type_str(&graph.interner).to_string())),
    })
}

fn outline_result_row(
    graph: &DirGraph,
    yield_items: &[YieldItem],
    nidx: NodeIndex,
    depth: usize,
    parent_idx: Option<NodeIndex>,
) -> Result<ResultRow, String> {
    let needs_node_type = yield_items.iter().any(|item| item.name == "node_type");
    let needs_node_id_type = yield_items.iter().any(|item| item.name == "node_id_type");
    let needs_parent_id = yield_items.iter().any(|item| item.name == "parent_id");
    let needs_parent_type = yield_items.iter().any(|item| item.name == "parent_type");
    let needs_parent_id_type = yield_items.iter().any(|item| item.name == "parent_id_type");
    let node = (needs_node_type || needs_node_id_type)
        .then(|| {
            outline_identity(
                graph,
                nidx,
                "node",
                false,
                needs_node_type,
                needs_node_id_type,
            )
        })
        .transpose()?;
    let parent =
        if parent_idx.is_some() && (needs_parent_id || needs_parent_type || needs_parent_id_type) {
            parent_idx
                .map(|index| {
                    outline_identity(
                        graph,
                        index,
                        "parent",
                        needs_parent_id,
                        needs_parent_type,
                        needs_parent_id_type,
                    )
                })
                .transpose()?
        } else {
            None
        };
    let mut row = ResultRow::new();
    for item in yield_items {
        let alias = item.alias.as_deref().unwrap_or(&item.name);
        let value = match item.name.as_str() {
            "node" => {
                row.node_bindings.insert(alias.to_string(), nidx);
                continue;
            }
            "depth" => Value::Int64(depth as i64),
            "parent_id" => parent
                .as_ref()
                .and_then(|identity| identity.id.clone())
                .unwrap_or(Value::Null),
            "node_type" => node
                .as_ref()
                .and_then(|identity| identity.node_type.clone())
                .unwrap_or(Value::Null),
            "node_id_type" => node
                .as_ref()
                .and_then(|identity| identity.id_type.clone())
                .unwrap_or(Value::Null),
            "parent_type" => parent
                .as_ref()
                .and_then(|identity| identity.node_type.clone())
                .unwrap_or(Value::Null),
            "parent_id_type" => parent
                .as_ref()
                .and_then(|identity| identity.id_type.clone())
                .unwrap_or(Value::Null),
            "node_token" => Value::Int64(nidx.index() as i64),
            "parent_token" => {
                parent_idx.map_or(Value::Null, |index| Value::Int64(index.index() as i64))
            }
            _ => continue,
        };
        row.projected.insert(alias.to_string(), value);
    }
    Ok(row)
}

fn outline_root_matches(candidate: &Value, requested: &Value) -> bool {
    candidate == requested
        || matches!(
            (candidate, requested),
            (Value::UniqueId(left), Value::Int64(right))
                if *right >= 0 && u32::try_from(*right) == Ok(*left)
        )
}

fn title_of(graph: &DirGraph, nidx: NodeIndex) -> Option<String> {
    let nd = graph.graph.node_view(nidx)?;
    match nd.title().as_ref() {
        Value::String(s) => Some(s.clone()),
        Value::Null => None,
        other => Some(format!("{other:?}")),
    }
}

fn id_of(graph: &DirGraph, nidx: NodeIndex) -> Option<String> {
    let nd = graph.graph.node_view(nidx)?;
    match nd.id().as_ref() {
        Value::String(s) => Some(s.clone()),
        Value::Null => None,
        other => Some(format!("{other:?}")),
    }
}

const OUTBOUND: bool = true;
const INBOUND: bool = false;

fn require_string_param(
    params: &HashMap<String, Value>,
    key: &str,
    proc: &str,
) -> Result<String, String> {
    match params.get(key) {
        Some(Value::String(s)) => Ok(s.clone()),
        Some(other) => Err(format!(
            "CALL {proc}: parameter '{key}' must be a string, got {other:?}"
        )),
        None => Err(format!(
            "CALL {proc}: missing required parameter '{key}'. \
             Use map syntax — e.g. CALL {proc}({{{key}: 'X'}}).{}",
            param_schema_hint(proc)
        )),
    }
}

/// `Err` rather than the default when the param is present but not a string,
/// so misuse surfaces loudly.
fn optional_string_param(
    params: &HashMap<String, Value>,
    key: &str,
    proc: &str,
) -> Result<Option<String>, String> {
    match params.get(key) {
        Some(Value::String(s)) => Ok(Some(s.clone())),
        Some(other) => Err(format!(
            "CALL {proc}: parameter '{key}' must be a string, got {other:?}"
        )),
        None => Ok(None),
    }
}

pub(super) fn require_node_yield(
    yield_items: &[YieldItem],
    proc: &str,
    expected: &str,
) -> Result<String, String> {
    if let Some(item) = yield_items.iter().find(|y| y.name == expected) {
        // Use alias when present so the binding is reachable downstream
        // by the name the agent picked.
        Ok(item.alias.clone().unwrap_or_else(|| expected.to_string()))
    } else {
        Err(format!(
            "CALL {proc}: must YIELD '{expected}'. Got {got:?}.",
            got = yield_items
                .iter()
                .map(|y| y.name.as_str())
                .collect::<Vec<_>>(),
        ))
    }
}

fn require_two_node_yields(
    yield_items: &[YieldItem],
    proc: &str,
    a: &str,
    b: &str,
) -> Result<(String, String), String> {
    let av = require_node_yield(yield_items, proc, a)?;
    let bv = require_node_yield(yield_items, proc, b)?;
    Ok((av, bv))
}

/// The [`require_node_yield`] lookup, for yields that flow into
/// `row.projected` rather than `row.node_bindings` (`count`, `distance_m`).
fn require_scalar_yield(
    yield_items: &[YieldItem],
    proc: &str,
    expected: &str,
) -> Result<String, String> {
    require_node_yield(yield_items, proc, expected)
}

fn call_param_i64(params: &HashMap<String, Value>, key: &str, default: i64) -> i64 {
    match params.get(key) {
        Some(Value::Int64(v)) => *v,
        Some(Value::Float64(v)) => *v as i64,
        _ => default,
    }
}

fn call_param_opt_i64(params: &HashMap<String, Value>, key: &str) -> Option<i64> {
    match params.get(key) {
        Some(Value::Int64(v)) => Some(*v),
        Some(Value::Float64(v)) => Some(*v as i64),
        _ => None,
    }
}

pub(super) fn type_indices<'a>(
    graph: &'a DirGraph,
    node_type: &str,
) -> Result<crate::graph::storage::disk::type_index::TypeNodesRef<'a>, String> {
    graph
        .type_indices
        .get(node_type)
        .ok_or_else(|| format!("Type '{node_type}' has no nodes in this graph"))
}

pub(super) fn make_node_row(yield_var: &str, nidx: NodeIndex) -> ResultRow {
    let mut row = ResultRow::new();
    row.node_bindings.insert(yield_var.to_string(), nidx);
    row
}

fn validate_direction(
    graph: &DirGraph,
    node_type: &str,
    edge_type: &str,
    outbound: bool,
    proc: &str,
) -> Result<(), String> {
    let info = match graph.connection_type_metadata.get(edge_type) {
        Some(i) => i,
        None => return Ok(()), // edge unknown → graph might be partial; trust the author
    };
    let on_correct_side = if outbound {
        info.source_types.contains(node_type)
    } else {
        info.target_types.contains(node_type)
    };
    if on_correct_side || info.source_types.is_empty() || info.target_types.is_empty() {
        // Allow when empty source/target_types — pre-rename graphs may not
        // have populated metadata even though the edge is real.
        return Ok(());
    }

    let mut sources: Vec<&String> = info.source_types.iter().collect();
    let mut targets: Vec<&String> = info.target_types.iter().collect();
    sources.sort();
    targets.sort();

    if outbound {
        if info.target_types.contains(node_type) {
            Err(format!(
                "DirectionMismatch in CALL {proc}: '{edge_type}' flows {sources:?} → \
                 {targets:?} — '{node_type}' is on the target side. \
                 Use missing_inbound_edge with the same parameters instead."
            ))
        } else {
            Err(format!(
                "DirectionMismatch in CALL {proc}: '{edge_type}' flows {sources:?} → \
                 {targets:?} — '{node_type}' isn't a source type for this edge. \
                 Set type to one of {sources:?}, or pick a different edge."
            ))
        }
    } else if info.source_types.contains(node_type) {
        Err(format!(
            "DirectionMismatch in CALL {proc}: '{edge_type}' flows {sources:?} → \
             {targets:?} — '{node_type}' is on the source side. \
             Use missing_required_edge with the same parameters instead."
        ))
    } else {
        Err(format!(
            "DirectionMismatch in CALL {proc}: '{edge_type}' flows {sources:?} → \
             {targets:?} — '{node_type}' isn't a target type for this edge. \
             Set type to one of {targets:?}, or pick a different edge."
        ))
    }
}