kglite 0.13.2

//! Cypher pipeline orchestration — single source of truth.
//!
//! Mirrors the canonical pipeline that previously lived inline at
//! `src/graph/pyapi/kg_core.rs::cypher`:
//!
//! ```text
//! parse_cypher → validate_schema → rewrite_text_score (+embed if needed)
//!   → optimize_with_disabled → [mark_lazy_eligibility] → is_mutation_query
//!   → generate_explain_result | execute | execute_mutable
//! ```
//!
//! [`execute_read`] takes `&DirGraph` (auto-commit reads + in-tx reads
//! against working/snapshot). [`execute_mut`] takes `&mut DirGraph`
//! (in-tx writes against `Transaction::working_mut()`).

use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::time::Instant;

use crate::datatypes::Value;
use crate::error::KgError;
use crate::graph::dir_graph::DirGraph;
use crate::graph::embedder::Embedder;
use crate::graph::languages::cypher;
use crate::graph::languages::cypher::ast::{
    Clause, CreateElement, CreatePattern, CypherQuery, OutputFormat, RemoveItem, SetItem,
};
use crate::graph::languages::cypher::result::CypherResult;
use crate::graph::languages::cypher::value_codec::ValueCodec;

/// Per-query knobs. Borrowed for the duration of one execute call.
/// Default values match the kg_core.rs Python boundary's defaults
/// (lazy_eligible=true, no deadline, no max_rows, no disabled passes,
/// no embedder).
pub struct ExecuteOptions<'a> {
    /// Parameter bindings (`$x` references). Empty map = no params.
    pub params: &'a HashMap<String, Value>,
    /// Optional execution deadline. Past this, the executor returns
    /// `CypherTimeout`. None = no deadline.
    pub deadline: Option<Instant>,
    /// Optional row cap. None = no cap.
    pub max_rows: Option<usize>,
    /// Lazy-projection mode.
    ///
    /// - `true` (Python default): call `mark_lazy_eligibility` after
    ///   optimize + pass `streaming=true` to the executor. The
    ///   `CypherResult.lazy` field may be `Some(LazyResultDescriptor)`;
    ///   callers that want eager rows must materialize via the lazy
    ///   helper in `src/graph/pyapi/result_view.rs`.
    /// - `false` (bolt-server, mcp-server): skip
    ///   `mark_lazy_eligibility` + pass `streaming=false`. The
    ///   executor materializes every row into `CypherResult.rows`.
    ///
    /// **Important:** setting `lazy_eligible=true` without having a
    /// lazy-materializer to consume `result.lazy` results in
    /// silently empty row sets — exactly the bolt-server bug fixed
    /// during the robustness pass. Default to `false` for safety;
    /// the Python boundary flips it to `true` to benefit from the
    /// lazy path in interactive use.
    pub lazy_eligible: bool,
    /// Optional set of planner passes to disable. None means "use
    /// the static empty set" (no allocation; the common case).
    pub disabled_passes: Option<&'a HashSet<String>>,
    /// Optional embedder for `text_score()` queries. If a query
    /// uses `text_score()` and this is `None`, execute returns
    /// `KgError::Argument("text_score requires embedder ...")`.
    pub embedder: Option<Arc<dyn Embedder>>,
    /// Optional operator-declared value codecs. When set, query-side
    /// literals bound to a codec'd property are decoded before
    /// validation/optimization (`'Q42'` → `42`), and result columns
    /// that are direct projections of a codec'd property are encoded
    /// back (`42` → `'Q42'`). `None`/empty = no transform (the common
    /// case; zero hot-path cost). See `cypher::value_codec`.
    pub value_codecs: Option<&'a [ValueCodec]>,
    /// Optional cooperative-cancellation flag. The executor and pattern
    /// matcher poll it at the same checkpoints they poll `deadline`
    /// (one relaxed atomic load per ~4K comparisons); once set, the
    /// run aborts with [`KgError::Cancelled`]. `None` = never cancelled
    /// (zero hot-path cost). This is the engine-agnostic primitive each
    /// binding flips from its own signal model — the Python wheel wires
    /// it to a scoped SIGINT handler so Ctrl-C interrupts long queries;
    /// servers leave it `None` and use their own deadline/teardown.
    ///
    /// A `&'static` flag (not an owned `Arc`) because the only setter is
    /// a process-global signal handler, which can't capture state — the
    /// Python wheel points this at a `static AtomicBool` its SIGINT
    /// handler flips. Bindings that need this provide a `'static` flag;
    /// the rest pass `None`.
    pub cancel: Option<&'static AtomicBool>,
    /// Optional role-scoped write whitelist. When `Some`, a Cypher
    /// `CREATE`/`SET` whose node type is not in the set is rejected (integrity,
    /// not secrecy — e.g. a coding role may write `Plan`/`Task` but not
    /// `Algorithm`). `None` = unrestricted (the default; zero hot-path cost).
    /// Only meaningful on the mutation path (`execute_mut`).
    pub write_scope: Option<&'a HashSet<String>>,
    /// Caller-supplied freshness provenance, stamped alongside `updated_at` on
    /// writes to `auto_timestamp` types: the git SHA the writer is working
    /// against and an actor id. `None` = not supplied. Mutation path only.
    pub git_sha: Option<&'a str>,
    pub modified_by: Option<&'a str>,
}

