reddb-io-server 1.27.5

//! Runtime evaluator for the Fase 2 `ast::Expr` tree.
//!
//! This module evaluates a parsed `Expr` against a concrete
//! `UnifiedRecord`, returning `Option<Value>` where `None` means
//! "unresolvable" (null / missing field / unsupported op). The
//! evaluator intentionally mirrors the semantics of the legacy
//! `evaluate_runtime_filter` / `evaluate_scalar_function` paths so
//! call sites can swap in `evaluate_runtime_expr` without behavioural
//! drift.
//!
//! Scope today (Week 3):
//! - `OrderByClause.expr` — sort key evaluation for ORDER BY expr.
//!
//! Scope tomorrow (Week 3 continuation):
//! - `Filter::Compare` RHS once the variant grows an `Expr` slot.
//! - `Projection::Expression` once the planner flips from Filter
//!   to Expr for scalar projection bodies.

use super::join_filter::resolve_runtime_field;
use std::collections::{HashMap, HashSet};

use crate::storage::query::unified::UnifiedRecord;
use crate::storage::RedDB;
use reddb_rql::ast::{Expr, FieldRef};
use reddb_types::Value;

/// Evaluate typed scalar-subquery literals without changing the original AST.
/// Its encrypted values remain available to sensitivity checks and plan caches;
/// the plaintext copy exists only for this evaluator call.
pub(super) fn evaluate_typed_expr_with_secret_literals(
    expression: &Expr,
    row: &dyn crate::storage::query::evaluator::Row,
) -> Result<Value, crate::storage::query::evaluator::EvalError> {
    if !expression_contains_secret_literal(expression) {
        return crate::storage::query::evaluator::evaluate(expression, row);
    }
    let mut expression = expression.clone();
    let mut pending = vec![&mut expression];
    while let Some(expression) = pending.pop() {
        match expression {
            Expr::Literal {
                value: value @ Value::Secret(_),
                ..
            } => {
                *value = super::execution_context::secret_query_input(value.clone())
                    .unwrap_or(Value::Null);
            }
            Expr::UnaryOp { operand, .. } | Expr::IsNull { operand, .. } => pending.push(operand),
            Expr::Cast { inner, .. } => pending.push(inner),
            Expr::BinaryOp { lhs, rhs, .. } => pending.extend([lhs.as_mut(), rhs.as_mut()]),
            Expr::FunctionCall { args, .. } => pending.extend(args),
            Expr::WindowFunctionCall { args, window, .. } => {
                pending.extend(args);
                pending.extend(&mut window.partition_by);
                pending.extend(window.order_by.iter_mut().map(|clause| &mut clause.expr));
            }
            Expr::Case {
                branches, else_, ..
            } => {
                for (condition, value) in branches {
                    pending.extend([condition, value]);
                }
                if let Some(value) = else_ {
                    pending.push(value);
                }
            }
            Expr::InList { target, values, .. } => {
                pending.push(target);
                pending.extend(values);
            }
            Expr::Between {
                target, low, high, ..
            } => {
                pending.extend([target.as_mut(), low.as_mut(), high.as_mut()]);
            }
            _ => {}
        }
    }
    crate::storage::query::evaluator::evaluate(&expression, row)
}

fn expression_contains_secret_literal(expression: &Expr) -> bool {
    let mut pending = vec![expression];
    while let Some(expression) = pending.pop() {
        match expression {
            Expr::Literal {
                value: Value::Secret(_),
                ..
            } => return true,
            Expr::UnaryOp { operand, .. } | Expr::IsNull { operand, .. } => pending.push(operand),
            Expr::Cast { inner, .. } => pending.push(inner),
            Expr::BinaryOp { lhs, rhs, .. } => pending.extend([lhs.as_ref(), rhs.as_ref()]),
            Expr::FunctionCall { args, .. } => pending.extend(args),
            Expr::WindowFunctionCall { args, window, .. } => {
                pending.extend(args);
                pending.extend(&window.partition_by);
                pending.extend(window.order_by.iter().map(|clause| &clause.expr));
            }
            Expr::Case {
                branches, else_, ..
            } => {
                for (condition, value) in branches {
                    pending.extend([condition, value]);
                }
                if let Some(value) = else_ {
                    pending.push(value);
                }
            }
            Expr::InList { target, values, .. } => {
                pending.push(target);
                pending.extend(values);
            }
            Expr::Between {
                target, low, high, ..
            } => {
                pending.extend([target.as_ref(), low.as_ref(), high.as_ref()]);
            }
            _ => {}
        }
    }
    false
}

/// Evaluate an `Expr` against a record and return its resulting
/// `Value`, or `None` if the expression cannot be resolved (missing
/// column, type mismatch, unsupported feature for this phase).
pub(super) fn evaluate_runtime_expr(
    expr: &Expr,
    record: &UnifiedRecord,
    table_name: Option<&str>,
    table_alias: Option<&str>,
) -> Option<Value> {
    evaluate_runtime_expr_with_db(None, expr, record, table_name, table_alias)
}

pub(super) fn evaluate_runtime_expr_with_db(
    db: Option<&RedDB>,
    expr: &Expr,
    record: &UnifiedRecord,
    table_name: Option<&str>,
    table_alias: Option<&str>,
) -> Option<Value> {
    reddb_engine::query::runtime::expression::evaluate_runtime_expr_with_host(
        &NativeExpressionHost {
            db,
            record,
            table_name,
            table_alias,
        },
        expr,
    )
}

struct NativeExpressionHost<'a> {
    db: Option<&'a RedDB>,
    record: &'a UnifiedRecord,
    table_name: Option<&'a str>,
    table_alias: Option<&'a str>,
}
impl reddb_engine::query::runtime::expression::RuntimeExpressionHost for NativeExpressionHost<'_> {
    fn secret_query_input(&self, value: Value) -> Option<Value> {
        super::execution_context::secret_query_input(value)
    }
    fn resolve_field(&self, field: &FieldRef) -> Option<Value> {
        resolve_runtime_field(self.record, field, self.table_name, self.table_alias)
    }
    fn evaluate_special_function(
        &self,
        name: &str,
        args: &[Expr],
        evaluate: &dyn Fn(&Expr) -> Option<Value>,
    ) -> Option<Value> {
        match name {
            "CONFIG" => evaluate_runtime_config_function(self.db, args, evaluate),
            "KV" => evaluate_runtime_kv_function(self.db, args, evaluate),
            "__SECRET_REF" => {
                let key = expr_path_text(args.first()?)?.to_ascii_lowercase();
                crate::runtime::impl_core::current_secret_value(&key).or(Some(Value::Null))
            }
            "__KV_REF" => {
                let key = expr_path_text(args.first()?)?.to_ascii_lowercase();
                crate::runtime::impl_core::current_kv_value(&key)
                    .map(Value::text)
                    .or(Some(Value::Null))
            }
            _ => None,
        }
    }
    fn evaluate_function(&self, name: &str, args: &[Value]) -> Option<Value> {
        // ML and semantic-cache scalars live on the `RedDB` handle
        // (model registry, shared cache). Route them before the
        // db-less builtin dispatcher so the values land with full
        // context.
        if matches!(
            name,
            "ML_CLASSIFY"
                | "ML_PREDICT_PROBA"
                | "SEMANTIC_CACHE_GET"
                | "SEMANTIC_CACHE_PUT"
                | "EMBED"
        ) {
            if let Some(db) = self.db {
                return dispatch_ml_function(db, name, args);
            }
            return None;
        }
        if matches!(
            name,
            "CA_REGISTER" | "CA_DROP" | "CA_STATE" | "CA_LIST" | "CA_REFRESH" | "CA_QUERY"
        ) {
            if let Some(db) = self.db {
                return dispatch_ca_function(db, name, args);
            }
            return None;
        }
        if matches!(
            name,
            "LIST_HYPERTABLES" | "LIST_MODELS" | "SHOW_HYPERTABLES" | "SHOW_MODELS"
        ) {
            if let Some(db) = self.db {
                return dispatch_introspection_function(db, name);
            }
            return None;
        }
        if name == "HYPERTABLE_PRUNE_CHUNKS" {
            if let Some(db) = self.db {
                return dispatch_hypertable_prune(db, args);
            }
            return None;
        }
        if matches!(
            name,
            "HYPERTABLE_DROP_CHUNKS_BEFORE"
                | "HYPERTABLE_SWEEP_EXPIRED"
                | "HYPERTABLE_SHOW_CHUNKS"
                | "HYPERTABLE_SWEEP_ALL_EXPIRED"
                | "HYPERTABLE_SET_TTL"
                | "HYPERTABLE_GET_TTL"
                | "HYPERTABLE_CHUNKS_EXPIRING_WITHIN"
        ) {
            if let Some(db) = self.db {
                return dispatch_hypertable_retention(db, name, args);
            }
            return None;
        }
        if matches!(name, "MODEL_REGISTER" | "MODEL_DROP") {
            if let Some(db) = self.db {
                return dispatch_model_function(db, name, args);
            }
            return None;
        }
        if name == "RED.LCA" {
            if let Some(db) = self.db {
                return dispatch_vcs_lca_function(db, args);
            }
            return None;
        }
        // The shared walk has already uppercased the function name.
        dispatch_builtin_function(name, args)
    }
}

pub(super) fn lookup_latest_kv_value(db: &RedDB, collection: &str, key: &str) -> Option<Value> {
    if db
        .collection_contract(collection)
        .is_some_and(|contract| contract.declared_model == crate::catalog::CollectionModel::Vault)
    {
        return None;
    }
    let manager = db.store().get_collection(collection)?;
    let mut latest_id: u64 = 0;
    let mut latest_value: Option<Value> = None;
    // The parser rebuilds dotted paths by concatenating token display
    // strings; any segment that collides with a SQL keyword (DEFAULT,
    // LEFT, IN, …) comes back uppercase, but SET CONFIG stores keys
    // lowercase. Normalise here so CONFIG(red.ai.default.provider)
    // matches the key persisted by SET CONFIG.
    let key_lc = key.to_ascii_lowercase();
    manager.for_each_entity(|entity| {
        let Some(row) = entity.data.as_row() else {
            return true;
        };
        let entry_key = row.get_field("key").and_then(|value| match value {
            Value::Text(text) => Some(text.as_ref()),
            _ => None,
        });
        if entry_key == Some(key_lc.as_str()) && entity.id.raw() >= latest_id {
            latest_id = entity.id.raw();
            latest_value = Some(row.get_field("value").cloned().unwrap_or(Value::Null));
        }
        true
    });
    latest_value
}

fn evaluate_runtime_config_function(
    db: Option<&RedDB>,
    args: &[Expr],
    evaluate: &dyn Fn(&Expr) -> Option<Value>,
) -> Option<Value> {
    let key = expr_path_text(args.first()?)?;
    if let Some(value) = crate::runtime::impl_core::current_config_value(&key) {
        return Some(value);
    }
    if let Some(db) = db {
        // #1743 — gate the raw `red_config` fallback on `config:read`; without
        // it a principal denied by `current_config_value` would leak the value
        // through this bypass path.
        if crate::runtime::impl_core::config_read_permitted(&key) {
            // `$config.<path>` desugars to CONFIG("red.config/<path>") but SET CONFIG
            // stores under the bare key — so try the stripped key too (#1370).
            let key_str: &str = key.as_ref();
            let bare = key_str.strip_prefix("red.config/").unwrap_or(key_str);
            if let Some(value) = lookup_latest_kv_value(db, "red_config", &key)
                .or_else(|| lookup_latest_kv_value(db, "red_config", bare))
            {
                return Some(value);
            }
        }
    }
    args.get(1)
        .and_then(|expr| special_default_expr_value(expr, evaluate))
        .or(Some(Value::Null))
}

fn evaluate_runtime_kv_function(
    db: Option<&RedDB>,
    args: &[Expr],
    evaluate: &dyn Fn(&Expr) -> Option<Value>,
) -> Option<Value> {
    let collection = expr_path_text(args.first()?)?;
    let key = expr_path_text(args.get(1)?)?;
    if let Some(db) = db {
        // #1743 — `KV('red_config', …)` is the same leak surface as `$config`;
        // gate config-collection reads on `config:read`. Other collections are
        // unaffected.
        if crate::runtime::impl_core::kv_read_permitted(&collection, &key) {
            if let Some(value) = lookup_latest_kv_value(db, &collection, &key) {
                return Some(value);
            }
        }
    }
    args.get(2)
        .and_then(|expr| special_default_expr_value(expr, evaluate))
        .or(Some(Value::Null))
}

fn special_default_expr_value(
    expr: &Expr,
    evaluate: &dyn Fn(&Expr) -> Option<Value>,
) -> Option<Value> {
    match expr {
        Expr::Column { field, .. } => field_ref_path_text(field).map(Value::text),
        _ => evaluate(expr),
    }
}

fn expr_path_text(expr: &Expr) -> Option<String> {
    match expr {
        Expr::Column { field, .. } => field_ref_path_text(field),
        Expr::Literal { value, .. } => literal_path_text(value),
        _ => None,
    }
}

fn field_ref_path_text(field: &FieldRef) -> Option<String> {
    match field {
        FieldRef::TableColumn { table, column } => Some(if table.is_empty() {
            column.clone()
        } else {
            format!("{table}.{column}")
        }),
        FieldRef::NodeProperty { alias, property } => Some(format!("{alias}.{property}")),
        FieldRef::EdgeProperty { alias, property } => Some(format!("{alias}.{property}")),
        FieldRef::NodeId { alias } => Some(format!("{alias}.id")),
    }
}

fn literal_path_text(value: &Value) -> Option<String> {
    match value {
        Value::Null => None,
        Value::Text(text) => Some(text.to_string()),
        _ => Some(value.display_string()),
    }
}

/// Minimal built-in function dispatcher used by `Expr::FunctionCall`.
/// For Week 3 we cover only the pure-scalar functions that can be
/// evaluated from a `&[Value]` argument list — the geo / time-bucket
/// functions that require row-level access stay on the legacy
/// Projection::Function path for now. Week 4 folds them into a
/// proper registry keyed on Expr.
fn feature_vector_from_value(value: &Value) -> Option<Vec<f32>> {
    match value {
        Value::Vector(v) => Some(v.clone()),
        Value::Array(items) => {
            let mut out = Vec::with_capacity(items.len());
            for item in items {
                let n = match item {
                    Value::Integer(n) | Value::BigInt(n) => *n as f32,
                    Value::UnsignedInteger(n) => *n as f32,
                    Value::Float(f) => *f as f32,
                    _ => return None,
                };
                out.push(n);
            }
            Some(out)
        }
        _ => None,
    }
}