impl<'a> ExecuteOptions<'a> {
    /// Conservative defaults: `lazy_eligible: false` (safe for
    /// every consumer that doesn't have a lazy materializer), no
    /// deadline, no max_rows, no disabled passes, no embedder.
    /// Caller is expected to override at least `params`.
    ///
    /// Same as [`Self::eager`] — the two are synonyms. `new` is
    /// kept for Rust-convention API discovery; `eager` is the
    /// intent-named factory call-sites prefer.
    pub fn new(params: &'a HashMap<String, Value>) -> Self {
        Self::eager(params)
    }

    /// Eager-execution defaults — the safe default for any binding
    /// that doesn't have a lazy result materializer.
    ///
    /// This is the constructor non-Python bindings should reach for:
    /// `lazy_eligible: false`, no deadline, no max_rows, no disabled
    /// passes, no embedder. Override individual fields after
    /// construction if needed (deadline for timeouts, embedder when
    /// `text_score()` queries are expected).
    ///
    /// Lifted in 2026-05-25 to give the call-site the intent-named
    /// shape — previously mcp-server / bolt-server constructed the
    /// struct manually with identical defaults; now they call
    /// `ExecuteOptions::eager(params)` for self-documenting code.
    pub fn eager(params: &'a HashMap<String, Value>) -> Self {
        Self {
            params,
            deadline: None,
            max_rows: None,
            lazy_eligible: false,
            disabled_passes: None,
            embedder: None,
            value_codecs: None,
            cancel: None,
            write_scope: None,
            git_sha: None,
            modified_by: None,
        }
    }
}

/// Map an executor error string to a typed [`KgError`]. When the
/// caller's cooperative-cancellation flag is set, an aborted run is
/// reported as [`KgError::Cancelled`] (the binding maps that to its
/// interrupt type — `KeyboardInterrupt` in the Python wheel) rather
/// than a misleading `CypherExecution`. Otherwise it's a plain
/// execution error. `cancel == None` (every server binding) always
/// takes the `CypherExecution` branch, so behaviour is unchanged there.
#[inline]
fn exec_err(opts: &ExecuteOptions<'_>, message: String) -> KgError {
    if opts
        .cancel
        .is_some_and(|c| c.load(std::sync::atomic::Ordering::Relaxed))
    {
        KgError::Cancelled
    } else {
        KgError::CypherExecution {
            message,
            position: None,
        }
    }
}

/// Result of a successful execute. Wraps `CypherResult` with the
/// metadata callers need for output serialization (CSV, DataFrame,
/// PackStream record emission).
pub struct ExecuteOutcome {
    pub result: CypherResult,
    /// `true` when the query was a CREATE/SET/DELETE/REMOVE/MERGE.
    /// Read-only callers can pre-reject by checking this on a
    /// dry-run; in practice `execute_read` rejects mutations
    /// upfront via `KgError::Argument`.
    pub is_mutation: bool,
    /// Set when the user passes `RETURN ... FORMAT CSV` (kglite
    /// extension); pyapi + mcp-server format the result accordingly.
    pub output_format: OutputFormat,
    /// Set when the user prefixed the query with `EXPLAIN`. The
    /// `result` contains the rendered plan rows rather than real
    /// data; callers may want to format / display differently.
    pub explain: bool,
}