fn model_kind_from_json(hyperparams_json: &str) -> String {
    crate::serde_json::from_str::<crate::serde_json::Value>(hyperparams_json)
        .ok()
        .as_ref()
        .and_then(|v| v.get("kind"))
        .and_then(|k| k.as_str())
        .unwrap_or("logreg")
        .to_ascii_lowercase()
}

fn dispatch_ml_function(db: &RedDB, name: &str, args: &[Value]) -> Option<Value> {
    match name {
        "ML_CLASSIFY" | "ML_PREDICT_PROBA" => {
            let model_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let features = feature_vector_from_value(args.get(1)?)?;

            let version = db.ml_runtime().registry().get_active(&model_name).ok()??;
            let kind = model_kind_from_json(&version.hyperparams_json);
            let weights_json = std::str::from_utf8(&version.weights_blob).ok()?;

            use crate::storage::ml::classifier::IncrementalClassifier;
            let (class, probs) = match kind.as_str() {
                "nb" | "naive_bayes" => {
                    let m = crate::storage::ml::classifier::MultinomialNaiveBayes::from_json(
                        weights_json,
                    )?;
                    (m.predict(&features), m.predict_proba(&features))
                }
                _ => {
                    let m = crate::storage::ml::classifier::LogisticRegression::from_json(
                        weights_json,
                    )?;
                    (m.predict(&features), m.predict_proba(&features))
                }
            };

            if name == "ML_PREDICT_PROBA" {
                Some(Value::Array(
                    probs.into_iter().map(|p| Value::Float(p as f64)).collect(),
                ))
            } else {
                Some(
                    class
                        .map(|c| Value::Integer(c as i64))
                        .unwrap_or(Value::Null),
                )
            }
        }
        "SEMANTIC_CACHE_GET" => {
            let _ns = args.first()?;
            let embedding = feature_vector_from_value(args.get(1)?)?;
            Some(match db.semantic_cache().lookup(&embedding) {
                Some(resp) => Value::text(resp),
                None => Value::Null,
            })
        }
        "SEMANTIC_CACHE_PUT" => {
            let _ns = args.first()?;
            let prompt = match args.get(1)? {
                Value::Text(s) => s.to_string(),
                other => format!("{:?}", other),
            };
            let response = match args.get(2)? {
                Value::Text(s) => s.to_string(),
                other => format!("{:?}", other),
            };
            let embedding = feature_vector_from_value(args.get(3)?)?;
            db.semantic_cache()
                .insert(prompt, response, embedding, None);
            Some(Value::Boolean(true))
        }
        "EMBED" => {
            let text = match args.first()? {
                Value::Text(s) => s.to_string(),
                other => other.display_string(),
            };
            let provider_hint = args.get(1).and_then(|v| match v {
                Value::Text(s) => Some(s.to_string()),
                _ => None,
            });
            embed_text(db, &text, provider_hint.as_deref())
        }
        _ => None,
    }
}

pub(super) fn embed_text_public(
    db: &RedDB,
    text: &str,
    provider_hint: Option<&str>,
) -> Option<Value> {
    embed_text(db, text, provider_hint)
}

fn embed_text(db: &RedDB, text: &str, provider_hint: Option<&str>) -> Option<Value> {
    // Resolve provider from explicit hint, then fall back to the
    // runtime-wide default captured in `red_config`. Ollama and other
    // OpenAI-compatible endpoints share the same embeddings wire shape;
    // only the api_base and model differ.
    let kv_getter = |k: &str| -> Result<Option<String>, crate::RedDBError> {
        Ok(
            lookup_latest_kv_value(db, "red_config", k).and_then(|v| match v {
                Value::Text(s) => Some(s.to_string()),
                _ => None,
            }),
        )
    };
    // Embeddings follow the embeddings task pointer (ADR-0068 §5), not the
    // inference/ASK provider, when no explicit hint is supplied.
    let provider = match provider_hint {
        Some(name) => crate::ai::parse_provider(name).ok()?,
        None => crate::ai::resolve_embeddings_provider(&kv_getter).ok()?,
    };
    if !provider.is_openai_compatible() {
        // Anthropic / providers without embedding parity fall out —
        // caller will see Null. A later sprint widens this.
        return None;
    }

    let crate::ai::AiConnection { api_key, api_base } =
        crate::ai::resolve_connection(&provider, None, &|kv_key| {
            Ok(
                lookup_latest_kv_value(db, "red_config", kv_key).and_then(|v| match v {
                    Value::Text(s) => Some(s.to_string()),
                    _ => None,
                }),
            )
        })
        .ok()?;

    let model = crate::ai::resolve_embeddings_model(&provider, &kv_getter);

    let transport = crate::runtime::ai::transport::AiTransport::new(
        crate::runtime::ai::transport::AiTransportConfig::default(),
    );
    let request = crate::ai::OpenAiEmbeddingRequest {
        api_key,
        model,
        inputs: vec![text.to_string()],
        dimensions: None,
        api_base,
    };
    match crate::runtime::ai::block_on_ai(async move {
        crate::ai::openai_embeddings_async(&transport, request).await
    })
    .and_then(|result| result)
    {
        Ok(resp) => resp.embeddings.into_iter().next().map(Value::Vector),
        Err(_) => None,
    }
}

pub(super) fn dispatch_model_function_public(
    db: &RedDB,
    name: &str,
    args: &[Value],
) -> Option<Value> {
    dispatch_model_function(db, name, args)
}

/// `MODEL_REGISTER(name, kind, weights_json [, hyperparams_json, metrics_json])` —
/// appends a new version to the model registry and activates it.
/// `kind` is `'logreg'` or `'nb'` and is stored inside
/// `hyperparams_json` for ML_CLASSIFY to decode.
///
/// `MODEL_DROP(name)` — archives every version and clears activation.
fn dispatch_model_function(db: &RedDB, name: &str, args: &[Value]) -> Option<Value> {
    use crate::storage::ml::ModelVersion;

    match name {
        "MODEL_REGISTER" => {
            let model_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let kind = match args.get(1)? {
                Value::Text(s) => s.to_ascii_lowercase(),
                _ => return Some(Value::Null),
            };
            let weights_json = match args.get(2)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let hyperparams_json = match args.get(3) {
                Some(Value::Text(s)) => {
                    // Caller-supplied hyperparams; stamp `kind` in if
                    // not already there so downstream decoding picks
                    // the right classifier.
                    if s.contains("\"kind\"") {
                        s.to_string()
                    } else if s.trim() == "{}" || s.trim().is_empty() {
                        format!("{{\"kind\":\"{kind}\"}}")
                    } else {
                        let trimmed = s.trim_start_matches('{').to_string();
                        format!("{{\"kind\":\"{kind}\",{trimmed}")
                    }
                }
                _ => format!("{{\"kind\":\"{kind}\"}}"),
            };
            let metrics_json = match args.get(4) {
                Some(Value::Text(s)) => s.to_string(),
                _ => "{}".to_string(),
            };

            let version = ModelVersion {
                model: model_name.clone(),
                version: 0,
                weights_blob: weights_json.into_bytes(),
                hyperparams_json,
                metrics_json,
                training_data_hash: None,
                training_sql: None,
                parent_version: None,
                created_at_ms: 0,
                created_by: None,
                archived: false,
            };
            let v = db
                .ml_runtime()
                .registry()
                .register_version(model_name, version, /*make_active=*/ true)
                .ok()?;
            Some(Value::Integer(v as i64))
        }
        "MODEL_DROP" => {
            let model_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let reg = db.ml_runtime().registry();
            let versions = reg.list_versions(&model_name).ok()?;
            for v in versions {
                let _ = reg.archive_version(&model_name, v.version);
            }
            Some(Value::Boolean(true))
        }
        _ => None,
    }
}