/// Read-only execution. Errors if the query mutates.
///
/// Caller responsibilities:
/// - Provide a `&DirGraph` (snapshot for auto-commit, or
///   `tx.current()` for in-tx reads).
/// - Decode params (`Bolt`/`Py` → `Value`) before calling.
/// - Map the returned `KgError` to the binding's error type
///   (PyErr subclass via `From`, `BoltError` via the
///   `kg_to_bolt`/`string_to_bolt` helpers in bolt-server).
pub fn execute_read(
    graph: &DirGraph,
    query: &str,
    opts: &ExecuteOptions<'_>,
) -> Result<ExecuteOutcome, KgError> {
    let (parsed, params, encode_plan) = prepare(graph, query, opts)?;
    let is_mutation = cypher::is_mutation_query(&parsed);

    // EXPLAIN: render plan rows, skip execution.
    if parsed.explain {
        let result = cypher::generate_explain_result(&parsed, graph);
        return Ok(ExecuteOutcome {
            result,
            is_mutation,
            output_format: parsed.output_format,
            explain: true,
        });
    }

    if is_mutation {
        return Err(KgError::Argument(
            "execute_read called with a mutation query (CREATE/SET/DELETE/REMOVE/MERGE) \
             — use execute_mut against a mutable graph view"
                .to_string(),
        ));
    }

    let mut result = cypher::CypherExecutor::with_params(graph, &params, opts.deadline)
        .with_max_rows(opts.max_rows)
        .with_streaming(opts.lazy_eligible)
        .with_cancel(opts.cancel)
        .execute(&parsed)
        .map_err(|message| exec_err(opts, message))?;
    // value_codecs: encode codec'd-property result columns back to the typed
    // form (`42` → `'Q42'`). Applies to eager rows; lazy results (Python's
    // streaming path) materialize later and aren't covered — the configured
    // consumer (mcp-server) runs eager.
    cypher::value_codec::apply_encode(&mut result, &encode_plan);

    Ok(ExecuteOutcome {
        result,
        is_mutation: false,
        output_format: parsed.output_format,
        explain: false,
    })
}

/// Whether this statement has no fallible operation after its first write.
///
/// This is intentionally a proof whitelist, not a general optimiser:
///
/// - one standalone node `CREATE` evaluates properties, validates scope/schema,
///   and checks primary-key uniqueness before inserting its only node;
/// - a terminal variable-only `DELETE` collects bindings and validates plain
///   delete edge constraints before removing anything.
///
/// Both shapes are safe only on the default in-memory backend and without an
/// execution budget, which is checked after a write. Deadline/cancellation is
/// safe: CREATE polls before insertion, while DELETE completes every poll and
/// validation in its preflight phase before entering a non-interruptible commit
/// phase. Every other mutation retains the full rollback checkpoint.
fn can_skip_rollback_checkpoint(
    graph: &DirGraph,
    query: &CypherQuery,
    opts: &ExecuteOptions<'_>,
) -> bool {
    if !graph.graph.supports_checkpoint_free_mutation() || query.profile || opts.max_rows.is_some()
    {
        return false;
    }

    match query.clauses.as_slice() {
        [Clause::Create(create)] => matches!(
            create.patterns.as_slice(),
            [pattern] if matches!(pattern.elements.as_slice(), [CreateElement::Node(_)])
        ),
        clauses => {
            let Some((Clause::Delete(delete), prefix)) = clauses.split_last() else {
                return false;
            };
            delete
                .expressions
                .iter()
                .all(|expr| matches!(expr, cypher::ast::Expression::Variable(_)))
                && prefix
                    .iter()
                    .all(|clause| !cypher::executor::write::clause_is_mutation(clause))
        }
    }
}

/// Mutating execution. Caller passes `&mut DirGraph` (typically
/// from `Transaction::working_mut()`). For pure reads, use
/// [`execute_read`] instead.
///
/// Note: a read query passed to `execute_mut` runs against the
/// mutable graph view as a read. The function returns
/// `is_mutation: false` in that case so the caller knows nothing
/// was changed.
pub fn execute_mut(
    graph: &mut DirGraph,
    query: &str,
    opts: &ExecuteOptions<'_>,
) -> Result<ExecuteOutcome, KgError> {
    let (parsed, params, encode_plan) = prepare(graph, query, opts)?;
    let is_mutation = cypher::is_mutation_query(&parsed);

    // EXPLAIN never executes the mutation. Return before collision preflight,
    // disk promotion, or an atomic rollback checkpoint so inspecting a write
    // plan remains a read-only, O(plan) operation.
    if parsed.explain {
        let result = cypher::generate_explain_result(&parsed, graph);
        return Ok(ExecuteOutcome {
            result,
            is_mutation,
            output_format: parsed.output_format,
            explain: true,
        });
    }

    if is_mutation {
        let mut names = Vec::new();
        collect_mutation_names(&parsed, &mut names);
        graph
            .interner
            .validate_names(names)
            .map_err(KgError::from)?;
    }

    // A statement is atomic even when the caller supplied an already-
    // materialized transaction working copy. Keep a checkpoint whenever the
    // executor can still fail after its first write. A deliberately narrow
    // in-memory fast path covers two shapes whose executors finish all
    // fallible work before mutating; see `can_skip_rollback_checkpoint`.
    let rollback_checkpoint = (is_mutation && !can_skip_rollback_checkpoint(graph, &parsed, opts))
        .then(|| graph.fork_transaction());

    if is_mutation {
        if let Err(error) = graph.prepare_disk_mutation() {
            if let Some(checkpoint) = rollback_checkpoint {
                *graph = checkpoint;
            }
            return Err(KgError::FileIo(error));
        }
    }

    let mut result = if is_mutation {
        let interrupt = crate::graph::algorithms::Interrupt {
            deadline: opts.deadline,
            cancel: opts.cancel,
        };
        // Install the execution-scoped write whitelist for the duration of this
        // mutation, then clear it unconditionally (even on error) so it never
        // leaks into a later execution on the same working copy.
        graph.active_write_scope = opts.write_scope.cloned();
        graph.active_git_sha = opts.git_sha.map(str::to_string);
        graph.active_modified_by = opts.modified_by.map(str::to_string);
        let r = if opts.max_rows.is_some() {
            cypher::executor::write::execute_mutable_bounded(
                graph,
                &parsed,
                params,
                interrupt,
                opts.max_rows,
            )
        } else {
            cypher::execute_mutable(graph, &parsed, params, interrupt)
        };
        graph.active_write_scope = None;
        graph.active_git_sha = None;
        graph.active_modified_by = None;
        let r = match r {
            Ok(result) => result,
            Err(message) => {
                if let Some(checkpoint) = rollback_checkpoint {
                    *graph = checkpoint;
                }
                return Err(exec_err(opts, message));
            }
        };
        // A Cypher write occurred — advance the graph version so any
        // version-keyed caches (the plan cache) and OCC see the change.
        // Bumps the working copy directly so a read-after-write *within* the
        // same transaction re-plans against the mutated state; the eventual
        // commit recomputes the live version independently (see Session::commit).
        graph.bump_version();
        r
    } else {
        cypher::CypherExecutor::with_params(graph, &params, opts.deadline)
            .with_max_rows(opts.max_rows)
            .with_streaming(opts.lazy_eligible)
            .with_cancel(opts.cancel)
            .execute(&parsed)
            .map_err(|message| exec_err(opts, message))?
    };
    // Encode codec'd-property result columns (e.g. `CREATE (...) RETURN n.id`
    // reads back `'Q42'`). Eager path only; see execute_read.
    cypher::value_codec::apply_encode(&mut result, &encode_plan);

    Ok(ExecuteOutcome {
        result,
        is_mutation,
        output_format: parsed.output_format,
        explain: false,
    })
}

fn collect_pattern_names<'a>(pattern: &'a CreatePattern, out: &mut Vec<&'a str>) {
    for element in &pattern.elements {
        match element {
            CreateElement::Node(node) => {
                out.extend(node.label.as_deref());
                out.extend(node.extra_labels.iter().map(String::as_str));
                out.extend(node.properties.iter().map(|(name, _)| name.as_str()));
            }
            CreateElement::Edge(edge) => {
                out.push(edge.connection_type.as_str());
                out.extend(edge.properties.iter().map(|(name, _)| name.as_str()));
            }
        }
    }
}