pub(super) fn dispatch_introspection_function_public(db: &RedDB, name: &str) -> Option<Value> {
    dispatch_introspection_function(db, name)
}

/// `HYPERTABLE_PRUNE_CHUNKS(name, lo_ns, hi_ns)` — returns the chunk
/// start timestamps that overlap `[lo_ns, hi_ns)` for the given
/// hypertable. Exposes the partition-pruning primitive over real
/// allocated chunks. Uses RANGE semantics (the only kind hypertables
/// have today).
pub(super) fn dispatch_hypertable_prune_public(db: &RedDB, args: &[Value]) -> Option<Value> {
    dispatch_hypertable_prune(db, args)
}

pub(super) fn dispatch_hypertable_retention_public(
    db: &RedDB,
    name: &str,
    args: &[Value],
) -> Option<Value> {
    dispatch_hypertable_retention(db, name, args)
}

/// Retention + introspection scalars on top of HypertableRegistry.
///
/// * `HYPERTABLE_SHOW_CHUNKS(name)` — array of `"name:start_ns"` for
///   every chunk of the hypertable.
/// * `HYPERTABLE_DROP_CHUNKS_BEFORE(name, cutoff_ns)` — drops every
///   chunk whose `max_ts_ns <= cutoff`. Returns the drop count.
/// * `HYPERTABLE_SWEEP_EXPIRED(name [, now_ns])` — drops every chunk
///   whose effective TTL has fired. Returns the count.
fn dispatch_hypertable_retention(db: &RedDB, name: &str, args: &[Value]) -> Option<Value> {
    let registry = db.hypertables();

    // HYPERTABLE_SWEEP_ALL_EXPIRED takes an optional now_ns and
    // doesn't need a hypertable name — handle before the per-name
    // dispatch path.
    if name == "HYPERTABLE_SWEEP_ALL_EXPIRED" {
        if db.is_replica_role() {
            return Some(Value::Integer(0));
        }
        let now_ns = args
            .first()
            .and_then(|v| match v {
                Value::Integer(n) | Value::BigInt(n) => Some(*n as u64),
                Value::UnsignedInteger(n) => Some(*n),
                _ => None,
            })
            .unwrap_or_else(|| {
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_nanos() as u64)
                    .unwrap_or(0)
            });
        let dropped = registry.sweep_all_expired(now_ns).len();
        return Some(Value::Integer(dropped as i64));
    }

    let ht_name = match args.first()? {
        Value::Text(s) => s.to_string(),
        _ => return Some(Value::Null),
    };
    match name {
        "HYPERTABLE_SHOW_CHUNKS" => Some(Value::Array(
            registry
                .show_chunks(&ht_name)
                .into_iter()
                .map(|c| Value::text(format!("{}:{}", c.id.hypertable, c.id.start_ns)))
                .collect(),
        )),
        "HYPERTABLE_DROP_CHUNKS_BEFORE" => {
            if db.is_replica_role() {
                return Some(Value::Integer(0));
            }
            let cutoff = match args.get(1)? {
                Value::Integer(n) | Value::BigInt(n) => *n as u64,
                Value::UnsignedInteger(n) => *n,
                _ => return Some(Value::Null),
            };
            let dropped = registry.drop_chunks_before(&ht_name, cutoff).len();
            Some(Value::Integer(dropped as i64))
        }
        "HYPERTABLE_SWEEP_EXPIRED" => {
            if db.is_replica_role() {
                return Some(Value::Integer(0));
            }
            let now_ns = args
                .get(1)
                .and_then(|v| match v {
                    Value::Integer(n) | Value::BigInt(n) => Some(*n as u64),
                    Value::UnsignedInteger(n) => Some(*n),
                    _ => None,
                })
                .unwrap_or_else(|| {
                    std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map(|d| d.as_nanos() as u64)
                        .unwrap_or(0)
                });
            let dropped = registry.sweep_expired(&ht_name, now_ns).len();
            Some(Value::Integer(dropped as i64))
        }
        "HYPERTABLE_SET_TTL" => {
            if db.is_replica_role() {
                return Some(Value::Boolean(false));
            }
            // HYPERTABLE_SET_TTL(name, duration | null)
            let ttl_ns = match args.get(1)? {
                Value::Null => None,
                Value::Text(s) => {
                    Some(crate::storage::timeseries::retention::parse_duration_ns(s)?)
                }
                Value::Integer(n) | Value::BigInt(n) if *n >= 0 => Some(*n as u64),
                Value::UnsignedInteger(n) => Some(*n),
                _ => return Some(Value::Null),
            };
            Some(Value::Boolean(
                registry.set_default_ttl_ns(&ht_name, ttl_ns),
            ))
        }
        "HYPERTABLE_GET_TTL" => {
            let spec = registry.get(&ht_name)?;
            Some(match spec.default_ttl_ns {
                Some(n) => Value::Integer(n as i64),
                None => Value::Null,
            })
        }
        "HYPERTABLE_CHUNKS_EXPIRING_WITHIN" => {
            // HYPERTABLE_CHUNKS_EXPIRING_WITHIN(name, now_ns, horizon_ns)
            let now_ns = match args.get(1)? {
                Value::Integer(n) | Value::BigInt(n) => *n as u64,
                Value::UnsignedInteger(n) => *n,
                _ => return Some(Value::Null),
            };
            let horizon_ns = match args.get(2)? {
                Value::Integer(n) | Value::BigInt(n) => *n as u64,
                Value::UnsignedInteger(n) => *n,
                _ => return Some(Value::Null),
            };
            let expiring = registry.chunks_expiring_within(&ht_name, now_ns, horizon_ns);
            Some(Value::Array(
                expiring
                    .into_iter()
                    .map(|c| Value::text(format!("{}:{}", c.id.hypertable, c.id.start_ns)))
                    .collect(),
            ))
        }
        _ => None,
    }
}

fn dispatch_hypertable_prune(db: &RedDB, args: &[Value]) -> Option<Value> {
    use crate::storage::query::planner::partition_pruning::{
        prune_range, PruneKind, PruneOp, PrunePartitioning, PrunePredicate, PruneValue, RangeChild,
    };

    let ht_name = match args.first()? {
        Value::Text(s) => s.to_string(),
        _ => return Some(Value::Null),
    };
    let lo = match args.get(1)? {
        Value::Integer(n) | Value::BigInt(n) => *n as u64,
        Value::UnsignedInteger(n) => *n,
        _ => return Some(Value::Null),
    };
    let hi = match args.get(2)? {
        Value::Integer(n) | Value::BigInt(n) => *n as u64,
        Value::UnsignedInteger(n) => *n,
        _ => return Some(Value::Null),
    };

    let registry = db.hypertables();
    let spec = registry.get(&ht_name)?;
    let chunks = registry.show_chunks(&ht_name);

    // Build RangeChild list — one per chunk — then feed the predicate
    // `spec.time_column >= lo AND spec.time_column < hi` to the
    // existing pruner primitive.
    let children: Vec<RangeChild> = chunks
        .iter()
        .map(|c| RangeChild {
            name: format!("{}:{}", c.id.hypertable, c.id.start_ns),
            low: Some(PruneValue::Int(c.id.start_ns as i64)),
            high_exclusive: Some(PruneValue::Int(c.end_ns_exclusive as i64)),
        })
        .collect();

    let partitioning = PrunePartitioning {
        kind: PruneKind::Range,
        column: spec.time_column.clone(),
    };
    let pred = PrunePredicate::And(vec![
        PrunePredicate::Compare {
            column: spec.time_column.clone(),
            op: PruneOp::GtEq,
            value: PruneValue::Int(lo as i64),
        },
        PrunePredicate::Compare {
            column: spec.time_column.clone(),
            op: PruneOp::Lt,
            value: PruneValue::Int(hi as i64),
        },
    ]);
    let kept = prune_range(&partitioning, &children, &pred);
    Some(Value::Array(kept.into_iter().map(Value::text).collect()))
}

fn dispatch_introspection_function(db: &RedDB, name: &str) -> Option<Value> {
    match name {
        "LIST_HYPERTABLES" | "SHOW_HYPERTABLES" => {
            let names: Vec<Value> = db
                .hypertables()
                .list()
                .into_iter()
                .map(|s| Value::text(s.name))
                .collect();
            Some(Value::Array(names))
        }
        "LIST_MODELS" | "SHOW_MODELS" => {
            let summaries = db.ml_runtime().registry().summaries().ok()?;
            Some(Value::Array(
                summaries.into_iter().map(|s| Value::text(s.name)).collect(),
            ))
        }
        _ => None,
    }
}

pub(super) fn dispatch_ca_function_public(db: &RedDB, name: &str, args: &[Value]) -> Option<Value> {
    dispatch_ca_function(db, name, args)
}

fn dispatch_ca_function(db: &RedDB, name: &str, args: &[Value]) -> Option<Value> {
    use crate::storage::timeseries::continuous_aggregate::{
        ContinuousAggregateColumn, ContinuousAggregateSpec,
    };
    use crate::storage::timeseries::AggregationType;

    let engine = db.continuous_aggregates();

    match name {
        "CA_REGISTER" => {
            // CA_REGISTER(name, source, bucket_duration, alias, agg, field
            //             [, refresh_lag, max_interval])
            let ca_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let source = match args.get(1)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let bucket = match args.get(2)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let alias = match args.get(3)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let agg = match args.get(4)? {
                Value::Text(s) => AggregationType::from_str(s)?,
                _ => return Some(Value::Null),
            };
            let field = match args.get(5)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let lag = args
                .get(6)
                .and_then(|v| match v {
                    Value::Text(s) => Some(s.to_string()),
                    _ => None,
                })
                .unwrap_or_else(|| "0s".to_string());
            let max_interval = args
                .get(7)
                .and_then(|v| match v {
                    Value::Text(s) => Some(s.to_string()),
                    _ => None,
                })
                .unwrap_or_else(|| "365d".to_string());

            let spec = ContinuousAggregateSpec::from_durations(
                ca_name,
                source,
                &bucket,
                vec![ContinuousAggregateColumn {
                    alias,
                    source_column: field,
                    agg,
                }],
                &lag,
                &max_interval,
            )?;
            engine.register(spec);
            Some(Value::Boolean(true))
        }
        "CA_DROP" => {
            let ca_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            engine.drop_aggregate(&ca_name);
            Some(Value::Boolean(true))
        }
        "CA_STATE" => {
            // Returns a JSON-ish summary — buckets count + last refreshed.
            let ca_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            match engine.state(&ca_name) {
                Some(state) => Some(Value::text(format!(
                    "{{\"last_refreshed_bucket_ns\":{},\"bucket_count\":{}}}",
                    state.last_refreshed_bucket_ns(),
                    state.bucket_count()
                ))),
                None => Some(Value::Null),
            }
        }
        "CA_LIST" => {
            let names: Vec<Value> = engine
                .list()
                .into_iter()
                .map(|s| Value::text(s.name))
                .collect();
            Some(Value::Array(names))
        }
        "CA_REFRESH" => {
            // CA_REFRESH(name [, now_ns]) — scans the source collection
            // for row entities, extracts the timestamp column from the
            // spec and every aggregated source_column, folds into
            // buckets. `now_ns` defaults to wall-clock now.
            let ca_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let now_ns = args
                .get(1)
                .and_then(|v| match v {
                    Value::Integer(n) | Value::BigInt(n) => Some(*n as u64),
                    Value::UnsignedInteger(n) => Some(*n),
                    _ => None,
                })
                .unwrap_or_else(|| {
                    std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map(|d| d.as_nanos() as u64)
                        .unwrap_or(0)
                });

            let specs = engine.list();
            let spec = specs.into_iter().find(|s| s.name == ca_name)?;
            let store = db.store();
            let time_col = "ts".to_string();
            let columns = spec.columns.clone();
            let source_name = spec.source.clone();
            let source_cb: crate::storage::timeseries::continuous_aggregate::ContinuousAggregateSource =
                std::sync::Arc::new(move |src: &str, start: u64, end: u64| {
                    let mut out = Vec::new();
                    let Some(mgr) = store.get_collection(src) else {
                        return out;
                    };
                    for entity in mgr.query_all(|_| true) {
                        let crate::storage::unified::entity::EntityData::Row(row) = &entity.data
                        else {
                            continue;
                        };
                        let ts = match row.get_field(&time_col) {
                            Some(Value::Integer(n) | Value::BigInt(n)) => *n as u64,
                            Some(Value::UnsignedInteger(n)) => *n,
                            _ => continue,
                        };
                        if ts < start || ts >= end {
                            continue;
                        }
                        let mut values = std::collections::HashMap::new();
                        for col in &columns {
                            let v = row.get_field(&col.source_column).and_then(|v| match v {
                                Value::Integer(n) | Value::BigInt(n) => Some(*n as f64),
                                Value::UnsignedInteger(n) => Some(*n as f64),
                                Value::Float(f) => Some(*f),
                                _ => None,
                            });
                            if let Some(f) = v {
                                values.insert(col.alias.clone(), f);
                            }
                        }
                        out.push(
                            crate::storage::timeseries::continuous_aggregate::RefreshPoint {
                                ts_ns: ts,
                                values,
                            },
                        );
                    }
                    out
                });
            let _ = source_name;
            let absorbed = engine.refresh(&ca_name, now_ns, &source_cb);
            Some(Value::Integer(absorbed as i64))
        }
        "CA_QUERY" => {
            // CA_QUERY(name, bucket_start_ns, alias) — returns the
            // aggregated value for a single bucket using the aggregator
            // declared at registration time.
            let ca_name = match args.first()? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let bucket_start = match args.get(1)? {
                Value::Integer(n) | Value::BigInt(n) => *n as u64,
                Value::UnsignedInteger(n) => *n,
                _ => return Some(Value::Null),
            };
            let alias = match args.get(2)? {
                Value::Text(s) => s.to_string(),
                _ => return Some(Value::Null),
            };
            let state = engine.state(&ca_name)?;
            let spec = engine.list().into_iter().find(|s| s.name == ca_name)?;
            let agg = spec
                .columns
                .iter()
                .find(|c| c.alias == alias)
                .map(|c| c.agg)?;
            state
                .query(bucket_start, &alias, agg)
                .map(Value::Float)
                .or(Some(Value::Null))
        }
        _ => None,
    }
}