fn collect_set_names<'a>(items: &'a [SetItem], out: &mut Vec<&'a str>) {
    for item in items {
        match item {
            SetItem::Property { property, .. } => out.push(property),
            SetItem::Label { label, .. } => out.push(label),
            SetItem::Map { .. } => {}
        }
    }
}

fn collect_mutation_names<'a>(query: &'a CypherQuery, out: &mut Vec<&'a str>) {
    collect_clause_names(&query.clauses, out);
}

fn collect_clause_names<'a>(clauses: &'a [Clause], out: &mut Vec<&'a str>) {
    for clause in clauses {
        match clause {
            Clause::Create(create) => {
                for pattern in &create.patterns {
                    collect_pattern_names(pattern, out);
                }
            }
            Clause::Set(set) => collect_set_names(&set.items, out),
            Clause::Remove(remove) => {
                for item in &remove.items {
                    match item {
                        RemoveItem::Property { property, .. } => out.push(property),
                        RemoveItem::Label { label, .. } => out.push(label),
                    }
                }
            }
            Clause::Merge(merge) => {
                collect_pattern_names(&merge.pattern, out);
                if let Some(items) = &merge.on_create {
                    collect_set_names(items, out);
                }
                if let Some(items) = &merge.on_match {
                    collect_set_names(items, out);
                }
            }
            Clause::Foreach { body, .. } => collect_clause_names(body, out),
            Clause::CallSubquery { body, .. } => collect_mutation_names(body, out),
            Clause::Union(union) => collect_mutation_names(&union.query, out),
            _ => {}
        }
    }
}

/// Shared preparation: parse → validate → rewrite_text_score → embed
/// (if needed) → optimize → optional mark_lazy. Returns the
/// parsed+optimized AST + the (possibly-augmented-with-embeddings)
/// param map.
///
/// The params map is borrowed from `opts.params` in the common case
/// (no text_score). When text_score() is present, we clone-on-write
/// to inject the embedding result vectors into the map — the
/// returned `HashMap<String, Value>` is owned in that case.
///
/// **GIL note for binding implementers.** If `opts.embedder` is a
/// Python-backed embedder (PyEmbedderAdapter), the binding MUST
/// release the GIL before calling `execute_read`/`execute_mut`
/// (Python's `py.detach`). The embed call inside this fn will then
/// re-acquire the GIL briefly to invoke Python; if you forget to
/// release first, it deadlocks.
/// Output of [`prepare`]: the parsed+optimized query, the (possibly
/// embedding-augmented) param map, and the column-indexed value-codec encode
/// plan (empty when no codecs apply).
type PreparedQuery = (
    Arc<CypherQuery>,
    HashMap<String, Value>,
    Vec<Option<ValueCodec>>,
);