/// Crate-local shim that exposes `dispatch_builtin_function` to the
/// scalar evaluator without making the dispatcher itself part of the
/// runtime API. Once the evaluator owns the dispatch table this shim
/// disappears.
pub(super) fn scalar_dispatch_builtin(name: &str, args: &[Value]) -> Option<Value> {
    dispatch_builtin_function(name, args)
}

fn dispatch_vcs_lca_function(db: &RedDB, args: &[Value]) -> Option<Value> {
    if args.len() != 2 {
        return Some(Value::Null);
    }
    let a = text_arg(args.first()?)?;
    let b = text_arg(args.get(1)?)?;
    vcs_lca_from_store(db, a, b)
        .map(Value::text)
        .or(Some(Value::Null))
}

pub(super) fn dispatch_vcs_lca_function_public(db: &RedDB, args: &[Value]) -> Option<Value> {
    dispatch_vcs_lca_function(db, args)
}

fn text_arg(value: &Value) -> Option<&str> {
    match value {
        Value::Text(text) => Some(text.as_ref()),
        _ => None,
    }
}

fn vcs_lca_from_store(db: &RedDB, a: &str, b: &str) -> Option<String> {
    let commits = vcs_commit_index(db);
    let mut a_ancestors = HashSet::new();
    collect_vcs_ancestors(a, &commits, &mut a_ancestors);

    let mut visited = HashSet::new();
    let mut stack = vec![b.to_string()];
    let mut best: Option<(u64, String)> = None;
    while let Some(hash) = stack.pop() {
        if !visited.insert(hash.clone()) {
            continue;
        }
        if a_ancestors.contains(&hash) {
            let height = commits.get(&hash).map(|commit| commit.height).unwrap_or(0);
            match &best {
                Some((best_height, _)) if *best_height >= height => {}
                _ => best = Some((height, hash.clone())),
            }
            continue;
        }
        if let Some(commit) = commits.get(&hash) {
            for parent in &commit.parents {
                if !visited.contains(parent) {
                    stack.push(parent.clone());
                }
            }
        }
    }
    best.map(|(_, hash)| hash)
}

#[derive(Clone)]
struct VcsCommitNode {
    parents: Vec<String>,
    height: u64,
}

fn vcs_commit_index(db: &RedDB) -> HashMap<String, VcsCommitNode> {
    let Some(manager) = db
        .store()
        .get_collection(crate::application::vcs_collections::COMMITS)
    else {
        return HashMap::new();
    };
    manager
        .query_all(|_| true)
        .into_iter()
        .filter_map(|entity| {
            let row = entity.data.as_row()?;
            let hash = row_text_field(row, "id")?;
            Some((
                hash,
                VcsCommitNode {
                    parents: row_string_list_field(row, "parents"),
                    height: row_u64_field(row, "height").unwrap_or(0),
                },
            ))
        })
        .collect()
}

fn collect_vcs_ancestors(
    hash: &str,
    commits: &HashMap<String, VcsCommitNode>,
    out: &mut HashSet<String>,
) {
    if !out.insert(hash.to_string()) {
        return;
    }
    if let Some(commit) = commits.get(hash) {
        for parent in &commit.parents {
            collect_vcs_ancestors(parent, commits, out);
        }
    }
}

fn row_text_field(row: &crate::storage::unified::entity::RowData, field: &str) -> Option<String> {
    match row.get_field(field)? {
        Value::Text(value) => Some(value.to_string()),
        _ => None,
    }
}

fn row_u64_field(row: &crate::storage::unified::entity::RowData, field: &str) -> Option<u64> {
    match row.get_field(field)? {
        Value::UnsignedInteger(value) => Some(*value),
        Value::Integer(value) if *value >= 0 => Some(*value as u64),
        _ => None,
    }
}

fn row_string_list_field(
    row: &crate::storage::unified::entity::RowData,
    field: &str,
) -> Vec<String> {
    match row.get_field(field) {
        Some(Value::Array(items)) => items
            .iter()
            .filter_map(|value| match value {
                Value::Text(text) => Some(text.to_string()),
                _ => None,
            })
            .collect(),
        _ => Vec::new(),
    }
}

fn dispatch_builtin_function(name: &str, args: &[Value]) -> Option<Value> {
    reddb_engine::query::runtime::value_functions::evaluate_value_function_with_host(
        &NativeValueFunctionHost,
        name,
        args,
    )
}