fn prepare(
    graph: &DirGraph,
    query: &str,
    opts: &ExecuteOptions<'_>,
) -> Result<PreparedQuery, KgError> {
    // Plan cache: a param-less, codec-free, no-disabled-passes query against an
    // unchanged graph reuses its fully-optimized plan, skipping parse + validate
    // + optimize. Keyed on (graph_id, version) so any mutation invalidates it
    // and it never leaks across graphs (see `cypher::plan_cache`). Lazy-marking
    // is applied fresh per call since it depends on `opts.lazy_eligible`.
    let cacheable = opts.params.is_empty()
        && opts.disabled_passes.is_none_or(|s| s.is_empty())
        && opts.value_codecs.is_none_or(|c| c.is_empty());
    if cacheable {
        if let Some(plan) =
            cypher::plan_cache::get(graph.graph_id(), graph.version(), opts.lazy_eligible, query)
        {
            // Stored post lazy-marking for this `lazy_eligible` — a hit is a
            // pure Arc clone, no parse / validate / optimize / mutation.
            return Ok((plan, HashMap::new(), Vec::new()));
        }
    }

    let mut parsed = cypher::parse_cypher(query)?;

    // value_codecs: decode operator-declared literals bound to a codec'd
    // property (`{id:'Q42'}` / `WHERE n.id = 'Q42'` → `42`) BEFORE anything
    // else, so validation, optimization, and execution all treat the decoded
    // form as canonical. No-op (one is_empty check) when none are configured.
    let codecs = opts.value_codecs.unwrap_or(&[]);
    cypher::value_codec::apply_decode(&mut parsed, codecs);
    // Build the result-side encode plan now, while the RETURN clause is a clean
    // pre-optimize projection (fusion later rewrites *how* columns are computed,
    // not the output schema). Column-indexed; empty when no codecs / no RETURN.
    let encode_plan = cypher::value_codec::build_encode_plan(&parsed, codecs);

    // Schema validation — property typos in pattern literals
    // (`{ttle: 'Alice'}`) get caught with a "did you mean?" hint.
    cypher::validate_schema(&parsed, graph).map_err(KgError::from)?;

    // Non-fatal: warn (stderr) when a MATCH references an unknown node label
    // or relationship type — the most common "why is my query empty?" typo.
    cypher::warn_unknown_pattern_refs(&parsed, graph);

    // text_score() rewrite. Scans for `text_score(...)` calls in the
    // AST and rewrites them to `vector_score(...)`, collecting the
    // texts to embed alongside.
    let rewrite = cypher::rewrite_text_score(&mut parsed, opts.params).map_err(|message| {
        KgError::CypherExecution {
            message,
            position: None,
        }
    })?;

    // If text_score(...) was used (and we're NOT in EXPLAIN mode —
    // EXPLAIN renders plan rows without executing, so no embedding
    // needed), run the embedder and inject the result vectors into
    // the param map. Otherwise pass the caller's params through.
    let params: Cow<'_, HashMap<String, Value>> =
        if !rewrite.texts_to_embed.is_empty() && !parsed.explain {
            Cow::Owned(embed_into_params(opts, &rewrite)?)
        } else {
            Cow::Borrowed(opts.params)
        };

    // Optimize. Empty disabled-set is the common case; avoid the
    // HashSet allocation when no passes are disabled.
    let disabled_default = cypher::planner::empty_disabled_set();
    let disabled_ref = opts.disabled_passes.unwrap_or(disabled_default);
    cypher::planner::optimize_with_disabled(&mut parsed, graph, &params, disabled_ref);

    // Lazy marking — only when the caller asked for it. Done BEFORE caching so
    // the cached plan is ready-to-execute for this `lazy_eligible` (the cache
    // key includes it), making hits a pure Arc clone. Without this the executor
    // materializes rows eagerly; with it, `result.lazy` may be Some and
    // `result.rows` empty and the caller must materialize (Python's ResultView
    // does; bolt-server doesn't, so it passes `lazy_eligible: false`).
    if opts.lazy_eligible {
        cypher::mark_lazy_eligibility(&mut parsed);
    }

    let plan = Arc::new(parsed);
    // Cache the ready-to-execute plan. Only when `params` stayed empty — a
    // `text_score()` rewrite injects embedding params, making the plan
    // call-specific, so those are never cached (and thus never hit above).
    if cacheable && params.is_empty() {
        cypher::plan_cache::insert(
            graph.graph_id(),
            graph.version(),
            opts.lazy_eligible,
            query,
            plan.clone(),
        );
    }

    Ok((plan, params.into_owned(), encode_plan))
}

/// Run the embedder on collected texts; inject the JSON-encoded
/// vectors into a clone of the param map. Caller-supplied params
/// are not mutated. Returns the augmented map.
fn embed_into_params(
    opts: &ExecuteOptions<'_>,
    rewrite: &cypher::planner::simplification::TextScoreRewrite,
) -> Result<HashMap<String, Value>, KgError> {
    let model = opts
        .embedder
        .as_ref()
        .ok_or_else(|| KgError::CypherExecution {
            message: "text_score() requires a registered embedding model. \
                      Call g.set_embedder(model) first (Python) or pass an embedder \
                      via ExecuteOptions::embedder (downstream Rust consumers)."
                .to_string(),
            position: None,
        })?;
    model.load().map_err(|message| KgError::CypherExecution {
        message,
        position: None,
    })?;
    let texts: Vec<String> = rewrite
        .texts_to_embed
        .iter()
        .map(|(_, t)| t.clone())
        .collect();
    let embed_result = model.embed(&texts);
    model.unload();
    let embeddings: Vec<Vec<f32>> = embed_result.map_err(|message| KgError::CypherExecution {
        message,
        position: None,
    })?;
    if embeddings.len() != texts.len() {
        return Err(KgError::CypherExecution {
            message: format!(
                "text_score: model.embed() returned {} vectors for {} texts",
                embeddings.len(),
                texts.len()
            ),
            position: None,
        });
    }
    let mut params = opts.params.clone();
    for (i, (param_name, _)) in rewrite.texts_to_embed.iter().enumerate() {
        let json = format!(
            "[{}]",
            embeddings[i]
                .iter()
                .map(|f| f.to_string())
                .collect::<Vec<_>>()
                .join(", ")
        );
        params.insert(param_name.clone(), Value::String(json));
    }
    Ok(params)
}