struct NativeValueFunctionHost;
impl reddb_engine::query::runtime::value_functions::ValueFunctionHost for NativeValueFunctionHost {
    fn current_unix_ms(&self) -> i64 {
        reddb_engine::query::runtime::temporal::current_unix_ms()
    }
    fn fallback_function(&self, name: &str, args: &[Value]) -> Option<Value> {
        match name {
            "PG_ADVISORY_LOCK" => {
                let key = value_as_i64(args.first()?)?;
                crate::auth::locks::global()
                    .acquire(key, crate::runtime::impl_core::current_connection_id());
                Some(Value::Null)
            }
            "PG_TRY_ADVISORY_LOCK" => {
                let key = value_as_i64(args.first()?)?;
                Some(Value::Boolean(crate::auth::locks::global().try_acquire(
                    key,
                    crate::runtime::impl_core::current_connection_id(),
                )))
            }
            "PG_ADVISORY_UNLOCK" => {
                let key = value_as_i64(args.first()?)?;
                Some(Value::Boolean(crate::auth::locks::global().release(
                    key,
                    crate::runtime::impl_core::current_connection_id(),
                )))
            }
            "PG_ADVISORY_UNLOCK_ALL" => {
                let dropped = crate::auth::locks::global()
                    .release_all(crate::runtime::impl_core::current_connection_id());
                Some(Value::Integer(dropped as i64))
            }
            // Phase 2.5.3 multi-tenancy: reads the thread-local session
            // tenant installed by `SET TENANT 'id'` or transport
            // middleware. Returns NULL when no tenant is bound so RLS
            // policies like `USING (tenant_id = CURRENT_TENANT())` deny
            // every row for unauthenticated / unscoped sessions.
            "CURRENT_TENANT" => Some(
                crate::runtime::impl_core::current_tenant()
                    .map(Value::text)
                    .unwrap_or(Value::Null),
            ),
            // Session identity scalars — `WITHIN ... USER '<u>' AS ROLE '<r>'`
            // overrides win over the transport-installed thread-local.
            // Anonymous callers with no override get NULL.
            "CURRENT_USER" | "SESSION_USER" | "USER" => Some(
                crate::runtime::impl_core::current_user_projected()
                    .map(Value::text)
                    .unwrap_or(Value::Null),
            ),
            "CURRENT_ROLE" => Some(
                crate::runtime::impl_core::current_role_projected()
                    .map(Value::text)
                    .unwrap_or(Value::Null),
            ),
            // ── H3 hexagonal index scalars (PRD #1574 slice 1, #1575) ──
            // Pure wrappers over `crate::geo::h3`; the engine stays
            // h3o-agnostic. Cell ids are u64 → Value::UnsignedInteger.
            "H3_INDEX" => {
                let lat = value_as_f64(args.first()?)?;
                let lon = value_as_f64(args.get(1)?)?;
                let res = value_as_i64(args.get(2)?)?.clamp(0, 15) as u8;
                Some(Value::UnsignedInteger(crate::geo::h3::lat_lng_to_cell(
                    lat, lon, res,
                )))
            }
            "H3_CELL" => {
                let Some((lat, lon)) = geo_point(args.first()?) else {
                    return Some(Value::Null);
                };
                let res_value = value_as_i64(args.get(1)?)?;
                let Some(res) = crate::geo::h3::valid_resolution(res_value) else {
                    return Some(Value::Null);
                };
                let cell = crate::geo::h3::lat_lng_to_cell(lat, lon, res);
                if cell == 0 {
                    Some(Value::Null)
                } else {
                    Some(Value::UnsignedInteger(cell))
                }
            }
            "H3_PARENT" => {
                let cell = value_as_u64(args.first()?)?;
                let res_value = value_as_i64(args.get(1)?)?;
                let Some(res) = crate::geo::h3::valid_resolution(res_value) else {
                    return Some(Value::Null);
                };
                let parent = crate::geo::h3::cell_to_parent(cell, res);
                if parent == 0 {
                    Some(Value::Null)
                } else {
                    Some(Value::UnsignedInteger(parent))
                }
            }
            "H3_TO_LATLNG" => {
                let cell = value_as_u64(args.first()?)?;
                let (lat, lon) = crate::geo::h3::cell_to_lat_lng(cell);
                Some(Value::Array(vec![Value::Float(lat), Value::Float(lon)]))
            }
            "H3_RING" => {
                let cell = value_as_u64(args.first()?)?;
                let k = value_as_i64(args.get(1)?)?.max(0) as u32;
                Some(Value::Array(
                    crate::geo::h3::grid_disk(cell, k)
                        .into_iter()
                        .map(Value::UnsignedInteger)
                        .collect(),
                ))
            }
            "VERIFY_PASSWORD" => {
                let stored = args.first()?;
                let candidate = args.get(1)?;
                let hash: &str = match stored {
                    Value::Password(hash) => hash.as_str(),
                    Value::Text(hash) => hash.as_ref(),
                    _ => return Some(Value::Boolean(false)),
                };
                let plain: &str = match candidate {
                    Value::Text(plain) => plain.as_ref(),
                    _ => return Some(Value::Boolean(false)),
                };
                Some(Value::Boolean(crate::auth::store::verify_password(
                    plain, hash,
                )))
            }
            _ => None,
        }
    }
}

use reddb_engine::query::runtime::{
    value_functions::{geo_point, value_as_f64},
    values::{value_as_i64, value_as_u64},
};

#[cfg(test)]
mod tests {
    use super::dispatch_builtin_function;
    use reddb_types::Value;

    #[test]
    fn shared_value_functions_bound_json_growth_and_signed_abs() {
        assert_eq!(
            dispatch_builtin_function("ABS", &[Value::Integer(i64::MIN)]),
            None
        );
        for path in ["$[4294967296]", "$[18446744073709551615]"] {
            let json = Value::Json(b"[]".to_vec());
            assert_eq!(
                dispatch_builtin_function("JSON_EXTRACT", &[json.clone(), Value::text(path)]),
                Some(Value::Null)
            );
            assert_eq!(
                dispatch_builtin_function(
                    "JSON_SET",
                    &[json, Value::text(path), Value::Integer(1)]
                ),
                Some(Value::Null)
            );
        }
    }

    #[test]
    fn shared_runtime_expression_host_preserves_defaults_and_depth() {
        use reddb_rql::ast::{BinOp, Expr, FieldRef, Span};
        let mut record = crate::storage::query::unified::UnifiedRecord::new();
        record.set("quantity", Value::Integer(3));
        let quantity = Expr::Column {
            field: FieldRef::TableColumn {
                table: "o".into(),
                column: "quantity".into(),
            },
            span: Span::synthetic(),
        };
        let call = |name: &str, args| Expr::FunctionCall {
            name: name.into(),
            args,
            span: Span::synthetic(),
        };
        let missing = || Expr::lit(Value::text("__browser_expression_fixture_missing__"));
        let expression = call("config", vec![missing(), quantity.clone()]);
        assert_eq!(
            super::evaluate_runtime_expr(&expression, &record, Some("orders"), Some("o")),
            Some(Value::text("o.quantity")),
            "a bare special default remains path text"
        );
        let expression = call(
            "kv",
            vec![
                missing(),
                missing(),
                Expr::BinaryOp {
                    op: BinOp::Add,
                    lhs: Box::new(quantity),
                    rhs: Box::new(Expr::lit(Value::Integer(2))),
                    span: Span::synthetic(),
                },
            ],
        );
        assert_eq!(
            super::evaluate_runtime_expr(&expression, &record, Some("orders"), Some("o")),
            Some(Value::Integer(5))
        );
        let mut expression = Expr::lit(Value::Integer(7));
        for _ in 0..=reddb_rql::limits::JSON_LITERAL_MAX_DEPTH {
            expression = call("CONFIG", vec![missing(), expression]);
        }
        assert_eq!(
            super::evaluate_runtime_expr(&expression, &record, Some("orders"), Some("o")),
            Some(Value::Null),
            "special defaults must not reset the visited depth"
        );
    }

    #[test]
    fn test_shared_time_bucket_conversion_and_overflow() {
        for (args, expected) in [
            (
                vec![Value::text("5ms"), Value::TimestampMs(17)],
                Some(Value::UnsignedInteger(15_000_000)),
            ),
            (
                vec![Value::Integer(5), Value::Integer(17)],
                Some(Value::UnsignedInteger(15)),
            ),
            (
                vec![Value::text("raw"), Value::Integer(17)],
                Some(Value::UnsignedInteger(17)),
            ),
            (vec![Value::text("5ms"), Value::TimestampMs(i64::MAX)], None),
            (
                vec![Value::text("18446744073709551615s"), Value::Integer(17)],
                None,
            ),
        ] {
            assert_eq!(dispatch_builtin_function("TIME_BUCKET", &args), expected);
        }
    }

    #[test]
    fn test_shared_text_bounds_and_regex() {
        assert_eq!(
            dispatch_builtin_function(
                "SUBSTRING",
                &[
                    Value::text("aé🦀z"),
                    Value::Integer(2),
                    Value::Integer(4_294_967_296),
                ]
            ),
            Some(Value::text("é🦀z"))
        );
        assert_eq!(
            dispatch_builtin_function(
                "SUBSTRING",
                &[Value::text("aé🦀z"), Value::Integer(i64::MAX),]
            ),
            Some(Value::text(""))
        );
        assert_eq!(
            dispatch_builtin_function("RIGHT", &[Value::text("aé🦀z"), Value::Integer(i64::MIN),]),
            Some(Value::text(""))
        );
        assert_eq!(
            dispatch_builtin_function(
                "SUBSTRING",
                &[Value::text("é42z"), Value::text("é([0-9]+)z"),]
            ),
            Some(Value::text("42"))
        );
        assert_eq!(
            dispatch_builtin_function("SUBSTRING", &[Value::text("b"), Value::text("(a)?b"),]),
            Some(Value::Null)
        );
    }

    #[test]
    fn test_money_constructor_two_args() {
        let value = dispatch_builtin_function(
            "MONEY",
            &[
                Value::AssetCode("BTC".to_string()),
                Value::text("0.125".to_string()),
            ],
        )
        .unwrap();
        assert_eq!(
            value,
            Value::Money {
                asset_code: "BTC".to_string(),
                minor_units: 125,
                scale: 3,
            }
        );
    }

    #[test]
    fn test_money_extractors() {
        let money = Value::Money {
            asset_code: "USDT".to_string(),
            minor_units: 12345,
            scale: 2,
        };
        assert_eq!(
            dispatch_builtin_function("MONEY_ASSET", std::slice::from_ref(&money)).unwrap(),
            Value::AssetCode("USDT".to_string())
        );
        assert_eq!(
            dispatch_builtin_function("MONEY_MINOR", std::slice::from_ref(&money)).unwrap(),
            Value::BigInt(12345)
        );
        assert_eq!(
            dispatch_builtin_function("MONEY_SCALE", std::slice::from_ref(&money)).unwrap(),
            Value::Integer(2)
        );
    }

    // ─────────────────────────────────────────────────────────────
    // JSON functions (Phase 1.4 PG parity)
    // ─────────────────────────────────────────────────────────────

    fn json_text(s: &str) -> Value {
        Value::text(s.to_string())
    }

    #[test]
    fn json_extract_scalar_and_nested() {
        let doc = json_text(r#"{"a":1,"b":{"c":"hello","d":[10,20,30]}}"#);
        // Top-level scalar.
        assert_eq!(
            dispatch_builtin_function("JSON_EXTRACT", &[doc.clone(), json_text("$.a")]).unwrap(),
            Value::text("1".to_string())
        );
        // Nested string (quoted for JSON_EXTRACT).
        assert_eq!(
            dispatch_builtin_function("JSON_EXTRACT", &[doc.clone(), json_text("$.b.c")]).unwrap(),
            Value::text("\"hello\"".to_string())
        );
        // Unquoted via JSON_EXTRACT_TEXT.
        assert_eq!(
            dispatch_builtin_function("JSON_EXTRACT_TEXT", &[doc.clone(), json_text("$.b.c")])
                .unwrap(),
            Value::text("hello".to_string())
        );
        // Array index.
        assert_eq!(
            dispatch_builtin_function("JSON_EXTRACT_TEXT", &[doc.clone(), json_text("$.b.d[1]")])
                .unwrap(),
            Value::text("20".to_string())
        );
        // Missing path → Null.
        assert_eq!(
            dispatch_builtin_function("JSON_EXTRACT", &[doc, json_text("$.missing")]).unwrap(),
            Value::Null
        );
    }

    #[test]
    fn json_array_length_and_typeof() {
        let arr = json_text(r#"[1,2,3,4]"#);
        assert_eq!(
            dispatch_builtin_function("JSON_ARRAY_LENGTH", &[arr.clone()]).unwrap(),
            Value::Integer(4)
        );
        assert_eq!(
            dispatch_builtin_function("JSON_TYPEOF", &[arr]).unwrap(),
            Value::text("array".to_string())
        );
        assert_eq!(
            dispatch_builtin_function("JSON_TYPEOF", &[json_text(r#"{"k":1}"#)]).unwrap(),
            Value::text("object".to_string())
        );
        assert_eq!(
            dispatch_builtin_function("JSON_TYPEOF", &[json_text("null")]).unwrap(),
            Value::text("null".to_string())
        );
    }

    #[test]
    fn json_valid_accepts_and_rejects() {
        assert_eq!(
            dispatch_builtin_function("JSON_VALID", &[json_text(r#"{"a":1}"#)]).unwrap(),
            Value::Boolean(true)
        );
        assert_eq!(
            dispatch_builtin_function("JSON_VALID", &[json_text("not json")]).unwrap(),
            Value::Boolean(false)
        );
    }

    #[test]
    fn json_array_and_object_builders() {
        // JSON_ARRAY(1, "x", true) → [1,"x",true]
        let arr = dispatch_builtin_function(
            "JSON_ARRAY",
            &[
                Value::Integer(1),
                Value::text("x".to_string()),
                Value::Boolean(true),
            ],
        )
        .unwrap();
        if let Value::Json(bytes) = arr {
            let text = String::from_utf8(bytes).unwrap();
            assert_eq!(text, r#"[1,"x",true]"#);
        } else {
            panic!("expected Json value");
        }

        // JSON_OBJECT("k1", 1, "k2", "v") → {"k1":1,"k2":"v"}
        let obj = dispatch_builtin_function(
            "JSON_OBJECT",
            &[
                Value::text("k1".to_string()),
                Value::Integer(1),
                Value::text("k2".to_string()),
                Value::text("v".to_string()),
            ],
        )
        .unwrap();
        if let Value::Json(bytes) = obj {
            let text = String::from_utf8(bytes).unwrap();
            // Map keeps insertion order via BTreeMap (sorted alphabetically).
            assert_eq!(text, r#"{"k1":1,"k2":"v"}"#);
        } else {
            panic!("expected Json value");
        }
    }

    #[test]
    fn json_set_updates_existing_and_creates_new() {
        let doc = json_text(r#"{"a":1,"b":{"c":"x"}}"#);

        // Update existing nested field.
        let out = dispatch_builtin_function(
            "JSON_SET",
            &[
                doc.clone(),
                json_text("$.b.c"),
                Value::text("new".to_string()),
            ],
        )
        .unwrap();
        if let Value::Json(bytes) = out {
            let text = String::from_utf8(bytes).unwrap();
            // JSON_EXTRACT_TEXT on result must return "new".
            let extracted = dispatch_builtin_function(
                "JSON_EXTRACT_TEXT",
                &[Value::text(text), json_text("$.b.c")],
            )
            .unwrap();
            assert_eq!(extracted, Value::text("new".to_string()));
        } else {
            panic!("expected Json");
        }

        // Create a new nested path.
        let out = dispatch_builtin_function(
            "JSON_SET",
            &[doc, json_text("$.new.deep"), Value::Integer(42)],
        )
        .unwrap();
        if let Value::Json(bytes) = out {
            let text = String::from_utf8(bytes).unwrap();
            let extracted = dispatch_builtin_function(
                "JSON_EXTRACT_TEXT",
                &[Value::text(text), json_text("$.new.deep")],
            )
            .unwrap();
            assert_eq!(extracted, Value::text("42".to_string()));
        } else {
            panic!("expected Json");
        }
    }
}