#[cfg(test)]
mod version_soundness_tests {
    use super::*;
    use crate::graph::dir_graph::DirGraph;
    use crate::graph::storage::GraphRead;

    /// A Cypher write through `execute_mut` must advance the graph version so
    /// version-keyed caches (the plan cache) and a read-after-write within the
    /// same transaction observe the change.
    #[test]
    fn execute_mut_write_bumps_version() {
        let mut g = DirGraph::new();
        let params = HashMap::new();
        let opts = ExecuteOptions::eager(&params);
        let before = g.version();
        execute_mut(&mut g, "CREATE (:Item {id: 1})", &opts).expect("create");
        assert!(
            g.version() > before,
            "a Cypher write must bump version (was {before}, now {})",
            g.version()
        );
    }

    /// A read must NOT bump the version — otherwise repeated reads would
    /// perpetually invalidate the plan cache.
    #[test]
    fn execute_read_does_not_bump_version() {
        let mut g = DirGraph::new();
        let params = HashMap::new();
        let opts = ExecuteOptions::eager(&params);
        execute_mut(&mut g, "CREATE (:Item {id: 1})", &opts).expect("create");
        let after_write = g.version();
        let _ = execute_read(&g, "MATCH (n:Item) RETURN n.id", &opts).expect("read");
        assert_eq!(g.version(), after_write, "a read must not bump version");
    }

    #[test]
    fn cypher_collision_is_typed_and_atomic() {
        let mut g = DirGraph::new();
        let incoming = "CollisionType";
        g.interner
            .try_register(
                crate::graph::schema::InternedKey::from_str(incoming),
                "conflicting-existing",
            )
            .unwrap();
        let params = HashMap::new();
        let opts = ExecuteOptions::eager(&params);
        let error = match execute_mut(&mut g, "CREATE (:CollisionType {id: 1})", &opts) {
            Err(error) => error,
            Ok(_) => panic!("colliding Cypher name must be rejected"),
        };
        assert!(matches!(error, KgError::InternerCollision(_)));
        assert_eq!(g.graph.node_count(), 0);
        assert_eq!(g.version(), 0);
    }

    #[test]
    fn checkpointed_multi_create_rolls_back_late_expression_error() {
        let mut g = DirGraph::new();
        let params = HashMap::new();
        let opts = ExecuteOptions::eager(&params);
        let error = match execute_mut(
            &mut g,
            "CREATE (:Item {id: 1}), (:Item {id: 2, broken: duration({months: 2147483648})})",
            &opts,
        ) {
            Err(error) => error,
            Ok(_) => panic!("the second CREATE expression must fail"),
        };
        assert!(
            error.to_string().contains("duration()"),
            "unexpected error: {error}"
        );
        assert_eq!(g.graph.node_count(), 0, "the first CREATE must roll back");
        assert_eq!(g.version(), 0);
    }

    #[test]
    fn checkpoint_free_mutations_cancel_before_their_first_write() {
        static CANCEL: AtomicBool = AtomicBool::new(false);

        let mut g = DirGraph::new();
        let params = HashMap::new();
        let base_opts = ExecuteOptions::eager(&params);
        execute_mut(&mut g, "CREATE (:Item {id: 1})", &base_opts).unwrap();

        CANCEL.store(true, std::sync::atomic::Ordering::Relaxed);
        let mut cancelled_opts = ExecuteOptions::eager(&params);
        cancelled_opts.cancel = Some(&CANCEL);
        assert!(matches!(
            execute_mut(&mut g, "MATCH (n:Item) DELETE n", &cancelled_opts),
            Err(KgError::Cancelled)
        ));
        assert_eq!(g.graph.node_count(), 1);
        CANCEL.store(false, std::sync::atomic::Ordering::Relaxed);
    }
}