use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap};
use sha2::{Digest, Sha256};
use super::ast::{
AddWithOption, AssembleStmt, BatchEntry, BatchStmt, CalQuery, CalStatement, CoalesceStmt,
Comparator, Condition, DescribeStmt, DescribeTarget, ExistsStmt, ExplainStmt, Extractor,
FormatClause, GrainTypePlural, HistoryStmt, LetBinding, PipelineStage, RecallStmt, SetOp,
SetOpStmt, Source, Value, WithOption,
};
use super::errors::{CalError, Span};
use super::facade::{AssemblyManifest, CalStoreFacade};
use crate::store_types::{AddOptions, DiversityConfig, RecallParams};
use areev_core::error::{AreevError, Hash};
#[derive(Debug, Clone)]
pub struct CalExecutorConfig {
pub max_limit: u64,
pub default_limit: u64,
pub tier1_enabled: bool,
pub namespace_override: Option<String>,
pub user_id_override: Option<String>,
pub allow_destructive_ops: bool,
pub redact_budget_metadata: bool,
pub assembly_manifest_sample_rate: f64,
pub caller_scopes: Vec<String>,
}
impl Default for CalExecutorConfig {
fn default() -> Self {
Self {
max_limit: 1000,
default_limit: 50,
tier1_enabled: true,
allow_destructive_ops: true,
namespace_override: None,
user_id_override: None,
redact_budget_metadata: false,
assembly_manifest_sample_rate: 0.0,
caller_scopes: vec![],
}
}
}
#[derive(Debug, serde::Serialize)]
pub struct CalExecResult {
pub query: String,
pub query_hash: String,
pub result: CalResultPayload,
pub warnings: Vec<String>,
pub anonymized: Option<serde_json::Value>,
pub metadata: CalMetadata,
}
impl CalExecResult {
pub fn payload_json(&self) -> serde_json::Result<serde_json::Value> {
let mut v = serde_json::to_value(&self.result)?;
if let Some(a) = &self.anonymized {
if let Some(obj) = v.as_object_mut() {
obj.insert("anonymized".into(), a.clone());
}
}
if !self.warnings.is_empty() {
if let Some(obj) = v.as_object_mut() {
obj.insert("warnings".into(), serde_json::json!(self.warnings));
}
}
Ok(v)
}
}
#[derive(Debug, serde::Serialize)]
pub struct CalMetadata {
pub version: u32,
pub statement_type: String,
pub execution_time_ms: u64,
pub result_count: usize,
}
#[derive(Debug, serde::Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CalResultPayload {
Grains {
grains: Vec<CalGrainResult>,
total_available: Option<usize>,
},
Exists { exists: bool, hash: String },
Count { count: usize },
GroupCounts {
field: String,
groups: Vec<CalGrainResult>,
total_available: Option<usize>,
},
Granted { principal: String, object: String, hash: String },
Revoked { principal: String, grants_touched: usize },
GrantList { grants: Vec<serde_json::Value> },
LoopRan { run: serde_json::Value },
Reviewed { hash: String, decision: String },
RecApplied { hash: String, rollbackable: bool },
RecRolledBack { hash: String },
Remembered { hash: String },
EntityAt {
grain: Option<serde_json::Value>,
axis: String,
at_ms: i64,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
grains: Vec<serde_json::Value>,
},
RunTrace { run_id: String, trace: serde_json::Value },
RunsTouching { hash: String, runs: Vec<String> },
DerivedFrom { hash: String, grains: Vec<serde_json::Value> },
Forks { forks: Vec<serde_json::Value> },
Merged { hash: String, subject: String, relation: String, object: String },
RelatedEntities { start: String, entities: Vec<String> },
NoveltyMatches { matches: Vec<serde_json::Value> },
History { versions: Vec<CalVersionResult> },
Describe { info: serde_json::Value },
Explain { plan: CalQueryPlan },
Batch {
results: HashMap<String, CalResultPayload>,
},
Assembled {
grains: Vec<CalGrainResult>,
sources: Vec<super::assemble::SourceMeta>,
total_tokens: u32,
budget_limit: Option<u32>,
progressive: bool,
total_available: Option<usize>,
},
Diff {
source_hash: String,
target_hash: String,
changes: Vec<super::ast::FieldDiff>,
},
Formatted {
text: String,
format: String,
grain_count: usize,
#[serde(skip_serializing)]
grains: Vec<CalGrainResult>,
},
MultiFormatted {
formats: HashMap<String, String>,
grain_count: usize,
#[serde(skip_serializing)]
grains: Vec<CalGrainResult>,
},
Added {
hash: String,
grain_type: String,
#[serde(skip_serializing_if = "Option::is_none")]
extracted_count: Option<usize>,
#[serde(skip_serializing_if = "Vec::is_empty")]
extraction_warnings: Vec<String>,
},
Superseded { old_hash: String, new_hash: String },
Accumulated {
old_hash: String,
new_hash: String,
deltas: Vec<AccumulatedDelta>,
},
TemplateDefined { name: String },
TemplateDropped { name: String },
QueryDefined { name: String },
QueryDropped { name: String },
StreamAssemble {
assemble: Box<super::ast::AssembleStmt>,
with_options: Vec<super::ast::WithOption>,
},
Forgotten { target: String, count: u64 },
Purged { count: usize },
SubjectReport {
subject: String,
identity_names: Vec<String>,
grains: Vec<serde_json::Value>,
},
Unsupported { statement: String, message: String },
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct CalGrainResult {
pub hash: String,
pub grain_type: String,
pub score: f64,
pub fields: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub score_breakdown: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub explanation: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub relative_time: Option<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub is_deterministic: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub contested_by: Option<Vec<String>>,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct AccumulatedDelta {
pub field: String,
pub old_value: f64,
pub new_value: f64,
}
#[derive(Debug, serde::Serialize)]
pub struct CalVersionResult {
pub hash: String,
pub object: String,
pub created_at: i64,
pub confidence: f64,
pub superseded_by: Option<String>,
}
#[derive(Debug, serde::Serialize)]
pub struct CalQueryPlan {
pub statement_type: String,
pub grain_type: Option<String>,
pub query_routing: String,
pub index_usage: Vec<String>,
pub estimated_cost: String,
pub filters: Vec<String>,
pub pipeline: Vec<String>,
}
const MAX_LET_BINDINGS: usize = 5;
const MAX_GRAINS_PER_LET: usize = 1000;
#[derive(Debug, Clone)]
pub enum LetValue {
Grains(Vec<CalGrainResult>),
Extracted(Vec<String>),
}
#[derive(Debug)]
pub struct LetScope {
bindings: HashMap<String, LetValue>,
}
impl LetScope {
fn to_values(&self) -> HashMap<String, Vec<String>> {
self.bindings
.iter()
.map(|(name, value)| {
let values = match value {
LetValue::Extracted(v) => v.clone(),
LetValue::Grains(g) => g.iter().map(|r| r.hash.clone()).collect(),
};
(name.clone(), values)
})
.collect()
}
}
impl LetScope {
pub fn evaluate(
let_bindings: &[LetBinding],
executor: &CalExecutor,
store: &dyn CalStoreFacade,
query: &CalQuery,
warnings: &mut Vec<String>,
) -> std::result::Result<Self, CalError> {
if let_bindings.len() > MAX_LET_BINDINGS {
return Err(CalError::TooManyLetBindings {
count: let_bindings.len(),
max: MAX_LET_BINDINGS,
span: let_bindings.first().and_then(|b| b.span),
});
}
let mut scope = LetScope {
bindings: HashMap::new(),
};
for binding in let_bindings {
if scope.bindings.contains_key(&binding.name) {
return Err(CalError::DuplicateParameter {
name: binding.name.clone(),
span: binding.span,
});
}
let surrogate = CalQuery {
version: query.version,
statement: (*binding.source).clone(),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: scope.to_values(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let payload =
executor.execute_statement(&surrogate.statement, store, &surrogate, warnings)?;
let grains = extract_grains(payload);
let grains = if grains.len() > MAX_GRAINS_PER_LET {
warnings.push(format!(
"LET ${} produced {} grains (capped to {})",
binding.name,
grains.len(),
MAX_GRAINS_PER_LET
));
grains.into_iter().take(MAX_GRAINS_PER_LET).collect()
} else {
grains
};
let value = match binding.extractor {
Extractor::Subjects => {
let values: Vec<String> = grains
.iter()
.filter_map(|g| {
json_field(&g.fields, "subject")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
})
.collect();
LetValue::Extracted(values)
}
Extractor::Objects => {
let values: Vec<String> = grains
.iter()
.filter_map(|g| {
json_field(&g.fields, "object")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
})
.collect();
LetValue::Extracted(values)
}
Extractor::Hashes => {
let values: Vec<String> = grains.iter().map(|g| g.hash.clone()).collect();
LetValue::Extracted(values)
}
};
scope.bindings.insert(binding.name.clone(), value);
}
Ok(scope)
}
pub fn resolve(&self, name: &str) -> std::result::Result<&LetValue, CalError> {
self.bindings
.get(name)
.ok_or_else(|| CalError::UnboundParameter {
name: name.to_string(),
span: None,
})
}
}
impl Drop for LetScope {
fn drop(&mut self) {
self.bindings.clear();
self.bindings.shrink_to_fit();
}
}
const PLAN_CACHE_MAX: usize = 256;
pub struct CalExecutor {
config: CalExecutorConfig,
plan_cache: std::sync::Mutex<HashMap<String, crate::ast::CalQuery>>,
governance: Option<std::sync::Arc<dyn crate::governance::GovernanceHost>>,
}
impl CalExecutor {
pub fn new(config: CalExecutorConfig) -> Self {
Self { config, governance: None, plan_cache: Default::default() }
}
pub fn parse_cached(
&self,
input: &str,
) -> std::result::Result<crate::ast::CalQuery, CalError> {
if let Ok(cache) = self.plan_cache.lock() {
if let Some(q) = cache.get(input) {
return Ok(q.clone());
}
}
let query = crate::parser::parse(input)?;
if let Ok(mut cache) = self.plan_cache.lock() {
if cache.len() >= PLAN_CACHE_MAX {
cache.clear();
}
cache.insert(input.to_string(), query.clone());
}
Ok(query)
}
pub fn plan_cache_len(&self) -> usize {
self.plan_cache.lock().map(|c| c.len()).unwrap_or(0)
}
pub fn with_governance(
mut self,
host: std::sync::Arc<dyn crate::governance::GovernanceHost>,
) -> Self {
self.governance = Some(host);
self
}
pub fn config(&self) -> &CalExecutorConfig {
&self.config
}
pub fn with_defaults() -> Self {
Self::new(CalExecutorConfig::default())
}
fn check_caller_scope(&self, stmt: &CalStatement) -> std::result::Result<(), CalError> {
let scopes = &self.config.caller_scopes;
if scopes.is_empty() {
return Ok(()); }
let required = required_scope_for_statement(stmt);
if scopes.iter().any(|s| s == "admin") {
return Ok(());
}
if scopes.iter().any(|s| s == required) {
return Ok(());
}
if required == "read" && scopes.iter().any(|s| s == "write") {
return Ok(());
}
Err(CalError::InsufficientScope {
required: required.to_string(),
statement: statement_type_name(stmt),
})
}
pub fn execute(
&self,
input: &str,
store: &dyn CalStoreFacade,
) -> std::result::Result<CalExecResult, CalError> {
let start = std::time::Instant::now();
let query = self.parse_cached(input)?;
let query_hash = compute_query_hash(input);
let mut query = query;
let mut exec_warnings: Vec<String> = Vec::new();
if !query.let_bindings.is_empty() {
let scope = LetScope::evaluate(
&query.let_bindings,
self,
store,
&query,
&mut exec_warnings,
)?;
query.let_values = scope.to_values();
}
if let CalStatement::Recall(ref r) = query.statement {
Self::validate_pipeline_fields(&query.pipeline, &r.grain_type)?;
}
let payload =
self.execute_statement(&query.statement, store, &query, &mut exec_warnings)?;
let (payload, grouped_by) = self.apply_pipeline(payload, &query.pipeline, &mut exec_warnings)?;
let payload = apply_format_clause(
payload,
&query.format,
grouped_by.as_deref(),
RenderInputs {
user_vars: &query.user_vars,
store,
disclosure: disclosure_of(&query.with_options),
},
None,
&mut exec_warnings,
)?;
let mut warnings: Vec<String> = query.warnings.iter().map(|w| w.to_string()).collect();
warnings.extend(exec_warnings);
let elapsed_ms = start.elapsed().as_millis() as u64;
let stmt_type = statement_type_name(&query.statement);
let result_count = count_payload_results(&payload);
Ok(CalExecResult {
query: input.to_string(),
query_hash,
result: payload,
warnings,
anonymized: store.anon_egress_report(),
metadata: CalMetadata {
version: query.version.0,
statement_type: stmt_type,
execution_time_ms: elapsed_ms,
result_count,
},
})
}
pub fn execute_parsed(
&self,
query: crate::ast::CalQuery,
original_text: &str,
store: &dyn CalStoreFacade,
) -> std::result::Result<CalExecResult, CalError> {
let start = std::time::Instant::now();
let query_hash = compute_query_hash(original_text);
let mut query = query;
let mut exec_warnings: Vec<String> = Vec::new();
if !query.let_bindings.is_empty() {
let scope = LetScope::evaluate(
&query.let_bindings,
self,
store,
&query,
&mut exec_warnings,
)?;
query.let_values = scope.to_values();
}
if let CalStatement::Recall(ref r) = query.statement {
Self::validate_pipeline_fields(&query.pipeline, &r.grain_type)?;
}
let payload =
self.execute_statement(&query.statement, store, &query, &mut exec_warnings)?;
let (payload, grouped_by) = self.apply_pipeline(payload, &query.pipeline, &mut exec_warnings)?;
let payload = apply_format_clause(
payload,
&query.format,
grouped_by.as_deref(),
RenderInputs {
user_vars: &query.user_vars,
store,
disclosure: disclosure_of(&query.with_options),
},
None,
&mut exec_warnings,
)?;
let mut warnings: Vec<String> = query.warnings.iter().map(|w| w.to_string()).collect();
warnings.extend(exec_warnings);
let elapsed_ms = start.elapsed().as_millis() as u64;
let stmt_type = statement_type_name(&query.statement);
let result_count = count_payload_results(&payload);
Ok(CalExecResult {
query: original_text.to_string(),
query_hash,
result: payload,
warnings,
anonymized: store.anon_egress_report(),
metadata: CalMetadata {
version: query.version.0,
statement_type: stmt_type,
execution_time_ms: elapsed_ms,
result_count,
},
})
}
fn inject_write_namespace(
&self,
fields: &mut serde_json::Map<String, serde_json::Value>,
store: &dyn CalStoreFacade,
) {
if let Some(ref ns) = self.config.namespace_override {
fields.insert("namespace".into(), serde_json::Value::String(ns.clone()));
} else if !fields.contains_key("namespace") {
if let Some(ns) = store.default_namespace() {
fields.insert(
"namespace".into(),
serde_json::Value::String(ns.to_string()),
);
}
}
}
pub(super) fn execute_statement_internal(
&self,
stmt: &CalStatement,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
self.execute_statement(stmt, store, query, exec_warnings)
}
pub fn execute_statement(
&self,
stmt: &CalStatement,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
self.check_caller_scope(stmt)?;
match stmt {
CalStatement::Recall(recall) => {
self.execute_recall(recall, store, query, exec_warnings)
}
CalStatement::Exists(exists) => {
self.execute_exists(exists, store, exec_warnings, &query.let_values)
}
CalStatement::History(history) => {
self.execute_history(history, store, exec_warnings, &query.let_values)
}
CalStatement::Describe(describe) => self.execute_describe(describe, store),
CalStatement::Explain(explain) => self.execute_explain(explain, store, query),
CalStatement::Batch(batch) => self.execute_batch(batch, store, exec_warnings),
CalStatement::Coalesce(coalesce) => {
self.execute_coalesce(coalesce, store, query, exec_warnings)
}
CalStatement::SetOp(set_op) => self.execute_set_op(set_op, store, query, exec_warnings),
CalStatement::Assemble(assemble) => {
self.execute_assemble(assemble, store, query, exec_warnings)
}
CalStatement::Add(add) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "ADD".into(),
span: add.span,
});
}
let add_type = add.grain_type.as_str();
if !areev_core::types::registry::add_via_set_names().any(|n| n == add_type) {
let shapeable: Vec<&str> = areev_core::types::registry::add_via_set_names().collect();
return Ok(CalResultPayload::Unsupported {
statement: "add".into(),
message: format!(
"Grain type '{}' cannot be created via ADD. Addable types: {}.",
add_type,
shapeable.join(", ")
),
});
}
for fa in &add.fields {
if let super::ast::Value::Parameter { name } = &fa.value {
return Ok(CalResultPayload::Unsupported {
statement: "add".into(),
message: format!(
"Unresolved parameter ${} in SET clause. Parameters must be bound via LET.",
name
),
});
}
}
let mut fields = serde_json::Map::new();
for fa in &add.fields {
fields.insert(fa.field.clone(), cal_value_to_json(&fa.value));
}
if !add.reason.is_empty() {
fields.insert(
"add_reason".into(),
serde_json::Value::String(add.reason.clone()),
);
}
if add
.with_options
.iter()
.any(|o| matches!(o, super::ast::AddWithOption::Occurrence))
{
exec_warnings.push(
"CAL-W014: WITH occurrence is inert here — tool occurrences are \
recorded via record_tool_call (every binding), which stamps the \
per-call identity that keeps retries distinct."
.to_string(),
);
}
self.inject_write_namespace(&mut fields, store);
let options = build_add_options(&add.with_options, exec_warnings);
match store.cal_add_with_options(add.grain_type.as_str(), &fields, options) {
Ok(result) => Ok(CalResultPayload::Added {
hash: hex::encode(result.hash.as_bytes()),
grain_type: add.grain_type.as_str().to_string(),
extracted_count: if result.extracted_count > 0 {
Some(result.extracted_count)
} else {
None
},
extraction_warnings: result.extraction_warnings,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "add".into(),
message: format!("ADD failed: {e}"),
}),
}
}
CalStatement::Supersede(sup) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "SUPERSEDE".into(),
span: sup.span,
});
}
let raw_hash = sup.hash.strip_prefix("sha256:").unwrap_or(&sup.hash);
let old_hash = match Hash::from_hex(raw_hash) {
Ok(h) => h,
Err(e) => {
return Ok(CalResultPayload::Unsupported {
statement: "supersede".into(),
message: format!("invalid hash: {e}"),
})
}
};
let old_grain = match store.get(&old_hash) {
Ok(g) => g,
Err(e) => {
return Ok(CalResultPayload::Unsupported {
statement: "supersede".into(),
message: format!("cannot retrieve grain for SUPERSEDE: {e}"),
})
}
};
let grain_type_str = old_grain.grain_type.as_str();
let mut fields: serde_json::Map<String, serde_json::Value> =
old_grain.fields.into_iter().collect();
for sc in &sup.set_clauses {
fields.insert(sc.field.clone(), cal_value_to_json(&sc.value));
}
if !sup.reason.is_empty() {
fields.insert(
"supersede_reason".into(),
serde_json::Value::String(sup.reason.clone()),
);
}
match store.cal_supersede(&old_hash, grain_type_str, &fields) {
Ok(new_hash) => Ok(CalResultPayload::Superseded {
old_hash: hex::encode(old_hash.as_bytes()),
new_hash: hex::encode(new_hash.as_bytes()),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "supersede".into(),
message: format!("SUPERSEDE failed: {e}"),
}),
}
}
CalStatement::AddWorkflow(wf) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "ADD WORKFLOW".into(),
span: wf.span,
});
}
let mut fields = serde_json::Map::new();
fields.insert("name".into(), serde_json::Value::String(wf.name.clone()));
fields.insert(
"nodes".into(),
serde_json::Value::Array(
wf.nodes
.iter()
.map(|n| serde_json::Value::String(n.clone()))
.collect(),
),
);
let edges_json: Vec<serde_json::Value> = wf
.edges
.iter()
.map(|e| {
let mut m = serde_json::Map::new();
m.insert("src".into(), serde_json::Value::String(e.src.clone()));
m.insert("dst".into(), serde_json::Value::String(e.dst.clone()));
if let Some(ref c) = e.cond {
m.insert("cond".into(), serde_json::Value::String(c.clone()));
}
serde_json::Value::Object(m)
})
.collect();
fields.insert("edges".into(), serde_json::Value::Array(edges_json));
if !wf.bindings.is_empty() {
let mut bind_map = serde_json::Map::new();
for b in &wf.bindings {
bind_map.insert(
b.node.clone(),
serde_json::Value::String(format!("sha256:{}", b.hash)),
);
}
fields.insert("bindings".into(), serde_json::Value::Object(bind_map));
}
let mut retries_map = serde_json::Map::new();
for e in &wf.edges {
if let Some(r) = e.repeat {
let merged = retries_map
.get(&e.dst)
.and_then(|v| v.as_u64())
.map_or(r, |prev| (prev as u32).max(r));
retries_map.insert(
e.dst.clone(),
serde_json::Value::Number(serde_json::Number::from(merged)),
);
}
}
if !retries_map.is_empty() {
fields.insert("retries".into(), serde_json::Value::Object(retries_map));
}
self.inject_write_namespace(&mut fields, store);
let options = build_add_options(&wf.with_options, exec_warnings);
match store.cal_add_with_options("workflow", &fields, options) {
Ok(result) => Ok(CalResultPayload::Added {
hash: hex::encode(result.hash.as_bytes()),
grain_type: "workflow".into(),
extracted_count: None,
extraction_warnings: vec![],
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "add workflow".into(),
message: format!("ADD workflow failed: {e}"),
}),
}
}
CalStatement::SupersedeWorkflow(wf) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "SUPERSEDE WORKFLOW".into(),
span: wf.span,
});
}
let raw_hash = wf.hash.strip_prefix("sha256:").unwrap_or(&wf.hash);
let old_hash = match Hash::from_hex(raw_hash) {
Ok(h) => h,
Err(e) => {
return Ok(CalResultPayload::Unsupported {
statement: "supersede workflow".into(),
message: format!("invalid hash: {e}"),
})
}
};
let mut fields = serde_json::Map::new();
fields.insert(
"nodes".into(),
serde_json::Value::Array(
wf.nodes
.iter()
.map(|n| serde_json::Value::String(n.clone()))
.collect(),
),
);
let edges_json: Vec<serde_json::Value> = wf
.edges
.iter()
.map(|e| {
let mut m = serde_json::Map::new();
m.insert("src".into(), serde_json::Value::String(e.src.clone()));
m.insert("dst".into(), serde_json::Value::String(e.dst.clone()));
if let Some(ref c) = e.cond {
m.insert("cond".into(), serde_json::Value::String(c.clone()));
}
serde_json::Value::Object(m)
})
.collect();
fields.insert("edges".into(), serde_json::Value::Array(edges_json));
if !wf.bindings.is_empty() {
let mut bind_map = serde_json::Map::new();
for b in &wf.bindings {
bind_map.insert(
b.node.clone(),
serde_json::Value::String(format!("sha256:{}", b.hash)),
);
}
fields.insert("bindings".into(), serde_json::Value::Object(bind_map));
}
let mut retries_map = serde_json::Map::new();
for e in &wf.edges {
if let Some(r) = e.repeat {
let merged = retries_map
.get(&e.dst)
.and_then(|v| v.as_u64())
.map_or(r, |prev| (prev as u32).max(r));
retries_map.insert(
e.dst.clone(),
serde_json::Value::Number(serde_json::Number::from(merged)),
);
}
}
if !retries_map.is_empty() {
fields.insert("retries".into(), serde_json::Value::Object(retries_map));
}
if !wf.reason.is_empty() {
fields.insert(
"supersede_reason".into(),
serde_json::Value::String(wf.reason.clone()),
);
}
match store.cal_supersede(&old_hash, "workflow", &fields) {
Ok(new_hash) => Ok(CalResultPayload::Superseded {
old_hash: hex::encode(old_hash.as_bytes()),
new_hash: hex::encode(new_hash.as_bytes()),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "supersede workflow".into(),
message: format!("SUPERSEDE workflow failed: {e}"),
}),
}
}
CalStatement::Accumulate(acc) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "ACCUMULATE".into(),
span: acc.span,
});
}
let add_ops: Vec<(String, f64)> = acc
.add_ops
.iter()
.map(|op| (op.field.clone(), op.delta))
.collect();
let mut set_map = serde_json::Map::new();
for s in &acc.set_ops {
set_map.insert(s.field.clone(), cal_value_to_json(&s.value));
}
if !acc.reason.is_empty() {
set_map.insert(
"supersede_reason".into(),
serde_json::Value::String(acc.reason.clone()),
);
}
match store.cal_accumulate(
acc.grain_type.as_str(),
&acc.target,
&add_ops,
&set_map,
&acc.reason,
) {
Ok(result) => Ok(CalResultPayload::Accumulated {
old_hash: hex::encode(result.old_hash.as_bytes()),
new_hash: hex::encode(result.new_hash.as_bytes()),
deltas: result
.applied_deltas
.iter()
.map(|(f, old, new)| AccumulatedDelta {
field: f.clone(),
old_value: *old,
new_value: *new,
})
.collect(),
}),
Err(areev_core::error::AreevError::AccumulateRetryExhausted) => {
tracing::error!(
cal_code = "CAL-E083",
"ACCUMULATE retry budget exhausted under sustained contention"
);
let (subject, relation) = match &acc.target {
super::ast::AccumulateTarget::TipResolved {
subject, relation, ..
} => (subject.clone(), relation.clone()),
super::ast::AccumulateTarget::Hash { .. } => {
(String::new(), String::new())
}
};
Err(CalError::AccumulateRetryExhausted {
subject,
relation,
span: acc.span,
})
}
Err(areev_core::error::AreevError::AccumulateInternal(detail)) => {
tracing::error!(
cal_code = "CAL-E084",
error = %detail,
"ACCUMULATE internal failure"
);
Err(CalError::AccumulateInternal { span: acc.span })
}
Err(areev_core::error::AreevError::AccumulateBackpressureRejected) => {
tracing::warn!(cal_code = "CAL-E085", "ACCUMULATE backpressure rejected");
let (subject, relation) = match &acc.target {
super::ast::AccumulateTarget::TipResolved {
subject, relation, ..
} => (subject.clone(), relation.clone()),
super::ast::AccumulateTarget::Hash { .. } => {
(String::new(), String::new())
}
};
Err(CalError::AccumulateBackpressureRejected {
subject,
relation,
span: acc.span,
})
}
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "accumulate".into(),
message: format!("ACCUMULATE failed: {e}"),
}),
}
}
CalStatement::Revert(_) => Ok(CalResultPayload::Unsupported {
statement: "revert".into(),
message: "Tier 1 REVERT semantics are not yet defined. \
Use EXPLAIN to preview without execution."
.into(),
}),
CalStatement::Forget(forget) => {
if !self.config.allow_destructive_ops {
return Ok(CalResultPayload::Unsupported {
statement: "forget".into(),
message: "Destructive operations are disabled for this session \
(started with --no-destructive-ops)."
.into(),
});
}
match &forget.target {
super::ast::ForgetTarget::Hash { hash } => {
let h = Hash::from_hex(hash).map_err(|_| CalError::InvalidHash {
found: hash.clone(),
span: forget.span,
})?;
let done = if forget.override_hold {
store.cal_delete_overriding(
&h,
forget.reason.as_deref().unwrap_or_default(),
)
} else {
store.cal_delete(&h, forget.reason.as_deref())
};
match done {
Ok(()) => Ok(CalResultPayload::Forgotten {
target: format!("hash:{hash}"),
count: 1,
}),
Err(e) => Err(map_store_err(e, forget.span)),
}
}
super::ast::ForgetTarget::User { user_id } => {
let because = match forget.reason.as_deref() {
Some(r) if !r.trim().is_empty() => r,
_ => {
return Err(CalError::MissingReason { span: forget.span });
}
};
let done = if forget.override_hold {
store.cal_forget_user_overriding(
user_id,
forget.text_mentions,
because,
)
} else {
store.cal_forget_user(user_id, forget.text_mentions, because)
};
match done {
Ok(proof) => Ok(CalResultPayload::Forgotten {
target: format!("subject:{user_id}"),
count: proof.count,
}),
Err(e) => Err(map_store_err(e, forget.span)),
}
}
super::ast::ForgetTarget::Scope { scope } => {
match store.cal_forget_scope(scope) {
Ok(proof) => Ok(CalResultPayload::Forgotten {
target: format!("scope:{scope}"),
count: proof.count,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "forget".into(),
message: format!("FORGET SCOPE failed: {e}"),
}),
}
}
}
}
CalStatement::DefineTemplate(def) => {
store
.define_template(
&def.name,
&def.source,
def.description.as_deref(),
def.parent.as_deref(),
&def.grain_types,
)
.map_err(|e| {
let s = e.to_string();
if s.contains("AUT-E") {
return CalError::NotAuthorized {
detail: s,
span: None,
store_code: authz_store_code(&e),
};
}
if s.contains("CAL-E04")
|| s.contains("CAL-E11")
|| s.contains("Unknown template")
|| s.contains("Invalid template")
|| s.contains("syntax")
{
CalError::TemplateSyntaxError {
detail: s,
span: None,
}
} else {
CalError::TemplateInvalidName {
name: def.name.clone(),
span: None,
}
}
})?;
Ok(CalResultPayload::TemplateDefined {
name: def.name.clone(),
})
}
CalStatement::DropTemplate(drop) => {
if !self.config.allow_destructive_ops {
return Ok(CalResultPayload::Unsupported {
statement: "drop_template".into(),
message: "Destructive operations are disabled for this session \
(started with --no-destructive-ops)."
.into(),
});
}
match store.drop_template(&drop.name) {
Ok(()) => Ok(CalResultPayload::TemplateDropped {
name: drop.name.clone(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "drop_template".into(),
message: format!("DROP TEMPLATE failed: {e}"),
}),
}
}
CalStatement::DefineQuery(def) => {
store
.define_query(
&def.name,
&def.body,
def.description.as_deref(),
&def.params,
)
.map_err(|e| {
let detail = e.to_string();
if detail.contains("AUT-E") {
CalError::NotAuthorized {
detail,
span: None,
store_code: authz_store_code(&e),
}
} else {
CalError::InvalidQueryBody { detail, span: None }
}
})?;
Ok(CalResultPayload::QueryDefined {
name: def.name.clone(),
})
}
CalStatement::DropQuery(drop) => {
if !self.config.allow_destructive_ops {
return Ok(CalResultPayload::Unsupported {
statement: "drop_query".into(),
message: "Destructive operations are disabled for this session \
(started with --no-destructive-ops)."
.into(),
});
}
match store.drop_query(&drop.name) {
Ok(()) => Ok(CalResultPayload::QueryDropped {
name: drop.name.clone(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "drop_query".into(),
message: format!("DROP QUERY failed: {e}"),
}),
}
}
CalStatement::RunQuery(run) => self.execute_run_query(run, store, query, exec_warnings),
CalStatement::Purge(purge) => {
if !self.config.allow_destructive_ops {
return Ok(CalResultPayload::Unsupported {
statement: "purge".into(),
message: "Destructive operations are disabled for this session \
(started with --no-destructive-ops)."
.into(),
});
}
let because = match purge.reason.as_deref() {
Some(r) if !r.trim().is_empty() => r,
_ => {
return Err(CalError::MissingReason { span: purge.span });
}
};
let min_age = purge.min_age_days.unwrap_or(30.0);
let batch_limit = purge.limit.unwrap_or(1000);
let ns = purge.namespace.as_deref();
match store.cal_purge_stale(
min_age,
ns,
batch_limit,
purge.grain_type.as_deref(),
because,
) {
Ok(count) => Ok(CalResultPayload::Purged { count }),
Err(e) => Err(map_store_err(e, purge.span)),
}
}
CalStatement::ReportSubject(rs) => {
match store.cal_subject_report(&rs.subject_id, rs.text_mentions) {
Ok(report) => Ok(CalResultPayload::SubjectReport {
subject: rs.subject_id.clone(),
identity_names: report.identity_names,
grains: report.grains,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "report_subject".into(),
message: format!("REPORT SUBJECT failed: {e}"),
}),
}
}
CalStatement::EntityAt(ea) => {
let axis = ea.axis.clone().unwrap_or_else(|| "world".to_string());
if ea.namespaces.is_empty() {
return match store.cal_entity_at(&ea.subject, &ea.relation, ea.at_ms, &axis) {
Ok(grain) => Ok(CalResultPayload::EntityAt {
grain,
axis,
at_ms: ea.at_ms,
grains: Vec::new(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "entity_at".into(),
message: format!("ENTITY AT failed: {e}"),
}),
};
}
match store.cal_entity_at_scoped(
&ea.namespaces,
&ea.subject,
&ea.relation,
ea.at_ms,
&axis,
) {
Ok(rows) => Ok(CalResultPayload::EntityAt {
grain: None,
axis,
at_ms: ea.at_ms,
grains: rows
.into_iter()
.map(|(ns, grain)| serde_json::json!({"namespace": ns, "grain": grain}))
.collect(),
}),
Err(e) => {
let detail = e.to_string();
if detail.contains("AUT-E") {
return Err(CalError::NotAuthorized {
detail,
span: ea.span,
store_code: authz_store_code(&e),
});
}
Ok(CalResultPayload::Unsupported {
statement: "entity_at".into(),
message: format!("ENTITY AT failed: {e}"),
})
}
}
}
CalStatement::RunTrace(rt) => {
match store.cal_run_trace(&rt.run_id, rt.limit.unwrap_or(64).min(1024)) {
Ok(trace) => Ok(CalResultPayload::RunTrace {
run_id: rt.run_id.clone(),
trace,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "run_trace".into(),
message: format!("RUN TRACE failed: {e}"),
}),
}
}
CalStatement::RunsTouching(rt) => {
let h = Hash::from_hex(&rt.hash).map_err(|_| CalError::InvalidHash {
found: rt.hash.clone(),
span: rt.span,
})?;
match store.cal_runs_touching(&h, rt.depth.unwrap_or(4).min(8)) {
Ok(runs) => Ok(CalResultPayload::RunsTouching { hash: rt.hash.clone(), runs }),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "runs_touching".into(),
message: format!("RUNS TOUCHING failed: {e}"),
}),
}
}
CalStatement::DerivedFrom(df) => {
let h = Hash::from_hex(&df.hash).map_err(|_| CalError::InvalidHash {
found: df.hash.clone(),
span: df.span,
})?;
match store.cal_derived_from(&h) {
Ok(grains) => Ok(CalResultPayload::DerivedFrom { hash: df.hash.clone(), grains }),
Err(e) => {
let detail = e.to_string();
if detail.contains("AUT-E") {
return Err(CalError::NotAuthorized {
detail,
span: df.span,
store_code: authz_store_code(&e),
});
}
Ok(CalResultPayload::Unsupported {
statement: "derived_from".into(),
message: format!("DERIVED FROM failed: {e}"),
})
}
}
}
CalStatement::Merge(mg) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "MERGE".into(),
span: mg.span,
});
}
match store.cal_merge(
&mg.subject,
&mg.relation,
&mg.object,
mg.confidence.unwrap_or(0.9),
&mg.reason,
) {
Ok(hash) => Ok(CalResultPayload::Merged {
hash: hash.to_hex(),
subject: mg.subject.clone(),
relation: mg.relation.clone(),
object: mg.object.clone(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "merge".into(),
message: format!("MERGE failed: {e}"),
}),
}
}
CalStatement::Related(rel) => {
let relations: Vec<&str> = rel
.relations
.split(',')
.map(str::trim)
.filter(|r| !r.is_empty())
.collect();
match store.cal_related_scoped(
&rel.namespaces,
&rel.start,
&relations,
rel.direction.as_deref().unwrap_or("out"),
rel.depth.unwrap_or(2),
rel.limit.unwrap_or(64),
) {
Ok(entities) => Ok(CalResultPayload::RelatedEntities {
start: rel.start.clone(),
entities,
}),
Err(e) => {
let detail = e.to_string();
if detail.contains("AUT-E") {
return Err(CalError::NotAuthorized {
detail,
span: rel.span,
store_code: authz_store_code(&e),
});
}
Ok(CalResultPayload::Unsupported {
statement: "related".into(),
message: format!("RELATED failed: {e}"),
})
}
}
}
CalStatement::Novelty(nv) => {
match store.cal_novelty(
&nv.text,
nv.subject.as_deref(),
nv.relation.as_deref(),
nv.limit.unwrap_or(5),
) {
Ok(rows) => Ok(CalResultPayload::NoveltyMatches {
matches: rows
.into_iter()
.map(|(hash, similarity)| {
serde_json::json!({ "hash": hash, "similarity": similarity })
})
.collect(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "novelty".into(),
message: format!("NOVELTY failed: {e}"),
}),
}
}
CalStatement::ShowForks(_) => match store.open_forks() {
Ok(groups) => Ok(CalResultPayload::Forks {
forks: groups
.iter()
.map(|f| {
serde_json::json!({
"namespace": f.namespace,
"subject": f.subject,
"relation": f.relation,
"heads": f.heads,
})
})
.collect(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "show_forks".into(),
message: format!("SHOW FORKS failed: {e}"),
}),
},
CalStatement::Remember(rem) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "REMEMBER".into(),
span: rem.span,
});
}
match store.cal_remember(
&rem.content,
rem.session_id.as_deref(),
rem.role.as_deref(),
rem.run_id.as_deref(),
) {
Ok(hash) => Ok(CalResultPayload::Remembered { hash: hash.to_hex() }),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "remember".into(),
message: format!("REMEMBER failed: {e}"),
}),
}
}
CalStatement::Grant(grant) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "GRANT".into(),
span: grant.span,
});
}
match store.cal_grant(
&grant.principal,
&grant.verbs,
&grant.namespaces,
grant.reason.as_deref(),
) {
Ok(hash) => {
let object = areev_core::authz::Grant {
verbs: grant
.verbs
.iter()
.filter_map(|v| areev_core::authz::Verb::parse(v).ok())
.collect(),
namespaces: grant.namespaces.clone(),
}
.to_object_string();
Ok(CalResultPayload::Granted {
principal: grant.principal.clone(),
object,
hash: hash.to_hex(),
})
}
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "grant".into(),
message: format!("GRANT failed: {e}"),
}),
}
}
CalStatement::Revoke(revoke) => {
if !self.config.tier1_enabled {
return Err(CalError::Tier1NotEnabled {
statement: "REVOKE".into(),
span: revoke.span,
});
}
match store.cal_revoke(
&revoke.principal,
&revoke.verbs,
&revoke.namespaces,
revoke.reason.as_deref(),
) {
Ok(grants_touched) => Ok(CalResultPayload::Revoked {
principal: revoke.principal.clone(),
grants_touched,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "revoke".into(),
message: format!("REVOKE failed: {e}"),
}),
}
}
CalStatement::Approve(g) | CalStatement::Reject(g) => {
let decision = if matches!(stmt, CalStatement::Approve(_)) {
crate::governance::ReviewDecision::Approve
} else {
crate::governance::ReviewDecision::Reject
};
let name = if decision == crate::governance::ReviewDecision::Approve {
"approve"
} else {
"reject"
};
let Some(host) = &self.governance else {
return Ok(governance_unwired(name));
};
match host.review(store, &g.hash, decision, &g.reason) {
Ok(()) => Ok(CalResultPayload::Reviewed {
hash: g.hash.clone(),
decision: name.into(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: name.into(),
message: format!("{} failed: {e}", name.to_uppercase()),
}),
}
}
CalStatement::ApplyRec(g) => {
let Some(host) = &self.governance else {
return Ok(governance_unwired("apply"));
};
match host.apply(
store,
&g.hash,
&g.reason,
self.config.allow_destructive_ops,
) {
Ok(rollbackable) => Ok(CalResultPayload::RecApplied {
hash: g.hash.clone(),
rollbackable,
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "apply".into(),
message: format!("APPLY failed: {e}"),
}),
}
}
CalStatement::RollbackRec(g) => {
let Some(host) = &self.governance else {
return Ok(governance_unwired("rollback"));
};
match host.rollback(store, &g.hash, &g.reason) {
Ok(()) => Ok(CalResultPayload::RecRolledBack { hash: g.hash.clone() }),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "rollback".into(),
message: format!("ROLLBACK failed: {e}"),
}),
}
}
CalStatement::RunLoop(run) => {
let Some(host) = &self.governance else {
return Ok(governance_unwired("run_loop"));
};
let opts = crate::governance::RunLoopOptions {
full_sweep: run.full_sweep,
min_new: run.min_new,
if_stale_ms: run.if_stale_ms,
};
match host.run_loop(store, &opts) {
Ok(report) => Ok(CalResultPayload::LoopRan { run: report }),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "run_loop".into(),
message: format!("RUN LOOP failed: {e}"),
}),
}
}
CalStatement::ShowGrants(show) => {
match store.cal_show_grants(show.principal.as_deref()) {
Ok(rows) => Ok(CalResultPayload::GrantList {
grants: rows
.into_iter()
.map(|row| {
let parsed =
areev_core::authz::Grant::from_object_string(&row.object)
.ok();
serde_json::json!({
"principal": row.principal,
"object": row.object,
"verbs": parsed.as_ref().map(|g| g
.verbs
.iter()
.map(|v| v.as_str())
.collect::<Vec<_>>()),
"namespaces": parsed.as_ref().map(|g| g.namespaces.clone()),
"hash": row.hash,
})
})
.collect(),
}),
Err(e) => Ok(CalResultPayload::Unsupported {
statement: "show_grants".into(),
message: format!("SHOW GRANTS failed: {e}"),
}),
}
}
}
}
fn execute_run_query(
&self,
run: &super::ast::RunQueryStmt,
store: &dyn CalStoreFacade,
outer_query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let entry = store
.get_query(&run.name)
.ok_or_else(|| CalError::QueryNotFound {
name: run.name.clone(),
span: run.span,
})?;
let mut body = entry.body.clone();
let mut param_values: HashMap<String, String> = HashMap::new();
for p in &entry.params {
if let Some(ref default) = p.default {
param_values.insert(p.name.clone(), value_to_cal_literal(default));
}
}
for (name, value) in &run.bindings {
param_values.insert(name.clone(), value_to_cal_literal(value));
}
for p in &entry.params {
if p.default.is_none() && !param_values.contains_key(&p.name) {
return Err(CalError::MissingQueryParam {
name: p.name.clone(),
query: run.name.clone(),
span: run.span,
});
}
}
let declared_names: std::collections::HashSet<&str> =
entry.params.iter().map(|p| p.name.as_str()).collect();
for (name, _) in &run.bindings {
if !declared_names.contains(name.as_str()) {
exec_warnings.push(format!(
"CAL-W006: Parameter \"${}\" supplied but not used in query \"{}\"",
name, run.name
));
}
}
for (name, literal) in ¶m_values {
body = body.replace(&format!("${}", name), literal);
}
let parsed = self.parse_cached(&body).map_err(|e| CalError::InvalidQueryBody {
detail: e.to_string(),
span: run.span,
})?;
let span = run.span.unwrap_or_else(Span::zero);
super::parser::check_read_only_statement(&parsed.statement, &span)?;
let mut merged_query = parsed;
for opt in &outer_query.with_options {
let existing_idx = merged_query
.with_options
.iter()
.position(|o| std::mem::discriminant(o) == std::mem::discriminant(opt));
if let Some(idx) = existing_idx {
merged_query.with_options[idx] = opt.clone();
} else {
merged_query.with_options.push(opt.clone());
}
}
if outer_query.format.is_some() {
merged_query.format = outer_query.format.clone();
}
for stage in &outer_query.pipeline {
merged_query.pipeline.push(stage.clone());
}
for (k, v) in &outer_query.user_vars {
merged_query
.user_vars
.entry(k.clone())
.or_insert_with(|| v.clone());
}
let result =
self.execute_statement(&merged_query.statement, store, &merged_query, exec_warnings)?;
let _ = store.update_query_last_run(&run.name);
Ok(result)
}
fn execute_recall(
&self,
recall: &RecallStmt,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let mut params = RecallParams::default();
if let Some(gt) = recall.grain_type.to_grain_type() {
params.grain_type = Some(gt);
}
if let Some(ref about) = recall.about {
params.query = Some(about.text.clone());
}
if params.query.is_none() {
if let Some(ref like) = recall.like {
params.query = Some(like.text.clone());
}
}
if let Some(ref where_clause) = recall.where_clause {
self.apply_where_clause(&where_clause.condition, &mut params, exec_warnings, &query.let_values)?;
}
if recall.grain_type == GrainTypePlural::Consents {
let expand_did = |value: &str| -> Vec<String> {
let mut variants = vec![value.to_string()];
if !value.starts_with("did:") {
variants.push(format!("did:{}", value));
}
variants
};
if let Some(ref subj) = params.subject.take() {
let expanded = expand_did(subj);
match params.subject_in.as_mut() {
Some(existing) => existing.extend(expanded),
None => params.subject_in = Some(expanded),
}
}
if let Some(ref existing) = params.subject_in.clone() {
let mut seen = std::collections::HashSet::new();
let deduped: Vec<String> = existing
.iter()
.flat_map(|s| expand_did(s))
.filter(|s| seen.insert(s.clone()))
.collect();
params.subject_in = Some(deduped);
}
}
match (&recall.since, &recall.until) {
(Some(since), Some(until)) => {
params.temporal_expr = Some(format!(
"between {} and {}",
since.expression, until.expression
));
}
(Some(since), None) => {
params.temporal_expr = Some(since.expression.clone());
}
(None, Some(until)) => {
params.temporal_expr = Some(format!("before {}", until.expression));
}
(None, None) => {}
}
if let Some(ref between) = recall.between {
params.temporal_expr = Some(format!("between {} and {}", between.start, between.end));
}
if let Some(ref recent) = recall.recent {
params.limit = Some(recent.count.min(self.config.max_limit) as usize);
}
if let Some(limit) = recall.limit {
params.limit = Some(limit.min(self.config.max_limit) as usize);
}
if params.limit.is_none() {
params.limit = Some(self.config.default_limit as usize);
}
let contradictions_limit = if recall.contradictions.is_some() {
params.limit.replace(self.config.max_limit as usize)
} else {
None
};
let order_by: Option<(String, bool)> = query.pipeline.iter().find_map(|st| match st {
PipelineStage::OrderBy {
field, descending, ..
} => Some((field.clone(), *descending)),
_ => None,
});
let has_count = query
.pipeline
.iter()
.any(|st| matches!(st, PipelineStage::Count { .. }));
let residual_where = match recall.where_clause.as_ref() {
Some(w) => plan_residual_where(&w.condition, &recall.grain_type, exec_warnings)?,
None => None,
};
let has_post_filter = residual_where.is_some();
let wide_reason: Option<String> = if recall.contradictions.is_some() {
None
} else if let Some((ref f, _)) = order_by {
Some(format!("ORDER BY {f}"))
} else if has_count {
Some("COUNT".to_string())
} else if has_post_filter {
Some("a post-retrieval WHERE filter".to_string())
} else {
None
};
let pushed_down_sort = matches!(order_by, Some((ref f, _)) if f == "created_at");
if let (true, Some((ref field, descending))) = (pushed_down_sort, &order_by) {
params.order_by = Some(crate::store_types::SortKey {
field: field.clone(),
descending: *descending,
});
}
let widened_limit = if wide_reason.is_some() && !pushed_down_sort {
params.limit.replace(self.config.max_limit as usize)
} else {
None
};
self.apply_with_options(&query.with_options, &mut params)?;
if params.exhaustive.is_some() && recall.about.is_none() {
return Err(CalError::UnexpectedToken {
expected: "ABOUT clause (WITH exhaustive requires a semantic search query)".into(),
found: "no ABOUT clause".into(),
span: recall.span,
suggestion: Some(
"Add an ABOUT clause: RECALL facts ABOUT \"...\" WITH exhaustive".into(),
),
});
}
if let Some(ref ns) = self.config.namespace_override {
params.namespace = Some(ns.clone());
params.namespaces = None;
} else if params.namespace.is_none() && params.namespaces.is_none() {
if let Some(ns) = store.default_namespace() {
params.namespace = Some(ns.to_string());
}
}
if let Some(ref uid) = self.config.user_id_override {
params.user_id = Some(uid.clone());
}
let hits = store
.recall(¶ms)
.map_err(|e| map_store_err(e, recall.span))?;
let scan_was_bounded = params.limit.is_some_and(|l| hits.len() >= l);
let mut grains = hits_to_grain_results(&hits);
if recall.about.is_none() {
for g in &mut grains {
g.is_deterministic = true;
}
}
if let Some(ref residual) = residual_where {
grains.retain(|grain| grain_matches_condition_tree(grain, residual));
}
if let Some(reason) = wide_reason {
if scan_was_bounded {
exec_warnings.push(
super::errors::CalWarning::ScanBounded {
stage: reason,
scanned: self.config.max_limit as usize,
}
.to_string(),
);
}
if let Some((ref field, descending)) = order_by {
grains.sort_by(|a, b| {
let cmp = compare_json_values(
json_field(&a.fields, field),
json_field(&b.fields, field),
);
if descending {
cmp.reverse()
} else {
cmp
}
});
}
let pipeline_bounds = query.pipeline.iter().any(|st| {
matches!(
st,
PipelineStage::Limit { .. }
| PipelineStage::First { .. }
| PipelineStage::Count { .. }
)
});
if !pipeline_bounds {
if let Some(limit) = widened_limit {
grains.truncate(limit);
}
}
}
for opt in &query.with_options {
if let WithOption::Dedup { field: Some(field) } = opt {
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
grains.retain(|g| match json_field(&g.fields, field) {
Some(v) => seen.insert(value_dedup_key(v)),
None => true,
});
}
}
if params.score_breakdown == Some(true) {
exec_warnings.push(
super::errors::CalWarning::WithOptionInert {
option: "score_breakdown",
statement: "RECALL",
why: "this recall path returns fused grains, not per-leg scores, so \
structural hits all carry the sentinel score 1.0",
}
.to_string(),
);
}
if let Some(ref clause) = recall.contradictions {
let scope = match clause.inner {
Some(ref inner) => {
let inner_query = CalQuery {
statement: (**inner).clone(),
pipeline: Vec::new(),
format: None,
..query.clone()
};
let payload = self.execute_statement(
&inner_query.statement,
store,
&inner_query,
exec_warnings,
)?;
match payload {
CalResultPayload::Grains { ref grains, .. } => Some(
contradiction_scope_keys(grains, params.namespace.as_deref()),
),
_ => {
exec_warnings.push(
"CONTRADICTIONS OF (...) sub-query returned no grains; \
no keys to scope by"
.into(),
);
Some(std::collections::HashSet::new())
}
}
}
None => None,
};
apply_fork_status(&mut grains, store, true, scope.as_ref(), exec_warnings);
if scan_was_bounded {
exec_warnings.push(
super::errors::CalWarning::ContradictionScanBounded {
scanned: self.config.max_limit as usize,
}
.to_string(),
);
}
if let Some(limit) = contradictions_limit {
grains.truncate(limit);
}
} else if params.detect_contradictions == Some(true) {
apply_fork_status(&mut grains, store, false, None, exec_warnings);
}
let count = grains.len();
if query.pipeline.is_empty() {
if let Some(ref fmt) = query.format {
return apply_format_clause_to_grains(
&grains,
fmt,
None,
RenderInputs {
user_vars: &query.user_vars,
store,
disclosure: disclosure_of(&query.with_options),
},
&Default::default(),
exec_warnings,
);
}
}
Ok(CalResultPayload::Grains {
grains,
total_available: Some(count),
})
}
fn apply_where_clause(
&self,
condition: &Condition,
params: &mut RecallParams,
warnings: &mut Vec<String>,
let_values: &HashMap<String, Vec<String>>,
) -> std::result::Result<(), CalError> {
match condition {
Condition::Comparison {
field,
comparator,
value,
..
} => {
match (field.as_str(), comparator) {
("subject", Comparator::Eq) => {
params.subject = Some(value_to_string(value)?);
}
("relation", Comparator::Eq) => {
params.relation = Some(value_to_string(value)?);
}
("object", Comparator::Eq) => {
params.object = Some(value_to_string(value)?);
}
("namespace", Comparator::Eq) if self.config.namespace_override.is_none() => {
params.namespace = Some(value_to_string(value)?);
}
("user_id", Comparator::Eq) if self.config.user_id_override.is_none() => {
params.user_id = Some(value_to_string(value)?);
}
("confidence", Comparator::Gte) | ("confidence", Comparator::Gt) => {
params.confidence_threshold = Some(value_to_f64(value)?);
}
("importance", Comparator::Gte) | ("importance", Comparator::Gt) => {
params.importance_threshold = Some(value_to_f64(value)?);
}
("query", Comparator::Eq) => {
params.query = Some(value_to_string(value)?);
}
("session_id", Comparator::Eq) => {
params.session_id = Some(value_to_string(value)?);
}
("time", Comparator::Eq) => {
params.temporal_expr = Some(value_to_string(value)?);
}
("contradicted", Comparator::Eq) => {
if let Value::Boolean { value: b } = value {
params.include_contradicted = Some(*b);
}
}
("entity", Comparator::Eq) => {
params.entity = Some(value_to_string(value)?);
}
("hash", Comparator::Eq) | ("hash", Comparator::NotEq) => {}
("scope_path", Comparator::Eq) | ("scope", Comparator::Eq) => {
params.scope_path = Some(value_to_string(value)?);
}
_ => {}
}
Ok(())
}
Condition::And { left, right, .. } => {
self.apply_where_clause(left, params, warnings, let_values)?;
self.apply_where_clause(right, params, warnings, let_values)?;
Ok(())
}
Condition::In { field, values, span } => {
let mut str_values: Vec<String> = Vec::new();
for v in values {
match v {
Value::String { value } | Value::Hash { value } => {
str_values.push(value.clone())
}
Value::Number { value } => str_values.push(value.to_string()),
Value::Parameter { name } => match let_values.get(name.as_str()) {
Some(bound) => str_values.extend(bound.iter().cloned()),
None => {
return Err(CalError::UnboundParameter {
name: name.clone(),
span: *span,
})
}
},
Value::Array { values: inner } => {
for iv in inner {
if let Value::String { value } = iv {
str_values.push(value.clone());
}
}
}
Value::Boolean { value } => str_values.push(value.to_string()),
}
}
match field.as_str() {
"subject" => params.subject_in = Some(str_values),
"relation" => params.relation_in = Some(str_values),
"object" => params.object_in = Some(str_values),
"tags" => params.tags = Some(str_values),
"namespace" => {
params.namespaces = Some(str_values);
}
_ => {
}
}
Ok(())
}
Condition::NotIn { field, values, .. } => {
if field == "tags" {
let tag_strs: Vec<String> = values
.iter()
.filter_map(|v| match v {
Value::String { value } => Some(value.clone()),
_ => None,
})
.collect();
params.exclude_tags = Some(tag_strs);
}
Ok(())
}
Condition::Contains { field, value, .. } => {
match field.as_str() {
"subject" => {
params.subject_contains = Some(value.clone());
}
"object" => {
params.object_contains = Some(value.clone());
}
"content" | "summary" if params.query.is_none() => {
params.query = Some(value.clone());
}
_ => {}
}
Ok(())
}
Condition::Or { .. } => {
Ok(())
}
Condition::IsCategory {
field, category, ..
} => {
if field == "relation" {
let relations = super::relations::expand_category(category);
if relations.is_empty() {
warnings.push(format!(
"Unknown relation category '{}'; no relations expanded.",
category
));
} else {
params.relation_in =
Some(relations.iter().map(|r| r.to_string()).collect());
}
} else {
warnings.push(format!(
"CAL-W008: IS {} used on field '{}' — IS CATEGORY is only meaningful on the 'relation' field; this condition was ignored.",
category, field
));
}
Ok(())
}
_ => Ok(()),
}
}
fn apply_with_options(
&self,
options: &[WithOption],
params: &mut RecallParams,
) -> std::result::Result<(), CalError> {
for opt in options {
match opt {
WithOption::Superseded => {
params.exclude_superseded = Some(false);
}
WithOption::ScoreBreakdown => {
params.score_breakdown = Some(true);
}
WithOption::Explanation => {
params.explanation = Some(true);
}
WithOption::ContradictionDetection => {
params.detect_contradictions = Some(true);
}
WithOption::Diversity { lambda } => {
params.diversity = Some(if let Some(l) = lambda {
DiversityConfig::mmr_with_lambda(*l as f32)
} else {
DiversityConfig::mmr()
});
}
WithOption::Provenance => {
params.record_provenance = Some(true);
}
WithOption::Dedup { field: _ } => {
params.deduplicate = Some(true);
}
WithOption::Rerank { ref model } => {
let mut cfg = crate::store_types::RerankConfig::default();
if let Some(m) = model {
cfg.model = Some(m.clone());
}
params.rerank = Some(cfg);
}
WithOption::LlmRerank { .. } => {
return Err(CalError::LlmFeatureUnavailable { feature: "llm_rerank".into() });
}
WithOption::Hyde => {
return Err(CalError::LlmFeatureUnavailable { feature: "hyde".into() });
}
WithOption::QueryExpansion => {
params.query_expansion = Some(true);
}
WithOption::QueryDecompose => {
params.query_decompose = Some(true);
}
WithOption::ConflictResolution => {
params.conflict_resolution = Some(true);
}
WithOption::IncludeSources => {
params.include_sources = Some(true);
}
WithOption::AnnotateRelativeTime => {
params.annotate_relative_time = Some(true);
}
WithOption::RecencyWeight { weight } => {
params.recency_weight = Some(*weight);
}
WithOption::MinScore { score } => {
params.min_score = Some(*score);
}
WithOption::MultiHop { hops } => {
params.multi_hop = Some((*hops as u8).clamp(1, 3));
}
WithOption::SessionAffinity { boost } => {
params.session_affinity_boost = Some(boost.clamp(0.0, 1.0));
}
WithOption::SubjectAffinity { boost } => {
params.subject_affinity_boost = Some(boost.clamp(0.0, 1.0));
}
WithOption::SessionCoverage { min_per_ns } => {
params.min_per_namespace = Some((*min_per_ns as usize).clamp(1, 10));
}
WithOption::MaxNamespaces { max } => {
params.max_namespaces = Some((*max as usize).clamp(1, 100));
}
WithOption::Exhaustive { max_rounds } => {
let mut config = crate::store_types::ExhaustiveConfig::default();
if let Some(rounds) = max_rounds {
config.max_rounds = (*rounds as u8).clamp(1, 5);
}
config.validate();
params.exhaustive = Some(config);
}
WithOption::SessionCensus {
min_per_session,
min_score,
} => {
let mut config = crate::store_types::SessionCensusConfig::default();
if let Some(mps) = min_per_session {
config.min_per_session = (*mps as u8).clamp(1, 10);
}
if let Some(ms) = min_score {
config.min_score = ms.clamp(0.0, 1.0);
}
config.validate();
params.session_census = Some(config);
}
WithOption::AggregationIntent => {
params.aggregation_intent = Some(true);
}
WithOption::PreferenceEnrichment => {
params.preference_enrichment = Some(true);
}
WithOption::ProgressiveDisclosure { .. } => {}
WithOption::Consistency { .. } => {}
WithOption::Locale { .. } => {}
WithOption::Cache { .. } => {}
}
}
Ok(())
}
#[allow(unused_variables)] #[allow(clippy::ptr_arg)] fn apply_assemble_post_merge_options(
&self,
payload: CalResultPayload,
options: &[WithOption],
warnings: &mut Vec<String>,
store: &dyn CalStoreFacade,
about_text: &str,
) -> std::result::Result<CalResultPayload, CalError> {
let (mut grains, sources, total_tokens, budget_limit, _total_available) = match payload {
CalResultPayload::Assembled {
grains,
sources,
total_tokens,
budget_limit,
total_available,
..
} => (grains, sources, total_tokens, budget_limit, total_available),
other => return Ok(other),
};
for opt in options {
match opt {
WithOption::ConflictResolution => {
#[allow(clippy::type_complexity)]
let mut groups: HashMap<
(String, String),
Vec<(i64, usize, String)>,
> = HashMap::new();
for (idx, grain) in grains.iter().enumerate() {
let subj = grain
.fields
.get("subject")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let rel = grain
.fields
.get("relation")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if subj.is_empty() && rel.is_empty() {
continue;
}
let obj = grain
.fields
.get("object")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let created = grain
.fields
.get("created_at")
.and_then(|v| v.as_i64())
.unwrap_or(0);
groups
.entry((subj, rel))
.or_default()
.push((created, idx, obj));
}
let mut remove_indices: std::collections::HashSet<usize> =
std::collections::HashSet::new();
for members in groups.values() {
if members.len() < 2 {
continue;
}
let has_diff = members.windows(2).any(|w| w[0].2 != w[1].2);
if !has_diff {
continue;
}
let best_idx = members
.iter()
.max_by_key(|(ts, _, _)| *ts)
.map(|(_, idx, _)| *idx)
.unwrap();
for (_, idx, _) in members {
if *idx != best_idx {
remove_indices.insert(*idx);
}
}
}
if !remove_indices.is_empty() {
let mut idx = 0;
grains.retain(|_| {
let keep = !remove_indices.contains(&idx);
idx += 1;
keep
});
}
}
WithOption::Dedup { field: Some(field) } => {
let mut seen: std::collections::HashSet<String> =
std::collections::HashSet::new();
grains.retain(|grain| match json_field(&grain.fields, field) {
Some(v) => seen.insert(value_dedup_key(v)),
None => true,
});
}
WithOption::Dedup { field: None } => {
let threshold = 0.85_f64;
let mut seen_texts: Vec<String> = Vec::new();
grains.retain(|grain| {
let text = grain_result_text(grain);
for prev in &seen_texts {
if text_similarity(&text, prev) >= threshold {
return false;
}
}
seen_texts.push(text);
true
});
}
WithOption::MinScore { score } => {
grains.retain(|g| g.is_deterministic || g.score >= *score);
}
WithOption::Diversity { .. } => {
grains.sort_by(|a, b| b.score.partial_cmp(&a.score).unwrap_or(Ordering::Equal));
}
#[cfg(feature = "rerank")]
WithOption::Rerank { ref model } => {
if about_text.is_empty() {
warnings.push(
"WITH rerank on ASSEMBLE requires a topic — skipping \
post-merge reranking"
.into(),
);
} else if !grains.is_empty() {
let texts: Vec<String> = grains.iter().map(grain_result_text).collect();
let refs: Vec<&str> = texts.iter().map(|s| s.as_str()).collect();
let user_id = self.config.user_id_override.as_deref();
let perm = store
.rerank_passages(
about_text,
&refs,
super::facade::RerankType::CrossEncoder,
model.as_deref(),
user_id,
)
.map_err(|e| CalError::BudgetExceeded {
detail: format!("post-merge reranking failed: {}", e),
span: None,
})?;
let mut reranked = Vec::with_capacity(grains.len());
for &i in &perm {
if let Some(g) = grains.get(i) {
reranked.push(g.clone());
}
}
grains = reranked;
}
}
#[cfg(feature = "llm-rerank")]
WithOption::LlmRerank { ref model } => {
if about_text.is_empty() {
warnings.push(
"WITH llm_rerank on ASSEMBLE requires a topic — skipping \
post-merge reranking"
.into(),
);
} else if !grains.is_empty() {
let texts: Vec<String> = grains.iter().map(grain_result_text).collect();
let refs: Vec<&str> = texts.iter().map(|s| s.as_str()).collect();
let user_id = self.config.user_id_override.as_deref();
let perm = store
.rerank_passages(
about_text,
&refs,
super::facade::RerankType::Llm,
model.as_deref(),
user_id,
)
.map_err(|e| CalError::BudgetExceeded {
detail: format!("post-merge LLM reranking failed: {}", e),
span: None,
})?;
let mut reranked = Vec::with_capacity(grains.len());
for &i in &perm {
if let Some(g) = grains.get(i) {
reranked.push(g.clone());
}
}
grains = reranked;
}
}
_ => {}
}
}
let count = grains.len();
Ok(CalResultPayload::Assembled {
grains,
sources,
total_tokens,
budget_limit,
progressive: false,
total_available: Some(count),
})
}
fn execute_exists(
&self,
exists: &ExistsStmt,
store: &dyn CalStoreFacade,
exec_warnings: &mut Vec<String>,
let_values: &HashMap<String, Vec<String>>,
) -> std::result::Result<CalResultPayload, CalError> {
let mut params = RecallParams::default();
if let Some(gt) = exists.grain_type.to_grain_type() {
params.grain_type = Some(gt);
}
if let Some(ref about) = exists.about {
params.query = Some(about.text.clone());
}
if let Some(ref where_clause) = exists.where_clause {
self.apply_where_clause(&where_clause.condition, &mut params, exec_warnings, let_values)?;
}
let hash_str =
extract_hash_from_condition(exists.where_clause.as_ref().map(|w| &w.condition));
if let Some(ref hs) = hash_str {
let hash = Hash::from_hex(hs).map_err(|_| CalError::InvalidHash {
found: hs.clone(),
span: exists.span,
})?;
let found = store
.exists(&hash)
.map_err(|e| map_store_err(e, exists.span))?;
return Ok(CalResultPayload::Exists {
exists: found,
hash: hs.clone(),
});
}
let residual_where = match exists.where_clause.as_ref() {
Some(w) => plan_residual_where(&w.condition, &exists.grain_type, exec_warnings)?,
None => None,
};
params.limit = if residual_where.is_some() {
Some(self.config.max_limit as usize)
} else {
Some(1)
};
if let Some(ref ns) = self.config.namespace_override {
params.namespace = Some(ns.clone());
params.namespaces = None;
}
if let Some(ref uid) = self.config.user_id_override {
params.user_id = Some(uid.clone());
}
let hits = store
.recall(¶ms)
.map_err(|e| map_store_err(e, exists.span))?;
let (found, hash_out) = if let Some(ref residual) = residual_where {
let grains = hits_to_grain_results(&hits);
let first = grains
.iter()
.find(|g| grain_matches_condition_tree(g, residual));
(first.is_some(), first.map(|g| g.hash.clone()).unwrap_or_default())
} else {
(
!hits.is_empty(),
hits.first().map(|h| h.hash.to_hex()).unwrap_or_default(),
)
};
Ok(CalResultPayload::Exists {
exists: found,
hash: hash_out,
})
}
fn execute_history(
&self,
history: &HistoryStmt,
store: &dyn CalStoreFacade,
exec_warnings: &mut Vec<String>,
let_values: &HashMap<String, Vec<String>>,
) -> std::result::Result<CalResultPayload, CalError> {
if let Some(ref diff_hash_str) = history.diff_target {
let source_hash = Hash::from_hex(&history.hash).map_err(|_| CalError::InvalidHash {
found: history.hash.clone(),
span: history.span,
})?;
let target_hash = Hash::from_hex(diff_hash_str).map_err(|_| CalError::InvalidHash {
found: diff_hash_str.clone(),
span: history.span,
})?;
let grain_a = store.get(&source_hash).map_err(|e| match e {
AreevError::NotFound(_) => CalError::HashNotFound {
hash: history.hash.clone(),
span: history.span,
},
AreevError::CryptoError(_) => CalError::CryptoError {
detail: format!("DIFF source grain decrypt failed: {}", e),
span: history.span,
store_code: Some(e.code()),
},
other => map_store_err(other, history.span),
})?;
let grain_b = store.get(&target_hash).map_err(|e| match e {
AreevError::NotFound(_) => CalError::HashNotFound {
hash: diff_hash_str.clone(),
span: history.span,
},
AreevError::CryptoError(_) => CalError::CryptoError {
detail: format!("DIFF target grain decrypt failed: {}", e),
span: history.span,
store_code: Some(e.code()),
},
other => map_store_err(other, history.span),
})?;
let sub_a = grain_a.get_str("subject").unwrap_or("");
let sub_b = grain_b.get_str("subject").unwrap_or("");
let rel_a = grain_a.get_str("relation").unwrap_or("");
let rel_b = grain_b.get_str("relation").unwrap_or("");
if sub_a != sub_b || rel_a != rel_b {
exec_warnings.push(
"CAL-W005: DIFF targets have different subject/relation — diff may not be meaningful".to_string()
);
}
let changes = diff_grains(&grain_a, &grain_b);
return Ok(CalResultPayload::Diff {
source_hash: history.hash.clone(),
target_hash: diff_hash_str.clone(),
changes,
});
}
if history.hash.is_empty() {
if let Some(ref wc) = history.where_clause {
let mut params = RecallParams::default();
self.apply_where_clause(&wc.condition, &mut params, exec_warnings, let_values)?;
let residual_where =
plan_residual_where(&wc.condition, &GrainTypePlural::All, exec_warnings)?;
if let Some(ref ns) = self.config.namespace_override {
params.namespace = Some(ns.clone());
params.namespaces = None;
}
if let Some(ref uid) = self.config.user_id_override {
params.user_id = Some(uid.clone());
}
if params.limit.is_none() {
params.limit = Some(if residual_where.is_some() {
self.config.max_limit as usize
} else {
10
});
}
let hits = store
.recall(¶ms)
.map_err(|e| map_store_err(e, history.span))?;
let hits: Vec<crate::store_types::SearchHit> =
if let Some(ref residual) = residual_where {
let grains = hits_to_grain_results(&hits);
hits.into_iter()
.zip(grains)
.filter(|(_, g)| grain_matches_condition_tree(g, residual))
.map(|(h, _)| h)
.collect()
} else {
hits
};
let mut all_versions: Vec<CalVersionResult> = Vec::new();
for hit in &hits {
let ns = hit.grain.get_str("namespace").unwrap_or("");
let subj = hit.grain.get_str("subject").unwrap_or("");
let rel = hit.grain.get_str("relation").unwrap_or("");
if let Ok(entries) = store.get_history(ns, subj, rel) {
for v in entries {
if !all_versions
.iter()
.any(|existing| existing.hash == v.hash.to_hex())
{
all_versions.push(CalVersionResult {
hash: v.hash.to_hex(),
object: v.object,
created_at: v.created_at,
confidence: v.confidence,
superseded_by: v
.superseded_by
.map(|h: areev_core::error::Hash| h.to_hex()),
});
}
}
}
}
return Ok(CalResultPayload::History {
versions: all_versions,
});
}
return Err(CalError::InvalidHash {
found: String::new(),
span: history.span,
});
}
let hash = Hash::from_hex(&history.hash).map_err(|_| CalError::InvalidHash {
found: history.hash.clone(),
span: history.span,
})?;
let grain = match store.get(&hash) {
Ok(g) => g,
Err(AreevError::NotFound(_)) => {
return Ok(CalResultPayload::Unsupported {
statement: "history".into(),
message: format!("grain not found: {}", history.hash),
});
}
Err(other) => {
return Err(map_store_err(other, history.span));
}
};
let namespace = grain.get_str("namespace").unwrap_or("");
let subject = grain.get_str("subject").unwrap_or("");
let relation = grain.get_str("relation").unwrap_or("");
let entries = store
.get_history(namespace, subject, relation)
.map_err(|e: areev_core::error::AreevError| map_store_err(e, history.span))?;
let versions: Vec<CalVersionResult> = entries
.into_iter()
.map(|v| CalVersionResult {
hash: v.hash.to_hex(),
object: v.object,
created_at: v.created_at,
confidence: v.confidence,
superseded_by: v.superseded_by.map(|h: areev_core::error::Hash| h.to_hex()),
})
.collect();
Ok(CalResultPayload::History { versions })
}
fn execute_describe(
&self,
describe: &DescribeStmt,
store: &dyn CalStoreFacade,
) -> std::result::Result<CalResultPayload, CalError> {
let info = match &describe.target {
DescribeTarget::Schema => serde_json::json!({
"grain_types": [
"fact", "event", "state", "workflow", "tool", "recommendation",
"observation", "goal", "reasoning", "consensus", "consent", "skill"
],
"common_fields": [
"subject", "relation", "object", "namespace", "user_id",
"created_at", "confidence", "importance", "tags",
"session_id", "content", "summary"
],
"cal_version": 1,
"tier1_enabled": self.config.tier1_enabled,
"max_limit": self.config.max_limit,
"default_limit": self.config.default_limit,
"oms_version": "1.2",
"pipeline_stages": [
"SELECT", "ORDER BY", "LIMIT", "OFFSET", "COUNT",
"FIRST", "SUBJECTS", "OBJECTS", "HASHES", "GROUP BY", "PROJECT"
],
"with_options": [
"superseded", "explanation", "annotate_relative_time",
"conflict_resolution", "dedup", "contradiction_detection",
"diversity", "multi_hop", "provenance", "query_expansion",
"rerank"
]
}),
DescribeTarget::GrainType(gt) => {
let type_name = gt.as_str();
let specific_fields: &[&str] = match gt {
GrainTypePlural::Triggers => &[
"kind",
"workflow",
"connector",
"scope",
"enabled",
"cron",
],
GrainTypePlural::Facts => {
&["subject", "relation", "object", "confidence", "session_id"]
}
GrainTypePlural::Events => &[
"session_id",
"run_id",
"content",
"created_at",
"role",
"parent_message_id",
"model_id",
"stop_reason",
],
GrainTypePlural::Recommendations => &[
"target_ref",
"analyzer",
"severity",
"dedup_key",
"rec_status",
],
GrainTypePlural::States => &["context", "plan", "history"],
GrainTypePlural::Workflows => &[
"name",
"nodes",
"edges",
"bindings",
"retries",
"trigger",
"status",
"session_id",
],
GrainTypePlural::Tools => &[
"tool",
"input",
"content",
"is_error",
"duration_ms",
"session_id",
],
GrainTypePlural::Observations => &["sensor", "value", "unit", "session_id"],
GrainTypePlural::Goals => &[
"title",
"description",
"priority",
"status",
"parent_hash",
"session_id",
],
GrainTypePlural::Reasonings => {
&["premises", "conclusion", "confidence", "session_id"]
}
GrainTypePlural::Consensuses => {
&["participants", "agreement", "confidence", "session_id"]
}
GrainTypePlural::Consents => &[
"user_id",
"scope",
"granted",
"expires_at",
"subject_did",
"grantee_did",
"session_id",
],
GrainTypePlural::Skills => &[
"name",
"description",
"version",
"domain",
"holder_did",
"proficiency",
"transferable",
"practice_count",
"last_practiced_at",
"session_id",
],
GrainTypePlural::All => &[],
};
serde_json::json!({
"grain_type": type_name,
"purpose": gt
.to_grain_type()
.map(|t| areev_core::types::registry::meta(t).purpose),
"specific_fields": specific_fields,
"required_fields": crate::json_build::required_fields(
gt.to_grain_type().map(|t| t.as_str()).unwrap_or("")
),
"common_fields": [
"namespace", "user_id", "created_at", "tags", "importance"
]
})
}
DescribeTarget::Capabilities => {
let caps = store.describe_capabilities();
serde_json::json!({
"cal_version": caps.cal_version,
"conformance_level": caps.conformance_level,
"supported_statements": caps.supported_statements,
"max_sources": caps.max_sources,
"max_let_bindings": caps.max_let_bindings,
"max_budget_tokens": caps.max_budget_tokens,
"tier1_enabled": self.config.tier1_enabled,
"template_filters": super::templates::KNOWN_FILTERS,
"template_variable_namespaces": ["grain", "assembly", "budget", "source", "group"],
"oms_version": "1.2"
})
}
DescribeTarget::Server => {
let caps = store.describe_capabilities();
serde_json::json!({
"name": "areev",
"version": env!("CARGO_PKG_VERSION"),
"cal_version": caps.cal_version,
"conformance_level": caps.conformance_level,
"oms_version": "1.2",
"build_features": build_features_list()
})
}
DescribeTarget::Fields(opt_gt) => {
let gt_engine = opt_gt.as_ref().and_then(|g| g.to_grain_type());
let field_infos = store.describe_fields(gt_engine);
if !field_infos.is_empty() {
let fields_json: Vec<serde_json::Value> = field_infos
.iter()
.map(|fi| {
serde_json::json!({
"name": fi.name,
"type": fi.field_type,
"filterable": fi.filterable,
"sortable": fi.sortable,
})
})
.collect();
if let Some(gt) = opt_gt {
serde_json::json!({
"grain_type": gt.as_str(),
"fields": fields_json
})
} else {
serde_json::json!({
"fields": fields_json
})
}
} else {
let mut fields: Vec<serde_json::Value> = vec![
serde_json::json!({"name": "subject", "type": "string", "filterable": true, "sortable": true}),
serde_json::json!({"name": "relation", "type": "string", "filterable": true, "sortable": true}),
serde_json::json!({"name": "object", "type": "string", "filterable": true, "sortable": false}),
serde_json::json!({"name": "namespace", "type": "string", "filterable": true, "sortable": true}),
serde_json::json!({"name": "user_id", "type": "string", "filterable": true, "sortable": true}),
serde_json::json!({"name": "created_at", "type": "timestamp", "filterable": true, "sortable": true}),
serde_json::json!({"name": "confidence", "type": "number", "filterable": true, "sortable": true}),
serde_json::json!({"name": "importance", "type": "number", "filterable": true, "sortable": true}),
serde_json::json!({"name": "tags", "type": "array", "filterable": true, "sortable": false}),
serde_json::json!({"name": "valid_from", "type": "timestamp", "filterable": true, "sortable": true}),
serde_json::json!({"name": "valid_to", "type": "timestamp", "filterable": true, "sortable": true}),
serde_json::json!({"name": "system_valid_from", "type": "timestamp", "filterable": true, "sortable": true}),
serde_json::json!({"name": "system_valid_to", "type": "timestamp", "filterable": true, "sortable": true}),
];
if let Some(gt) = gt_engine {
for name in areev_core::types::registry::meta(gt).queryable_fields {
if fields.iter().any(|f| f["name"] == *name) {
continue;
}
fields.push(serde_json::json!({
"name": name,
"type": "field",
"filterable": true,
"sortable": false,
}));
}
}
if let Some(gt) = opt_gt {
serde_json::json!({
"grain_type": gt.as_str(),
"fields": fields
})
} else {
serde_json::json!({
"fields": fields
})
}
}
}
DescribeTarget::Templates => {
let templates = store.list_templates();
let template_list: Vec<serde_json::Value> = templates
.iter()
.map(|t| {
serde_json::json!({
"name": t.name,
"description": t.description,
"builtin": t.builtin,
"parent": t.parent,
})
})
.collect();
serde_json::json!({ "templates": template_list })
}
DescribeTarget::Queries => {
let queries = store.list_queries();
serde_json::json!({
"queries": queries,
})
}
DescribeTarget::Query(name) => {
let entry = store
.get_query(name)
.ok_or_else(|| CalError::QueryNotFound {
name: name.clone(),
span: None,
})?;
serde_json::json!({
"name": name,
"description": entry.description,
"builtin": entry.builtin,
"params": entry.params.iter().map(|p| {
serde_json::json!({
"name": p.name,
"required": p.default.is_none(),
"default": p.default.as_ref().map(|v| match v {
super::ast::Value::String { value } => value.clone(),
super::ast::Value::Number { value } => value.to_string(),
super::ast::Value::Boolean { value } => value.to_string(),
other => format!("{}", other),
}),
})
}).collect::<Vec<_>>(),
"body": entry.body,
"body_size": entry.body.len(),
})
}
DescribeTarget::Principal(name) => {
let rows = store.cal_show_grants(Some(name)).map_err(|e| {
let detail = e.to_string();
if detail.contains("AUT-E") {
CalError::NotAuthorized {
detail,
span: None,
store_code: authz_store_code(&e),
}
} else {
CalError::InvalidQuery {
detail,
span: None,
store_code: Some(e.code()),
}
}
})?;
let grants: Vec<serde_json::Value> = rows
.iter()
.filter_map(|row| {
let g =
areev_core::authz::Grant::from_object_string(&row.object).ok()?;
Some(serde_json::json!({
"verbs": g.verbs.iter().map(|v| v.as_str()).collect::<Vec<_>>(),
"namespaces": g.namespaces,
"hash": row.hash,
}))
})
.collect();
serde_json::json!({
"principal": name,
"grants": grants,
"note": if grants.is_empty() {
"no live grants — this principal can do nothing in a bound session"
} else {
"rights shown are the live grant heads; owner sessions bypass grants"
},
})
}
DescribeTarget::Loop
| DescribeTarget::Analyzers
| DescribeTarget::Outcomes
| DescribeTarget::LoopPolicy => {
let what = match &describe.target {
DescribeTarget::Loop => crate::governance::LoopInfo::Loop,
DescribeTarget::Analyzers => crate::governance::LoopInfo::Analyzers,
DescribeTarget::Outcomes => crate::governance::LoopInfo::Outcomes,
_ => crate::governance::LoopInfo::Policy,
};
match &self.governance {
None => serde_json::json!({
"error": "governance is not wired on this surface — no \
GovernanceHost attached",
}),
Some(host) => host.describe(store, what).map_err(|e| {
let detail = e.to_string();
if detail.contains("AUT-E") {
CalError::NotAuthorized {
detail,
span: None,
store_code: authz_store_code(&e),
}
} else {
CalError::InvalidQuery {
detail,
span: None,
store_code: Some(e.code()),
}
}
})?,
}
}
DescribeTarget::Stats | DescribeTarget::Integrity => {
let res = if matches!(&describe.target, DescribeTarget::Stats) {
store.cal_stats()
} else {
store.cal_verify()
};
res.map_err(|e| {
let detail = e.to_string();
if detail.contains("AUT-E") {
CalError::NotAuthorized {
detail,
span: None,
store_code: authz_store_code(&e),
}
} else {
CalError::InvalidQuery {
detail,
span: None,
store_code: Some(e.code()),
}
}
})?
}
DescribeTarget::Grammar => serde_json::json!({
"grammar": "CAL/1 grammar (simplified BNF)",
"version": 1,
"conformance_level": 2,
"features": [
"RECALL", "EXISTS", "ASSEMBLE", "HISTORY", "HISTORY DIFF",
"EXPLAIN", "DESCRIBE", "BATCH", "COALESCE",
"SET operations (UNION, INTERSECT, EXCEPT)",
"Pipeline stages (SELECT, ORDER BY, LIMIT, OFFSET, COUNT, FIRST, SUBJECTS, OBJECTS, HASHES, GROUP BY, PROJECT)",
"WITH options (superseded, score_breakdown, explanation, contradiction_detection, diversity)",
"LET bindings", "WHERE clause", "ABOUT semantic search",
"SINCE / BETWEEN temporal filters", "RECENT shorthand"
],
"url": "https://github.com/AreevAI/areev"
}),
};
Ok(CalResultPayload::Describe { info })
}
fn execute_explain(
&self,
explain: &ExplainStmt,
store: &dyn CalStoreFacade,
query: &CalQuery,
) -> std::result::Result<CalResultPayload, CalError> {
let inner = explain.inner.as_ref();
let stmt_type = statement_type_name(inner);
let (grain_type, query_routing, index_usage, mut filters) = match inner {
CalStatement::Recall(recall) => {
let gt = recall
.grain_type
.to_grain_type()
.map(|g| g.as_str().to_string());
let mut index_usage = Vec::new();
let mut filters = Vec::new();
let free_text = recall.about.is_some() || recall.like.is_some();
if free_text {
index_usage.push("bm25_fts".to_string());
}
if let Some(ref wc) = recall.where_clause {
collect_filter_names(&wc.condition, &mut filters);
if filters
.iter()
.any(|f| f == "subject" || f == "relation" || f == "object")
{
index_usage.push("hexastore".to_string());
}
}
if recall.since.is_some() || recall.until.is_some() || recall.between.is_some() {
filters.push("temporal".to_string());
}
let routing = if free_text && !index_usage.contains(&"hexastore".to_string()) {
"bm25".to_string()
} else if free_text {
"hybrid_rrf".to_string()
} else {
"structural".to_string()
};
(gt, routing, index_usage, filters)
}
CalStatement::Exists(exists) => {
let gt = exists
.grain_type
.to_grain_type()
.map(|g| g.as_str().to_string());
let routing = "structural".to_string();
let index_usage = vec!["hexastore".to_string()];
let mut filters = Vec::new();
if let Some(ref wc) = exists.where_clause {
collect_filter_names(&wc.condition, &mut filters);
}
(gt, routing, index_usage, filters)
}
CalStatement::History(h) => {
let mut filters = vec!["hash".to_string()];
if h.diff_target.is_some() {
filters.push("diff_target".to_string());
}
if h.where_clause.is_some() {
filters.push("where_clause".to_string());
}
(
None,
if h.diff_target.is_some() {
"diff_comparison".to_string()
} else {
"entity_latest".to_string()
},
vec![
"entity_latest".to_string(),
"supersession_chain".to_string(),
],
filters,
)
}
CalStatement::Assemble(assemble) => {
let mut index_usage = Vec::new();
let mut filters = Vec::new();
let source_count = assemble.sources.as_ref().map_or(1, |s| s.len());
filters.push(format!("sources: {}", source_count));
if assemble.budget.is_some() {
filters.push("budget_allocation".to_string());
}
for opt in &assemble.assemble_with {
let super::ast::AssembleWithOption::Dedup { .. } = opt;
filters.push("dedup".to_string());
}
index_usage.push("bm25_fts".to_string());
index_usage.push("hexastore".to_string());
(
None,
format!("multi_source_assemble({}_sources)", source_count),
index_usage,
filters,
)
}
CalStatement::Coalesce(coalesce) => {
let gt = coalesce
.grain_type
.to_grain_type()
.map(|g| g.as_str().to_string());
let branch_count = if coalesce.branches.is_empty() {
1
} else {
coalesce.branches.len()
};
let has_else = coalesce.else_branch.is_some();
let mut filters = vec![format!("fallback_chain: {} branches", branch_count)];
if has_else {
filters.push("else_fallback".to_string());
}
(
gt,
"coalesce_fallback".to_string(),
vec!["bm25_fts".to_string(), "hexastore".to_string()],
filters,
)
}
CalStatement::Batch(batch) => {
let stmt_count = batch.statements.len();
let has_labels = batch.labeled.is_some();
let has_formats = batch.statements.iter().any(|e| e.format.is_some());
let has_pipelines = batch.statements.iter().any(|e| !e.pipeline.is_empty());
let mut filters = vec![format!("parallel_execution: {} statements", stmt_count)];
if has_labels {
filters.push("labeled_results".to_string());
}
if has_formats {
filters.push("per_entry_format".to_string());
}
if has_pipelines {
filters.push("per_entry_pipeline".to_string());
}
(None, "parallel_batch".to_string(), vec![], filters)
}
CalStatement::Describe(_) => (None, "introspection".to_string(), vec![], vec![]),
CalStatement::ReportSubject(rs) => {
let mut filters = vec![format!("subject: {}", rs.subject_id)];
if rs.text_mentions {
filters.push("text_mentions".to_string());
}
(
None,
"subject_report".to_string(),
vec!["terms_dictionary".to_string(), "fts_postings".to_string()],
filters,
)
}
CalStatement::Purge(purge) => {
let mut filters = vec![];
if let Some(age) = purge.min_age_days {
filters.push(format!("min_age_days: {age}"));
}
if let Some(ref ns) = purge.namespace {
filters.push(format!("namespace: {ns}"));
}
if let Some(lim) = purge.limit {
filters.push(format!("batch_limit: {lim}"));
}
(
None,
"destructive_purge_stale".to_string(),
vec!["blobs_partition".to_string(), "decay_engine".to_string()],
filters,
)
}
CalStatement::Forget(forget) => {
let target_desc = match &forget.target {
super::ast::ForgetTarget::Hash { hash } => format!("hash: {hash}"),
super::ast::ForgetTarget::User { user_id } => format!("user: {user_id}"),
super::ast::ForgetTarget::Scope { scope } => format!("scope: {scope}"),
};
let indexes = match &forget.target {
super::ast::ForgetTarget::Hash { .. } => vec!["blobs_partition".to_string()],
_ => vec!["blobs_partition".to_string(), "key_store".to_string()],
};
(
None,
"destructive_forget".to_string(),
indexes,
vec![target_desc],
)
}
CalStatement::DefineTemplate(def) => (
None,
"template_registry_write".to_string(),
vec!["meta_partition".to_string()],
vec![format!("template_name: {}", def.name)],
),
CalStatement::DropTemplate(drop) => (
None,
"template_registry_delete".to_string(),
vec!["meta_partition".to_string()],
vec![format!("template_name: {}", drop.name)],
),
CalStatement::DefineQuery(def) => (
None,
"query_registry_write".to_string(),
vec!["meta_partition".to_string()],
vec![format!("query_name: {}", def.name)],
),
CalStatement::DropQuery(drop) => (
None,
"query_registry_delete".to_string(),
vec!["meta_partition".to_string()],
vec![format!("query_name: {}", drop.name)],
),
CalStatement::RunQuery(run) => (
None,
"saved_query_execute".to_string(),
vec!["meta_partition".to_string()],
vec![format!("query_name: {}", run.name)],
),
_ => (None, "unknown".to_string(), vec![], vec![]),
};
let mut pipeline_steps = vec![format!("execute_{}", stmt_type)];
for stage in &query.pipeline {
pipeline_steps.push(pipeline_stage_name(stage));
}
let mut policy_filters = Vec::new();
if self.config.namespace_override.is_some() {
policy_filters.push("namespace_override (capability token)".to_string());
} else if store.default_namespace().is_some() {
policy_filters.push("namespace_scope (session)".to_string());
}
if self.config.user_id_override.is_some() {
policy_filters.push("user_id_override (capability token)".to_string());
} else if store.active_user().is_some() {
policy_filters.push("user_id_scope (session)".to_string());
}
if !self.config.tier1_enabled {
policy_filters.push("tier1_disabled (read-only mode)".to_string());
}
filters.extend(policy_filters);
let ns_context = if self.config.namespace_override.is_some() {
"(namespace: from capability token)"
} else if store.default_namespace().is_some() {
"(namespace: from session)"
} else {
"(namespace: not set)"
};
let cost = match index_usage.len() {
0 => "O(n) full scan",
1 => "O(log n) index lookup",
_ => "O(log n) hybrid with RRF fusion",
};
Ok(CalResultPayload::Explain {
plan: CalQueryPlan {
statement_type: stmt_type,
grain_type,
query_routing,
index_usage,
estimated_cost: format!("{} {}", cost, ns_context),
filters,
pipeline: pipeline_steps,
},
})
}
fn execute_batch(
&self,
batch: &BatchStmt,
store: &dyn CalStoreFacade,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
if let Some(ref labeled) = batch.labeled {
let mut results: HashMap<String, CalResultPayload> = HashMap::new();
let mut seen_labels: std::collections::HashSet<String> =
std::collections::HashSet::new();
for (label, entry) in labeled {
if !seen_labels.insert(label.clone()) {
return Err(CalError::AssembleDuplicateLabel {
label: label.clone(),
span: batch.span,
});
}
let payload = self.execute_batch_entry(entry, store, exec_warnings)?;
results.insert(label.clone(), payload);
}
return Ok(CalResultPayload::Batch { results });
}
let mut results: HashMap<String, CalResultPayload> = HashMap::new();
for (idx, entry) in batch.statements.iter().enumerate() {
let payload = self.execute_batch_entry(entry, store, exec_warnings)?;
results.insert(idx.to_string(), payload);
}
Ok(CalResultPayload::Batch { results })
}
fn execute_batch_entry(
&self,
entry: &BatchEntry,
store: &dyn CalStoreFacade,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let surrogate_query = crate::ast::CalQuery {
version: crate::ast::CalVersion(1),
statement: entry.statement.clone(),
pipeline: entry.pipeline.clone(),
with_options: entry.with_options.clone(),
format: entry.format.clone(),
let_bindings: Vec::new(),
let_values: HashMap::new(),
user_vars: entry.user_vars.clone(),
warnings: Vec::new(),
};
let payload = self
.execute_statement(&entry.statement, store, &surrogate_query, exec_warnings)
.unwrap_or_else(|e| CalResultPayload::Unsupported {
statement: statement_type_name(&entry.statement),
message: e.to_string(),
});
let (payload, grouped_by) = if entry.pipeline.is_empty() {
(payload, None)
} else {
self.apply_pipeline(payload, &entry.pipeline, exec_warnings)?
};
let payload = apply_format_clause(
payload,
&entry.format,
grouped_by.as_deref(),
RenderInputs {
user_vars: &entry.user_vars,
store,
disclosure: disclosure_of(&entry.with_options),
},
None,
exec_warnings,
)?;
Ok(payload)
}
fn execute_coalesce(
&self,
coalesce: &CoalesceStmt,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
if !coalesce.branches.is_empty() {
for (i, branch) in coalesce.branches.iter().enumerate() {
let surrogate = CalQuery {
version: query.version,
statement: branch.query.clone(),
pipeline: Vec::new(),
with_options: query.with_options.clone(),
format: None,
let_bindings: Vec::new(),
let_values: query.let_values.clone(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let payload =
self.execute_statement(&surrogate.statement, store, &surrogate, exec_warnings)?;
let is_non_empty = match &payload {
CalResultPayload::Grains { grains, .. } => !grains.is_empty(),
CalResultPayload::Exists { exists, .. } => *exists,
CalResultPayload::History { versions } => !versions.is_empty(),
CalResultPayload::Count { count } => *count > 0,
_ => true,
};
if is_non_empty {
exec_warnings.push(format!(
"COALESCE: branch {} returned results; short-circuited {} remaining branch(es)",
i + 1,
coalesce.branches.len() - i - 1
+ if coalesce.else_branch.is_some() { 1 } else { 0 }
));
return Ok(payload);
}
}
if let Some(ref else_stmt) = coalesce.else_branch {
let surrogate = CalQuery {
version: query.version,
statement: *else_stmt.clone(),
pipeline: Vec::new(),
with_options: query.with_options.clone(),
format: None,
let_bindings: Vec::new(),
let_values: query.let_values.clone(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
return self.execute_statement(
&surrogate.statement,
store,
&surrogate,
exec_warnings,
);
}
return Ok(CalResultPayload::Grains {
grains: Vec::new(),
total_available: Some(0),
});
}
let recall = RecallStmt {
grain_type: coalesce.grain_type.clone(),
about: None,
where_clause: coalesce.where_clause.clone(),
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: coalesce.span,
};
let payload = self.execute_recall(&recall, store, query, exec_warnings)?;
match payload {
CalResultPayload::Grains { mut grains, .. } => {
grains.truncate(1);
let count = grains.len();
Ok(CalResultPayload::Grains {
grains,
total_available: Some(count),
})
}
other => Ok(other),
}
}
fn execute_set_op(
&self,
set_op: &SetOpStmt,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let mut operand_results: Vec<Vec<CalGrainResult>> = Vec::new();
for stmt in &set_op.operands {
let surrogate = crate::ast::CalQuery {
version: query.version,
statement: stmt.clone(),
pipeline: Vec::new(),
with_options: query.with_options.clone(),
format: None,
let_bindings: Vec::new(),
let_values: query.let_values.clone(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let payload = self.execute_statement(stmt, store, &surrogate, exec_warnings)?;
let grains = extract_grains(payload);
operand_results.push(grains);
}
if operand_results.is_empty() {
return Ok(CalResultPayload::Grains {
grains: Vec::new(),
total_available: Some(0),
});
}
let mut result = operand_results.remove(0);
for next in operand_results {
result = match set_op.op {
SetOp::Union => union_grains(result, next),
SetOp::Intersect => intersect_grains(result, &next),
SetOp::Except => except_grains(result, &next),
};
}
let count = result.len();
Ok(CalResultPayload::Grains {
grains: result,
total_available: Some(count),
})
}
fn sample_assembly_manifest(&self) -> bool {
let rate = self.config.assembly_manifest_sample_rate.clamp(0.0, 1.0);
if rate <= 0.0 {
return false;
}
if rate >= 1.0 {
return true;
}
use std::sync::atomic::{AtomicU64, Ordering};
static SAMPLE_SEQ: AtomicU64 = AtomicU64::new(0);
let n = SAMPLE_SEQ.fetch_add(1, Ordering::Relaxed);
let bucket = n.wrapping_mul(0x9e3779b97f4a7c15) % 1_000_000;
bucket < (rate * 1_000_000.0) as u64
}
#[allow(clippy::too_many_arguments)]
fn record_assembly_manifest(
&self,
assemble: &AssembleStmt,
store: &dyn CalStoreFacade,
candidates: Vec<String>,
included: Vec<String>,
payload: &CalResultPayload,
budget: serde_json::Value,
sources: serde_json::Value,
) {
if !self.sample_assembly_manifest() {
return;
}
let included_set: std::collections::HashSet<&str> =
included.iter().map(String::as_str).collect();
let dropped_hashes = candidates
.into_iter()
.filter(|h| !included_set.contains(h.as_str()))
.collect();
let rendered = match payload {
CalResultPayload::Formatted { text, .. } => text.as_bytes().to_vec(),
CalResultPayload::MultiFormatted { formats, .. } => {
let sorted: std::collections::BTreeMap<&str, &str> = formats
.iter()
.map(|(k, v)| (k.as_str(), v.as_str()))
.collect();
serde_json::to_vec(&sorted).unwrap_or_default()
}
_ => serde_json::to_vec(payload).unwrap_or_default(),
};
let mut digest = Sha256::new();
digest.update(&rendered);
let mut query_digest = Sha256::new();
query_digest.update(
serde_json::to_vec(assemble)
.unwrap_or_default()
.as_slice(),
);
store.note_assembly_manifest(&AssemblyManifest {
query_sha256: hex::encode(query_digest.finalize()),
included_hashes: included,
rendered_sha256: hex::encode(digest.finalize()),
budget,
dropped_hashes,
sources,
});
}
fn execute_assemble(
&self,
assemble: &AssembleStmt,
store: &dyn CalStoreFacade,
query: &CalQuery,
exec_warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let merged;
let query = if assemble.with_options.is_empty() {
query
} else {
merged = CalQuery {
with_options: query
.with_options
.iter()
.chain(&assemble.with_options)
.cloned()
.collect(),
..query.clone()
};
&merged
};
if assemble.streaming {
return Ok(CalResultPayload::StreamAssemble {
assemble: Box::new(assemble.clone()),
with_options: query.with_options.clone(),
});
}
if assemble.sources.is_some() {
let engine = super::assemble::AssembleEngine::new(self);
let result = engine.execute(assemble, store, query, exec_warnings)?;
let (candidates, source_snapshot, budget_snapshot) = match &result {
CalResultPayload::Assembled {
grains,
sources,
total_tokens,
budget_limit,
..
} => {
let mut all = Vec::new();
let mut offset = 0usize;
for source in sources {
let end = (offset + source.grain_count).min(grains.len());
all.extend(grains[offset..end].iter().map(|g| g.hash.clone()));
all.extend(source.omitted.iter().map(|g| g.hash.clone()));
offset = end;
}
(
all,
serde_json::to_value(sources).unwrap_or(serde_json::Value::Null),
serde_json::json!({
"limit": budget_limit,
"unit": "tokens",
"used": total_tokens,
}),
)
}
_ => (Vec::new(), serde_json::Value::Null, serde_json::Value::Null),
};
let result = if !query.with_options.is_empty() {
self.apply_assemble_post_merge_options(
result,
&query.with_options,
exec_warnings,
store,
&assemble.topic,
)?
} else {
result
};
let included = match &result {
CalResultPayload::Assembled { grains, .. } => {
grains.iter().map(|g| g.hash.clone()).collect()
}
_ => Vec::new(),
};
let result = if assemble.format.is_some() {
apply_format_clause(
result,
&assemble.format,
None,
RenderInputs {
user_vars: &query.user_vars,
store,
disclosure: disclosure_of(&assemble.with_options)
.or_else(|| disclosure_of(&query.with_options)),
},
Some((
assemble.context_name.as_deref().unwrap_or(&assemble.topic),
assemble.for_whom.as_deref().unwrap_or(""),
)),
exec_warnings,
)?
} else {
result
};
self.record_assembly_manifest(
assemble,
store,
candidates,
included,
&result,
budget_snapshot,
source_snapshot,
);
return Ok(result);
}
let base_grains = match &assemble.from {
Source::Query(recall_stmt) => {
let surrogate = crate::ast::CalQuery {
version: query.version,
statement: CalStatement::Recall(*recall_stmt.clone()),
pipeline: Vec::new(),
with_options: query.with_options.clone(),
format: None,
let_bindings: Vec::new(),
let_values: query.let_values.clone(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let payload =
self.execute_statement(&surrogate.statement, store, &surrogate, exec_warnings)?;
extract_grains(payload)
}
Source::Hashes(hashes) => {
let mut grains = Vec::new();
for hs in hashes {
match Hash::from_hex(hs) {
Ok(hash) => {
if let Ok(grain) = store.get(&hash) {
grains.push(CalGrainResult {
hash: hs.clone(),
grain_type: grain.grain_type.as_str().to_string(),
score: 1.0,
fields: serde_json::Value::Object(
grain
.fields
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
});
}
}
Err(_) => {
}
}
}
grains
}
Source::Parameter { name, .. } => {
return Ok(CalResultPayload::Unsupported {
statement: "assemble".into(),
message: format!(
"Parameter source ${} is not yet supported in single-source ASSEMBLE. \
Use a RECALL subquery instead.",
name
),
});
}
};
let grains = if let Some(ref wc) = assemble.where_clause {
validate_residual_subtree(
&wc.condition,
&GrainTypePlural::All,
"in ASSEMBLE WHERE (a post-composition filter)",
exec_warnings,
)?;
base_grains
.into_iter()
.filter(|grain| grain_matches_condition_tree(grain, &wc.condition))
.collect()
} else {
base_grains
};
let candidates: Vec<String> = grains.iter().map(|g| g.hash.clone()).collect();
let (grains, omitted) = if let Some(ref budget) = assemble.budget {
let mut grains = grains;
let limit = budget.tokens as usize;
if grains.len() > limit {
let dropped = grains.split_off(limit);
(grains, dropped)
} else {
(grains, Vec::new())
}
} else {
(grains, Vec::new())
};
store.note_assembly_budget(!omitted.is_empty());
let included: Vec<String> = grains.iter().map(|g| g.hash.clone()).collect();
let budget_snapshot = serde_json::json!({
"limit": assemble.budget.as_ref().map(|b| b.tokens),
"unit": "grains",
"used": grains.len(),
});
let source_snapshot = serde_json::json!([{
"label": "",
"grain_count": grains.len(),
"dropped_count": omitted.len(),
}]);
let result = if let Some(ref clause) = assemble.format {
let ctx = super::templates::AssemblyContext {
name: assemble
.context_name
.clone()
.unwrap_or_else(|| assemble.topic.clone()),
intent: assemble.for_whom.clone().unwrap_or_default(),
source_count: 1,
grain_count: grains.len(),
budget_total: assemble.budget.as_ref().map_or(0, |b| b.tokens as u64),
budget_used: grains.len() as u64,
budget_unit: "grains".to_string(),
};
let sources = [super::templates::RenderSource {
label: "",
grains: &grains,
omitted: &omitted,
priority: 1,
tokens_used: grains.len(),
truncated: !omitted.is_empty(),
}];
let plan = super::templates::RenderPlan {
assembly: Some(&ctx),
sources: Some(&sources),
};
apply_format_clause_to_grains(
&grains,
clause,
None,
RenderInputs {
user_vars: &query.user_vars,
store,
disclosure: disclosure_of(&query.with_options),
},
&plan,
exec_warnings,
)?
} else {
let count = grains.len();
CalResultPayload::Grains {
grains,
total_available: Some(count),
}
};
self.record_assembly_manifest(
assemble,
store,
candidates,
included,
&result,
budget_snapshot,
source_snapshot,
);
Ok(result)
}
fn validate_pipeline_fields(
stages: &[PipelineStage],
grain_type: &GrainTypePlural,
) -> std::result::Result<(), CalError> {
let check = |field: &str, span: Option<Span>| -> std::result::Result<(), CalError> {
if COMMON_FIELDS.contains(&field) {
return Ok(());
}
if field.contains(':') {
return Ok(());
}
let allowed = type_specific_fields(grain_type);
if allowed.contains(&field) {
return Ok(());
}
let suggestion = suggest_field(field, grain_type);
Err(CalError::FieldNotOnGrainType {
field: field.to_string(),
grain_type: grain_type.as_str().to_string(),
span,
suggestion,
})
};
for stage in stages {
match stage {
PipelineStage::Select { fields, span } => {
for f in fields {
check(f, *span)?;
}
}
PipelineStage::OrderBy { field, span, .. } => check(field, *span)?,
PipelineStage::GroupBy { fields, span } => {
for f in fields {
check(f, *span)?;
}
}
PipelineStage::Project { fields, span } => {
for pf in fields {
check(&pf.field, *span)?;
}
}
_ => {}
}
}
Ok(())
}
pub(crate) fn apply_pipeline(
&self,
payload: CalResultPayload,
stages: &[PipelineStage],
exec_warnings: &mut Vec<String>,
) -> std::result::Result<(CalResultPayload, Option<Vec<String>>), CalError> {
let mut current = payload;
let mut grouped_by: Option<Vec<String>> = None;
for stage in stages {
current = match (current, stage) {
(CalResultPayload::Grains { grains, .. }, PipelineStage::Limit { value, .. }) => {
let capped = (*value).min(self.config.max_limit) as usize;
let limited: Vec<_> = grains.into_iter().take(capped).collect();
let count = limited.len();
CalResultPayload::Grains {
grains: limited,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Offset { value, .. }) => {
let offset: Vec<_> = grains.into_iter().skip(*value as usize).collect();
let count = offset.len();
CalResultPayload::Grains {
grains: offset,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Count { .. }) => {
match grouped_by.as_deref() {
Some(fields) => {
let groups = count_by_fields(&grains, fields);
CalResultPayload::GroupCounts {
field: fields.join(", "),
total_available: Some(groups.len()),
groups,
}
}
None => CalResultPayload::Count {
count: grains.len(),
},
}
}
(
CalResultPayload::GroupCounts {
field,
groups,
total_available,
},
PipelineStage::Limit { value, .. },
) => {
let capped = (*value).min(self.config.max_limit) as usize;
CalResultPayload::GroupCounts {
field,
groups: groups.into_iter().take(capped).collect(),
total_available,
}
}
(
CalResultPayload::GroupCounts {
field,
groups,
total_available,
},
PipelineStage::Offset { value, .. },
) => CalResultPayload::GroupCounts {
field,
groups: groups.into_iter().skip(*value as usize).collect(),
total_available,
},
(
CalResultPayload::GroupCounts {
field,
groups,
total_available,
},
PipelineStage::First { .. },
) => CalResultPayload::GroupCounts {
field,
groups: groups.into_iter().take(1).collect(),
total_available,
},
(CalResultPayload::Grains { grains, .. }, PipelineStage::First { .. }) => {
let first: Vec<_> = grains.into_iter().take(1).collect();
CalResultPayload::Grains {
grains: first,
total_available: Some(1),
}
}
(
CalResultPayload::Grains { grains, .. },
PipelineStage::OrderBy {
field, descending, ..
},
) => {
let mut sorted = grains;
sorted.sort_by(|a, b| {
let va = json_field(&a.fields, field);
let vb = json_field(&b.fields, field);
let cmp = compare_json_values(va, vb);
if *descending {
cmp.reverse()
} else {
cmp
}
});
let count = sorted.len();
CalResultPayload::Grains {
grains: sorted,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Select { fields, .. }) => {
let projected: Vec<_> = grains
.into_iter()
.map(|g| {
let projected_fields = project_fields(&g.fields, fields);
CalGrainResult {
fields: projected_fields,
..g
}
})
.collect();
let count = projected.len();
CalResultPayload::Grains {
grains: projected,
total_available: Some(count),
}
}
(
CalResultPayload::Grains { grains, .. },
PipelineStage::Project { fields, .. },
) => {
let field_names: Vec<String> =
fields.iter().map(|pf| pf.field.clone()).collect();
let projected: Vec<_> = grains
.into_iter()
.map(|g| {
let mut new_map = serde_json::Map::new();
if let serde_json::Value::Object(map) = &g.fields {
for pf in fields {
if let Some(v) = map.get(&pf.field) {
let key = pf.alias.as_deref().unwrap_or(&pf.field);
new_map.insert(key.to_string(), v.clone());
}
}
}
let _ = field_names.len(); CalGrainResult {
fields: serde_json::Value::Object(new_map),
..g
}
})
.collect();
let count = projected.len();
CalResultPayload::Grains {
grains: projected,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Subjects { .. }) => {
let extracted: Vec<CalGrainResult> = grains
.iter()
.filter_map(|g| {
json_field(&g.fields, "subject").map(|v| CalGrainResult {
hash: String::new(),
grain_type: "extracted".into(),
score: 0.0,
fields: serde_json::json!({ "value": v }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
})
})
.collect();
let count = extracted.len();
CalResultPayload::Grains {
grains: extracted,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Objects { .. }) => {
let extracted: Vec<CalGrainResult> = grains
.iter()
.filter_map(|g| {
json_field(&g.fields, "object").map(|v| CalGrainResult {
hash: String::new(),
grain_type: "extracted".into(),
score: 0.0,
fields: serde_json::json!({ "value": v }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
})
})
.collect();
let count = extracted.len();
CalResultPayload::Grains {
grains: extracted,
total_available: Some(count),
}
}
(CalResultPayload::Grains { grains, .. }, PipelineStage::Hashes { .. }) => {
let extracted: Vec<CalGrainResult> = grains
.iter()
.map(|g| CalGrainResult {
hash: String::new(),
grain_type: "extracted".into(),
score: 0.0,
fields: serde_json::json!({ "value": g.hash }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
})
.collect();
let count = extracted.len();
CalResultPayload::Grains {
grains: extracted,
total_available: Some(count),
}
}
(
CalResultPayload::Grains {
grains,
total_available,
},
PipelineStage::GroupBy { fields, .. },
) => {
for f in fields {
if !grains.is_empty()
&& !grains
.iter()
.any(|g| json_field(&g.fields, f).is_some())
{
exec_warnings.push(
super::errors::CalWarning::GroupKeyAbsent {
field: f.clone(),
grains: grains.len(),
}
.to_string(),
);
}
}
grouped_by = Some(fields.clone());
let grouped = group_grains_by_field(grains, fields);
CalResultPayload::Grains {
grains: grouped,
total_available,
}
}
(
CalResultPayload::Grains { grains, .. },
PipelineStage::Filter { condition, .. },
) => {
let filtered: Vec<_> = grains
.into_iter()
.filter(|grain| grain_matches_condition_tree(grain, condition))
.collect();
let count = filtered.len();
CalResultPayload::Grains {
grains: filtered,
total_available: Some(count),
}
}
(other, stage) => {
exec_warnings.push(
super::errors::CalWarning::PipelineStageInert {
stage: pipeline_stage_name(stage),
payload: payload_kind_name(&other),
why: inert_stage_reason(&other),
}
.to_string(),
);
other
}
};
}
Ok((current, grouped_by))
}
}
fn diff_grains(
a: &areev_core::format::deserialize::DeserializedGrain,
b: &areev_core::format::deserialize::DeserializedGrain,
) -> Vec<super::ast::FieldDiff> {
use std::collections::BTreeSet;
let keys_a: BTreeSet<&str> = a.fields.keys().map(|s| s.as_str()).collect();
let keys_b: BTreeSet<&str> = b.fields.keys().map(|s| s.as_str()).collect();
let mut diffs = Vec::new();
for key in keys_b.difference(&keys_a) {
diffs.push(super::ast::FieldDiff::Added {
field: key.to_string(),
value: b.fields[*key].clone(),
});
}
for key in keys_a.difference(&keys_b) {
diffs.push(super::ast::FieldDiff::Removed {
field: key.to_string(),
value: a.fields[*key].clone(),
});
}
for key in keys_a.intersection(&keys_b) {
if a.fields[*key] != b.fields[*key] {
diffs.push(super::ast::FieldDiff::Changed {
field: key.to_string(),
old: a.fields[*key].clone(),
new: b.fields[*key].clone(),
});
}
}
diffs
}
#[allow(clippy::vec_init_then_push)]
fn build_features_list() -> Vec<&'static str> {
let mut features = Vec::new();
#[cfg(feature = "http")]
features.push("http");
#[cfg(feature = "grpc")]
features.push("grpc");
#[cfg(feature = "mcp")]
features.push("mcp");
#[cfg(feature = "a2a")]
features.push("a2a");
#[cfg(feature = "app")]
features.push("app");
#[cfg(feature = "import")]
features.push("import");
#[cfg(feature = "auth")]
features.push("auth");
#[cfg(feature = "rerank")]
features.push("rerank");
#[cfg(feature = "llm-rerank")]
features.push("llm-rerank");
#[cfg(feature = "pii_ner")]
features.push("pii_ner");
#[cfg(feature = "cal")]
features.push("cal");
features
}
fn compute_query_hash(input: &str) -> String {
use unicode_normalization::UnicodeNormalization as _;
let normalized: String = input.trim().nfc().collect();
let mut hasher = Sha256::new();
hasher.update(normalized.as_bytes());
hex::encode(hasher.finalize())
}
pub(super) fn map_store_err(e: AreevError, span: Option<super::errors::Span>) -> CalError {
let store_code = Some(e.code());
match e {
AreevError::AuthzDenied(_)
| AreevError::AuthzUnknownPrincipal(_)
| AreevError::AuthzConfigInvalid(_)
| AreevError::AuthzTokenUnrecognized => CalError::NotAuthorized {
detail: e.to_string(),
span,
store_code,
},
AreevError::CryptoError(_) => CalError::CryptoError {
detail: e.to_string(),
span,
store_code,
},
AreevError::Validation(_) => CalError::InvalidQuery {
detail: e.to_string(),
span,
store_code,
},
_ => CalError::StoreError {
detail: e.to_string(),
span,
store_code,
},
}
}
fn authz_store_code(e: &AreevError) -> Option<&'static str> {
let code = e.code();
if code.starts_with("AUT-") {
return Some(code);
}
let text = e.to_string();
["AUT-E001", "AUT-E002", "AUT-E003", "AUT-E004"]
.into_iter()
.find(|c| text.contains(c))
.or(Some(code))
}
fn build_add_options(opts: &[AddWithOption], warnings: &mut Vec<String>) -> AddOptions {
let mut options = AddOptions::default();
for opt in opts {
match opt {
AddWithOption::ExtractEventDate => {
options.extract_event_date = Some(true);
}
AddWithOption::AutoRelate => {
warnings.push(
"CAL-W013: WITH auto_relate is accepted but not implemented — no relations \
are inferred. Link grains explicitly with related_to, or use WITH \
multi_hop(n) on recall to widen through the entity graph."
.to_string(),
);
options.auto_relate = Some(true);
}
AddWithOption::ExtractMemories => {
}
AddWithOption::Sync => {
options.sync = Some(true);
}
AddWithOption::Occurrence => {}
}
}
options
}
fn value_to_cal_literal(value: &super::ast::Value) -> String {
match value {
super::ast::Value::String { value } => {
format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
}
super::ast::Value::Number { value } => value.to_string(),
super::ast::Value::Boolean { value } => value.to_string(),
super::ast::Value::Hash { value } => format!("#{value}"),
super::ast::Value::Parameter { name } => format!("${name}"),
super::ast::Value::Array { values } => {
let items: Vec<String> = values.iter().map(value_to_cal_literal).collect();
format!("[{}]", items.join(", "))
}
}
}
fn cal_value_to_json(val: &super::ast::Value) -> serde_json::Value {
match val {
super::ast::Value::String { value } => serde_json::Value::String(value.clone()),
super::ast::Value::Number { value } => {
if value.fract() == 0.0 && value.abs() < 9.007_199_254_740_992e15 {
serde_json::Value::Number(serde_json::Number::from(*value as i64))
} else {
serde_json::Number::from_f64(*value)
.map(serde_json::Value::Number)
.unwrap_or(serde_json::Value::Null)
}
}
super::ast::Value::Boolean { value } => serde_json::Value::Bool(*value),
super::ast::Value::Array { values } => {
serde_json::Value::Array(values.iter().map(cal_value_to_json).collect())
}
super::ast::Value::Hash { value } => serde_json::Value::String(value.clone()),
super::ast::Value::Parameter { name } => serde_json::Value::String(format!("${}", name)),
}
}
fn governance_unwired(statement: &str) -> CalResultPayload {
CalResultPayload::Unsupported {
statement: statement.into(),
message: "governance is not wired on this surface — the host attaches a \
GovernanceHost (CalExecutor::with_governance) to back RUN LOOP, \
APPROVE, REJECT, APPLY, ROLLBACK, and the loop DESCRIBE reads"
.into(),
}
}
fn statement_type_name(stmt: &CalStatement) -> String {
match stmt {
CalStatement::Recall(_) => "recall",
CalStatement::SetOp(_) => "set_op",
CalStatement::Exists(_) => "exists",
CalStatement::Assemble(_) => "assemble",
CalStatement::History(_) => "history",
CalStatement::Explain(_) => "explain",
CalStatement::Describe(_) => "describe",
CalStatement::Batch(_) => "batch",
CalStatement::Coalesce(_) => "coalesce",
CalStatement::Add(_) => "add",
CalStatement::AddWorkflow(_) => "add_workflow",
CalStatement::Supersede(_) => "supersede",
CalStatement::SupersedeWorkflow(_) => "supersede_workflow",
CalStatement::Accumulate(_) => "accumulate",
CalStatement::Revert(_) => "revert",
CalStatement::Forget(_) => "forget",
CalStatement::Purge(_) => "purge",
CalStatement::ReportSubject(_) => "report_subject",
CalStatement::DefineTemplate(_) => "define_template",
CalStatement::DropTemplate(_) => "drop_template",
CalStatement::DefineQuery(_) => "define_query",
CalStatement::DropQuery(_) => "drop_query",
CalStatement::RunQuery(_) => "run_query",
CalStatement::Grant(_) => "grant",
CalStatement::Revoke(_) => "revoke",
CalStatement::ShowGrants(_) => "show_grants",
CalStatement::Approve(_) => "approve",
CalStatement::Reject(_) => "reject",
CalStatement::ApplyRec(_) => "apply",
CalStatement::RollbackRec(_) => "rollback",
CalStatement::RunLoop(_) => "run_loop",
CalStatement::Remember(_) => "remember",
CalStatement::EntityAt(_) => "entity_at",
CalStatement::RunTrace(_) => "run_trace",
CalStatement::RunsTouching(_) => "runs_touching",
CalStatement::DerivedFrom(_) => "derived_from",
CalStatement::ShowForks(_) => "show_forks",
CalStatement::Merge(_) => "merge",
CalStatement::Related(_) => "related",
CalStatement::Novelty(_) => "novelty",
}
.to_string()
}
fn required_scope_for_statement(stmt: &CalStatement) -> &'static str {
match stmt {
CalStatement::Recall(_)
| CalStatement::SetOp(_)
| CalStatement::Exists(_)
| CalStatement::Assemble(_)
| CalStatement::History(_)
| CalStatement::Explain(_)
| CalStatement::Describe(_)
| CalStatement::Batch(_)
| CalStatement::Coalesce(_)
| CalStatement::RunQuery(_)
| CalStatement::ShowGrants(_)
| CalStatement::EntityAt(_)
| CalStatement::RunTrace(_)
| CalStatement::RunsTouching(_)
| CalStatement::DerivedFrom(_)
| CalStatement::ShowForks(_)
| CalStatement::Related(_)
| CalStatement::Novelty(_)
| CalStatement::ReportSubject(_) => "read",
CalStatement::Add(_)
| CalStatement::AddWorkflow(_)
| CalStatement::Supersede(_)
| CalStatement::SupersedeWorkflow(_)
| CalStatement::Accumulate(_)
| CalStatement::Revert(_)
| CalStatement::Remember(_)
| CalStatement::Merge(_) => "write",
CalStatement::Forget(_)
| CalStatement::Purge(_)
| CalStatement::DefineTemplate(_)
| CalStatement::DropTemplate(_)
| CalStatement::DefineQuery(_)
| CalStatement::DropQuery(_)
| CalStatement::Grant(_)
| CalStatement::Revoke(_)
| CalStatement::Approve(_)
| CalStatement::Reject(_)
| CalStatement::ApplyRec(_)
| CalStatement::RollbackRec(_)
| CalStatement::RunLoop(_) => "admin",
}
}
fn payload_kind_name(payload: &CalResultPayload) -> &'static str {
match payload {
CalResultPayload::Assembled { .. } => "assembled",
CalResultPayload::Count { .. } => "count",
CalResultPayload::GroupCounts { .. } => "group counts",
CalResultPayload::Formatted { .. } => "formatted",
CalResultPayload::Exists { .. } => "exists",
CalResultPayload::History { .. } => "history",
CalResultPayload::Explain { .. } => "explain",
CalResultPayload::Batch { .. } => "batch",
_ => "non-grain",
}
}
fn inert_stage_reason(payload: &CalResultPayload) -> &'static str {
match payload {
CalResultPayload::Assembled { .. } => {
"an assembly is a list of named sections, not a flat grain list — \
section order is FROM-clause order (PRIORITY weights the budget, \
it does not reorder); to order WITHIN a section, put the stage on \
that source's sub-query"
}
CalResultPayload::Count { .. } => "a count is a scalar; stage it before | COUNT",
CalResultPayload::GroupCounts { .. } => {
"a grouped count is already one row per group, ordered most \
frequent first; stage it before GROUP BY … COUNT"
}
CalResultPayload::Formatted { .. } => {
"FORMAT has already rendered the grains to text; stage it before FORMAT"
}
_ => "this statement does not return a grain list",
}
}
fn pipeline_stage_name(stage: &PipelineStage) -> String {
match stage {
PipelineStage::Select { .. } => "SELECT".to_string(),
PipelineStage::OrderBy {
field, descending, ..
} => {
format!(
"ORDER BY {} {}",
field,
if *descending { "DESC" } else { "ASC" }
)
}
PipelineStage::Limit { value, .. } => format!("LIMIT {}", value),
PipelineStage::Offset { value, .. } => format!("OFFSET {}", value),
PipelineStage::Count { .. } => "COUNT".to_string(),
PipelineStage::First { .. } => "FIRST".to_string(),
PipelineStage::Subjects { .. } => "SUBJECTS".to_string(),
PipelineStage::Objects { .. } => "OBJECTS".to_string(),
PipelineStage::Hashes { .. } => "HASHES".to_string(),
PipelineStage::GroupBy { fields, .. } => format!("GROUP BY {}", fields.join(", ")),
PipelineStage::Project { .. } => "PROJECT".to_string(),
PipelineStage::Filter { .. } => "WHERE (post-pipeline)".to_string(),
}
}
fn count_payload_results(payload: &CalResultPayload) -> usize {
match payload {
CalResultPayload::Grains { grains, .. } => grains.len(),
CalResultPayload::Exists { .. } => 1,
CalResultPayload::Count { .. } => 1,
CalResultPayload::GroupCounts { groups, .. } => groups.len(),
CalResultPayload::History { versions } => versions.len(),
CalResultPayload::Describe { .. } => 1,
CalResultPayload::Explain { .. } => 1,
CalResultPayload::Batch { results } => results.len(),
CalResultPayload::Assembled { grains, .. } => grains.len(),
CalResultPayload::Diff { changes, .. } => changes.len(),
CalResultPayload::Formatted { grain_count, .. } => *grain_count,
CalResultPayload::MultiFormatted { grain_count, .. } => *grain_count,
CalResultPayload::Added { .. } => 1,
CalResultPayload::Granted { .. } => 1,
CalResultPayload::Revoked { .. } => 1,
CalResultPayload::GrantList { grants } => grants.len(),
CalResultPayload::LoopRan { .. } => 1,
CalResultPayload::Reviewed { .. } => 1,
CalResultPayload::RecApplied { .. } => 1,
CalResultPayload::RecRolledBack { .. } => 1,
CalResultPayload::Remembered { .. } => 1,
CalResultPayload::EntityAt { grain, .. } => usize::from(grain.is_some()),
CalResultPayload::RunTrace { .. } => 1,
CalResultPayload::RunsTouching { runs, .. } => runs.len(),
CalResultPayload::DerivedFrom { grains, .. } => grains.len(),
CalResultPayload::Forks { forks } => forks.len(),
CalResultPayload::Merged { .. } => 1,
CalResultPayload::RelatedEntities { entities, .. } => entities.len(),
CalResultPayload::NoveltyMatches { matches } => matches.len(),
CalResultPayload::Superseded { .. } => 1,
CalResultPayload::Accumulated { .. } => 1,
CalResultPayload::Forgotten { .. } => 1,
CalResultPayload::Purged { count } => *count,
CalResultPayload::SubjectReport { grains, .. } => grains.len(),
CalResultPayload::Unsupported { .. } => 0,
CalResultPayload::TemplateDefined { .. } => 1,
CalResultPayload::TemplateDropped { .. } => 1,
CalResultPayload::QueryDefined { .. } => 1,
CalResultPayload::QueryDropped { .. } => 1,
CalResultPayload::StreamAssemble { .. } => 0,
}
}
pub(crate) fn extract_grains(payload: CalResultPayload) -> Vec<CalGrainResult> {
match payload {
CalResultPayload::Grains { grains, .. } => grains,
CalResultPayload::Assembled { grains, .. } => grains,
CalResultPayload::GroupCounts { groups, .. } => groups,
_ => Vec::new(),
}
}
fn hits_to_grain_results(hits: &[crate::store_types::SearchHit]) -> Vec<CalGrainResult> {
hits.iter()
.map(|hit| CalGrainResult {
hash: hit.hash.to_hex(),
grain_type: hit.grain.grain_type.as_str().to_string(),
score: hit.score,
fields: serde_json::Value::Object({
let mut map: serde_json::Map<String, serde_json::Value> = hit
.grain
.fields
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
if let Some(sup) = hit.superseded_by_hash {
map.entry("superseded_by".to_string())
.or_insert_with(|| serde_json::Value::String(sup.to_hex()));
}
map
}),
score_breakdown: hit
.score_breakdown
.as_ref()
.map(|sb| serde_json::to_value(sb).unwrap_or(serde_json::Value::Null)),
explanation: hit.explanation.clone(),
relative_time: hit.relative_time.clone(),
is_deterministic: false,
contested_by: None,
})
.collect()
}
fn apply_fork_status(
grains: &mut Vec<CalGrainResult>,
store: &dyn CalStoreFacade,
filter: bool,
restrict_to: Option<&std::collections::HashSet<(String, String, String)>>,
exec_warnings: &mut Vec<String>,
) {
let forks = match store.open_forks() {
Ok(f) => f,
Err(e) => {
exec_warnings.push(format!(
"contradiction detection unavailable ({e}); fork status not applied"
));
if filter {
grains.clear();
}
return;
}
};
let mut peers: HashMap<String, Vec<String>> = HashMap::new();
for f in &forks {
if let Some(keys) = restrict_to {
if !keys.contains(&(
f.namespace.clone(),
f.subject.clone(),
f.relation.clone(),
)) {
continue;
}
}
for (i, head) in f.heads.iter().enumerate() {
let others: Vec<String> = f
.heads
.iter()
.enumerate()
.filter(|(j, _)| *j != i)
.map(|(_, o)| o.clone())
.collect();
peers.insert(head.clone(), others);
}
}
if filter {
grains.retain(|g| peers.contains_key(&g.hash));
}
for g in grains.iter_mut() {
if let Some(others) = peers.get(&g.hash) {
g.contested_by = Some(others.clone());
}
}
}
fn contradiction_scope_keys(
grains: &[CalGrainResult],
default_ns: Option<&str>,
) -> std::collections::HashSet<(String, String, String)> {
grains
.iter()
.filter_map(|g| {
let s = g.fields.get("subject").and_then(|v| v.as_str())?;
let r = g.fields.get("relation").and_then(|v| v.as_str())?;
let ns = g
.fields
.get("namespace")
.and_then(|v| v.as_str())
.or(default_ns)
.unwrap_or_default();
Some((ns.to_string(), s.to_string(), r.to_string()))
})
.collect()
}
fn grain_result_text(grain: &CalGrainResult) -> String {
if grain.grain_type == "fact" {
let s = grain
.fields
.get("subject")
.and_then(|v| v.as_str())
.unwrap_or("");
let r = grain
.fields
.get("relation")
.and_then(|v| v.as_str())
.unwrap_or("");
let o = grain
.fields
.get("object")
.and_then(|v| v.as_str())
.unwrap_or("");
return format!("{} {} {}", s, r, o).trim().to_string();
}
for field in &[
"content",
"description",
"title",
"goal",
"query",
"result",
"output",
] {
if let Some(v) = grain.fields.get(*field).and_then(|v| v.as_str()) {
if !v.trim().is_empty() {
return v.to_string();
}
}
}
grain.grain_type.clone()
}
fn text_similarity(a: &str, b: &str) -> f64 {
if a == b {
return 1.0;
}
let words_a: Vec<&str> = a.split_whitespace().collect();
let words_b: Vec<&str> = b.split_whitespace().collect();
if words_a.is_empty() && words_b.is_empty() {
return 1.0;
}
if words_a.is_empty() || words_b.is_empty() {
return 0.0;
}
let bigrams = |words: &[&str]| -> std::collections::HashSet<(String, String)> {
if words.len() < 2 {
let mut s = std::collections::HashSet::new();
s.insert((words[0].to_lowercase(), String::new()));
return s;
}
words
.windows(2)
.map(|w| (w[0].to_lowercase(), w[1].to_lowercase()))
.collect()
};
let set_a = bigrams(&words_a);
let set_b = bigrams(&words_b);
let intersection = set_a.intersection(&set_b).count();
let union = set_a.union(&set_b).count();
if union == 0 {
return 1.0;
}
intersection as f64 / union as f64
}
fn value_to_string(value: &Value) -> std::result::Result<String, CalError> {
match value {
Value::String { value } => Ok(value.clone()),
Value::Hash { value } => Ok(value.clone()),
Value::Parameter { name } => Err(CalError::UnboundParameter {
name: name.clone(),
span: None,
}),
other => Err(CalError::IncompatibleTypes {
left: "string".into(),
right: format!("{:?}", other),
span: None,
suggestion: Some("expected a quoted string value".into()),
}),
}
}
fn value_to_f64(value: &Value) -> std::result::Result<f64, CalError> {
match value {
Value::Number { value } => Ok(*value),
other => Err(CalError::IncompatibleTypes {
left: "number".into(),
right: format!("{:?}", other),
span: None,
suggestion: Some("expected a numeric value".into()),
}),
}
}
fn extract_hash_from_condition(condition: Option<&Condition>) -> Option<String> {
match condition? {
Condition::Comparison {
field,
comparator: Comparator::Eq,
value: Value::Hash { value },
..
} if field == "hash" => Some(value.clone()),
_ => None,
}
}
#[derive(Clone, Copy)]
struct RenderInputs<'a> {
user_vars: &'a HashMap<String, String>,
store: &'a dyn CalStoreFacade,
disclosure: Option<crate::render::Disclosure>,
}
fn disclosure_of(opts: &[super::ast::WithOption]) -> Option<crate::render::Disclosure> {
opts.iter().find_map(|o| match o {
super::ast::WithOption::ProgressiveDisclosure { level } => Some(match level.as_deref() {
Some("summary") => crate::render::Disclosure::Summary,
Some("headlines") => crate::render::Disclosure::Headlines,
_ => crate::render::Disclosure::Full,
}),
_ => None,
})
}
fn apply_format_clause(
payload: CalResultPayload,
format: &Option<FormatClause>,
grouped_by: Option<&[String]>,
inputs: RenderInputs<'_>,
assemble_ident: Option<(&str, &str)>,
warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let Some(clause) = format else {
return Ok(payload);
};
let (grains, assembled) = match &payload {
CalResultPayload::Grains { grains, .. } => (grains, None),
CalResultPayload::GroupCounts { groups, .. } => (groups, None),
CalResultPayload::Assembled {
grains,
sources,
total_tokens,
budget_limit,
..
} => (grains, Some((sources, *total_tokens, *budget_limit))),
_ => return Ok(payload),
};
let mut render_sources: Vec<super::templates::RenderSource<'_>> = Vec::new();
let mut assembly_ctx: Option<super::templates::AssemblyContext> = None;
if let Some((metas, total_tokens, budget_limit)) = assembled {
let mut offset = 0usize;
for (i, m) in metas.iter().enumerate() {
let end = (offset + m.grain_count).min(grains.len());
render_sources.push(super::templates::RenderSource {
label: &m.label,
grains: &grains[offset..end],
omitted: &m.omitted,
priority: i + 1,
tokens_used: m.tokens_used as usize,
truncated: !m.omitted.is_empty(),
});
offset = end;
}
assembly_ctx = Some(super::templates::AssemblyContext {
name: assemble_ident.map(|(n, _)| n.to_string()).unwrap_or_default(),
intent: assemble_ident.map(|(_, i)| i.to_string()).unwrap_or_default(),
source_count: metas.len(),
grain_count: grains.len(),
budget_total: budget_limit.unwrap_or(0) as u64,
budget_used: total_tokens as u64,
budget_unit: "tokens".to_string(),
});
}
let plan = super::templates::RenderPlan {
assembly: assembly_ctx.as_ref(),
sources: (!render_sources.is_empty()).then_some(render_sources.as_slice()),
};
apply_format_clause_to_grains(
grains, clause, grouped_by, inputs, &plan, warnings,
)
}
fn apply_format_clause_to_grains(
grains: &[CalGrainResult],
clause: &FormatClause,
grouped_by: Option<&[String]>,
inputs: RenderInputs<'_>,
plan: &super::templates::RenderPlan<'_>,
warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
match clause {
FormatClause::Single(super::ast::FormatSpec::Json) if grouped_by.is_none() => {
Ok(CalResultPayload::Grains {
grains: grains.to_vec(),
total_available: Some(grains.len()),
})
}
FormatClause::Single(spec) => {
format_grain_results(
grains, spec, grouped_by, inputs, plan, warnings,
)
}
FormatClause::Multi(entries) => {
let mut formats = HashMap::new();
for entry in entries {
let rendered =
format_grain_results(
grains, &entry.spec, grouped_by, inputs, plan, warnings,
)?;
if let CalResultPayload::Formatted { text, format, .. } = rendered {
let key = entry.alias.clone().unwrap_or(format);
formats.insert(key, text);
}
}
Ok(CalResultPayload::MultiFormatted {
formats,
grain_count: grains.len(),
grains: grains.to_vec(),
})
}
}
}
fn grain_view(grain: &CalGrainResult) -> crate::render::GrainView<'_> {
crate::render::GrainView {
grain_type: &grain.grain_type,
hash: &grain.hash,
fields: &grain.fields,
created_at_sec: crate::render::created_at_sec_from_fields(&grain.fields),
}
}
fn template_tier(
plan: &super::templates::RenderPlan<'_>,
grain_count: usize,
) -> super::templates::DisclosureTier {
match plan.assembly {
Some(a) if a.budget_unit == "tokens" && a.budget_total > 0 => {
super::templates::select_tier(a.budget_total.min(u32::MAX as u64) as u32, grain_count)
}
_ => super::templates::DisclosureTier::Full,
}
}
fn format_grain_results(
grains: &[CalGrainResult],
format: &super::ast::FormatSpec,
grouped_by: Option<&[String]>,
inputs: RenderInputs<'_>,
plan: &super::templates::RenderPlan<'_>,
warnings: &mut Vec<String>,
) -> std::result::Result<CalResultPayload, CalError> {
let RenderInputs { user_vars, store, disclosure } = inputs;
let (text, format_name) = match format {
super::ast::FormatSpec::Json => {
if let Some(fields) = grouped_by {
let groups = collect_groups(grains, fields);
let json_groups: Vec<serde_json::Value> = groups
.iter()
.map(|(key, members)| {
serde_json::json!({
"group_key": key,
"count": members.len(),
"grains": members,
})
})
.collect();
let json = serde_json::to_string_pretty(&json_groups).unwrap_or_default();
(json, "json")
} else {
let json = serde_json::to_string_pretty(grains).unwrap_or_default();
(json, "json")
}
}
super::ast::FormatSpec::Markdown => {
let mut md = String::new();
if let Some(fields) = grouped_by {
let groups = collect_groups(grains, fields);
for (key, members) in &groups {
md.push_str(&format!(
"### {} ({} {})\n\n",
key,
members.len(),
if members.len() == 1 {
"memory"
} else {
"memories"
}
));
for grain in members {
md.push_str(&crate::render::render_grain_markdown_at(
&grain_view(grain),
disclosure,
));
md.push('\n');
}
}
} else {
for grain in grains {
md.push_str(&crate::render::render_grain_markdown_at(
&grain_view(grain),
disclosure,
));
md.push('\n');
}
}
(md, "markdown")
}
super::ast::FormatSpec::Yaml => {
let mut yaml = String::new();
for (i, grain) in grains.iter().enumerate() {
yaml.push_str(&format!("- hash: \"{}\"\n", grain.hash));
yaml.push_str(&format!(" grain_type: \"{}\"\n", grain.grain_type));
if let serde_json::Value::Object(map) = &grain.fields {
yaml.push_str(" fields:\n");
for (k, v) in map {
yaml.push_str(&format!(" {}: {}\n", k, v));
}
}
if i < grains.len() - 1 {
yaml.push('\n');
}
}
(yaml, "yaml")
}
super::ast::FormatSpec::Text => {
let mut text = String::new();
if let Some(fields) = grouped_by {
let groups = collect_groups(grains, fields);
let total_groups = groups.len();
for (idx, (key, members)) in groups.iter().enumerate() {
text.push_str(&format!(
"--- Group {}/{}: {} ({} {}) ---\n",
idx + 1,
total_groups,
key,
members.len(),
if members.len() == 1 {
"memory"
} else {
"memories"
}
));
for (i, grain) in members.iter().enumerate() {
text.push_str(&crate::render::render_grain_text_line(
&grain_view(grain),
Some(i + 1),
));
text.push('\n');
}
if idx + 1 < total_groups {
text.push('\n');
}
}
} else {
for grain in grains {
text.push_str(&crate::render::render_grain_text_line(
&grain_view(grain),
None,
));
text.push('\n');
}
}
(text, "text")
}
super::ast::FormatSpec::Sml => {
let level = crate::render::MetadataDetail::Minimal;
let mut sml = String::from("<grains>\n");
if let Some(fields) = grouped_by {
let groups = collect_groups(grains, fields);
for (key, members) in &groups {
let escaped_key = crate::render::sml_escape(key);
sml.push_str(&format!(
" <group key=\"{}\" count=\"{}\">\n",
escaped_key,
members.len()
));
for grain in members {
sml.push_str(" ");
sml.push_str(&crate::render::render_grain_sml_at(
&grain_view(grain),
level,
disclosure,
));
sml.push('\n');
}
sml.push_str(" </group>\n");
}
} else {
for grain in grains {
sml.push_str(" ");
sml.push_str(&crate::render::render_grain_sml_at(
&grain_view(grain),
level,
disclosure,
));
sml.push('\n');
}
}
sml.push_str("</grains>");
(sml, "sml")
}
super::ast::FormatSpec::Toon => {
let views: Vec<crate::render::GrainView<'_>> = grains.iter().map(grain_view).collect();
(crate::render::render_toon(&views), "toon")
}
super::ast::FormatSpec::Triples => {
let mut triples = String::new();
for grain in grains {
if let serde_json::Value::Object(map) = &grain.fields {
let s = map.get("subject").and_then(|v| v.as_str()).unwrap_or("_");
let r = map.get("relation").and_then(|v| v.as_str()).unwrap_or("_");
let o = map.get("object").and_then(|v| v.as_str()).unwrap_or("_");
triples.push_str(&format!("{}\t{}\t{}\n", s, r, o));
}
}
(triples, "triples")
}
super::ast::FormatSpec::Csv => {
let mut csv = String::from("hash,grain_type,subject,relation,object,confidence\n");
for grain in grains {
let get_field = |f: &str| -> String {
json_field(&grain.fields, f)
.map(|v| match v {
serde_json::Value::String(s) => {
if s.contains(',') || s.contains('"') || s.contains('\n') {
format!("\"{}\"", s.replace('"', "\"\""))
} else {
s.clone()
}
}
_ => v.to_string(),
})
.unwrap_or_default()
};
csv.push_str(&format!(
"{},{},{},{},{},{}\n",
grain.hash,
grain.grain_type,
get_field("subject"),
get_field("relation"),
get_field("object"),
get_field("confidence"),
));
}
(csv, "csv")
}
super::ast::FormatSpec::Table => {
let header = "| hash | grain_type | subject | relation | object | confidence |";
let separator = "| --- | --- | --- | --- | --- | --- |";
let mut table = format!("{}\n{}\n", header, separator);
for grain in grains {
let get_field = |f: &str| -> String {
json_field(&grain.fields, f)
.map(|v| match v {
serde_json::Value::String(s) => s.replace('|', "\\|"),
_ => v.to_string(),
})
.unwrap_or_default()
};
table.push_str(&format!(
"| {} | {} | {} | {} | {} | {} |\n",
&grain.hash[..8.min(grain.hash.len())],
grain.grain_type,
get_field("subject"),
get_field("relation"),
get_field("object"),
get_field("confidence"),
));
}
(table, "table")
}
super::ast::FormatSpec::Preset { name } | super::ast::FormatSpec::TemplateRef { name } => {
let info = store
.get_template(name)
.ok_or_else(|| CalError::TemplateNotFound {
name: name.clone(),
span: None,
})?;
let mut parsed = super::templates::parse_template_any(&info.source)?;
let parent_name = info.parent.clone().or_else(|| {
(parsed.is_sectioned() && !info.builtin).then(|| "readable".to_string())
});
if let Some(parent_name) = parent_name {
if let Some(parent_info) = store.get_template(&parent_name) {
let parent = super::templates::parse_template_any(&parent_info.source)?;
parsed = super::templates::merge_templates(&parent, &parsed);
}
}
let user_vars_map: std::collections::HashMap<String, String> = user_vars
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let ctx = super::templates::RenderContext {
now_secs: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0),
tier: template_tier(plan, grains.len()),
total_count: grains.len(),
user_vars: user_vars_map,
};
let rendered = match parsed.sections() {
Some(sections) => super::templates::render_sectioned(
sections,
plan.sources
.unwrap_or(&[super::templates::RenderSource::single(grains)]),
&ctx,
plan.assembly,
super::templates::MAX_RENDER_OUTPUT_SIZE,
)?,
None => {
if let Some((rendered, total)) =
super::templates::each_iteration_cap(&parsed, grains.len())
{
warnings.push(
super::errors::CalWarning::EachIterationCapped {
rendered,
total,
max: super::templates::MAX_EACH_ITERATIONS,
}
.to_string(),
);
}
super::templates::render(&parsed, grains, &ctx)?
}
};
store.record_template_run(name);
let key = match format {
super::ast::FormatSpec::Preset { .. } => "preset",
_ => "template",
};
(rendered, key)
}
super::ast::FormatSpec::Template { template } => {
let user_vars_map: std::collections::HashMap<String, String> = user_vars
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let ctx = super::templates::RenderContext {
now_secs: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0),
tier: template_tier(plan, grains.len()),
total_count: grains.len(),
user_vars: user_vars_map,
};
let sections = super::templates::TemplateSections {
element: Some(super::templates::parse_section(template)?),
..Default::default()
};
let rendered = super::templates::render_sectioned(
§ions,
plan.sources
.unwrap_or(&[super::templates::RenderSource::single(grains)]),
&ctx,
plan.assembly,
super::templates::MAX_RENDER_OUTPUT_SIZE,
)?;
(rendered, "template")
}
super::ast::FormatSpec::TemplateInline { sections } => {
let user_vars_map: std::collections::HashMap<String, String> = user_vars
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let ctx = super::templates::RenderContext {
now_secs: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0),
tier: template_tier(plan, grains.len()),
total_count: grains.len(),
user_vars: user_vars_map,
};
let parsed = super::templates::parse_sections(sections)?;
let rendered = super::templates::render_sectioned(
&parsed,
plan.sources
.unwrap_or(&[super::templates::RenderSource::single(grains)]),
&ctx,
plan.assembly,
super::templates::MAX_RENDER_OUTPUT_SIZE,
)?;
(rendered, "template")
}
};
Ok(CalResultPayload::Formatted {
text,
format: format_name.to_string(),
grain_count: grains.len(),
grains: grains.to_vec(),
})
}
pub const GROUP_KEY_JOINER: &str = " · ";
fn group_grains_by_field(grains: Vec<CalGrainResult>, fields: &[String]) -> Vec<CalGrainResult> {
let mut groups: BTreeMap<String, Vec<CalGrainResult>> = BTreeMap::new();
for grain in grains {
let key = group_key_of(&grain, fields);
groups.entry(key).or_default().push(grain);
}
for members in groups.values_mut() {
members.sort_by(|a, b| {
let ta = json_field(&a.fields, "created_at_sec").and_then(|v| v.as_f64());
let tb = json_field(&b.fields, "created_at_sec").and_then(|v| v.as_f64());
ta.partial_cmp(&tb).unwrap_or(Ordering::Equal)
});
}
let mut group_vec: Vec<(String, Vec<CalGrainResult>)> = groups.into_iter().collect();
group_vec.sort_by(|a, b| {
let earliest_a =
a.1.first()
.and_then(|g| json_field(&g.fields, "created_at_sec").and_then(|v| v.as_f64()));
let earliest_b =
b.1.first()
.and_then(|g| json_field(&g.fields, "created_at_sec").and_then(|v| v.as_f64()));
earliest_a
.partial_cmp(&earliest_b)
.unwrap_or(Ordering::Equal)
});
if let Some(pos) = group_vec.iter().position(|(k, _)| k.is_empty()) {
let empty = group_vec.remove(pos);
group_vec.push(empty);
}
group_vec
.into_iter()
.flat_map(|(_, members)| members)
.collect()
}
fn count_by_fields(grains: &[CalGrainResult], fields: &[String]) -> Vec<CalGrainResult> {
let mut rows: Vec<(String, Vec<String>, usize)> = collect_groups(grains, fields)
.into_iter()
.map(|(key, members)| (key, group_key_parts(members[0], fields), members.len()))
.collect();
rows.sort_by(|a, b| b.2.cmp(&a.2).then_with(|| a.0.cmp(&b.0)));
rows.into_iter()
.map(|(key, parts, count)| {
let mut row = serde_json::json!({ "key": key, "count": count });
if fields.len() > 1 {
row["keys"] = serde_json::json!(parts);
}
CalGrainResult {
hash: String::new(),
grain_type: "group".to_string(),
score: 0.0,
fields: row,
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
}
})
.collect()
}
fn group_key_of(grain: &CalGrainResult, fields: &[String]) -> String {
group_key_parts(grain, fields).join(GROUP_KEY_JOINER)
}
fn group_key_parts(grain: &CalGrainResult, fields: &[String]) -> Vec<String> {
fields
.iter()
.map(|f| {
json_field(&grain.fields, f)
.map(|v| match v {
serde_json::Value::String(s) => s.clone(),
_ => v.to_string(),
})
.unwrap_or_default()
})
.collect()
}
fn collect_groups<'a>(
grains: &'a [CalGrainResult],
fields: &[String],
) -> Vec<(String, Vec<&'a CalGrainResult>)> {
let mut groups: Vec<(String, Vec<&CalGrainResult>)> = Vec::new();
for grain in grains {
let key = group_key_of(grain, fields);
if let Some(last) = groups.last_mut() {
if last.0 == key {
last.1.push(grain);
continue;
}
}
groups.push((key, vec![grain]));
}
groups
}
const COMMON_FIELDS: &[&str] = &[
"subject",
"relation",
"object",
"namespace",
"user_id",
"confidence",
"importance",
"score",
"tags",
"type",
"time",
"hash",
"contradicted",
"verification_status",
"source_type",
"recall_priority",
"epistemic_status",
"query",
"created_at",
"content",
"summary",
"grain_type",
"scope",
"scope_path",
"priority",
"status",
"valid_from",
"valid_to",
"system_valid_from",
"system_valid_to",
];
fn type_specific_fields(grain_type: &GrainTypePlural) -> &'static [&'static str] {
match grain_type.to_grain_type() {
Some(ty) => areev_core::types::registry::meta(ty).queryable_fields,
None => &[],
}
}
fn is_known_type_specific_field(field: &str) -> bool {
areev_core::types::registry::GRAIN_TYPES
.iter()
.any(|m| m.queryable_fields.contains(&field))
}
fn suggest_field(field: &str, grain_type: &GrainTypePlural) -> Option<String> {
let valid_fields = type_specific_fields(grain_type);
for known in valid_fields {
if known.contains(field) || field.contains(known) {
return Some(known.to_string());
}
}
if is_known_type_specific_field(field) {
return Some(format!(
"'{}' exists on a different grain type, not on {}",
field,
grain_type.as_str()
));
}
None
}
const ENGINE_ONLY_FIELDS: &[&str] = &[
"query",
"time",
"entity",
"contradicted",
"scope",
"scope_path",
"tags",
];
const GRAIN_EVALUABLE_COMMON: &[&str] = &[
"subject",
"relation",
"object",
"namespace",
"user_id",
"confidence",
"importance",
"created_at",
"verification_status",
"source_type",
"recall_priority",
"epistemic_status",
"content",
"summary",
"hash",
"grain_type",
"type",
"score",
"valid_from",
"valid_to",
"system_valid_from",
"system_valid_to",
];
fn leaf_pushdown_consumed(condition: &Condition) -> bool {
match condition {
Condition::Comparison {
field, comparator, ..
} => matches!(
(field.as_str(), comparator),
(
"subject" | "relation" | "object" | "namespace" | "user_id" | "query" | "time"
| "entity" | "scope" | "scope_path",
Comparator::Eq
) | ("confidence" | "importance", Comparator::Gte | Comparator::Gt)
| ("contradicted", Comparator::Eq)
),
Condition::In { field, .. } => matches!(
field.as_str(),
"subject" | "relation" | "object" | "tags" | "namespace"
),
Condition::NotIn { field, .. } => field == "tags",
Condition::Contains { field, .. } => {
matches!(field.as_str(), "subject" | "object" | "content" | "summary")
}
Condition::IsCategory { .. } => true,
_ => false,
}
}
fn plan_residual_where(
condition: &Condition,
grain_type: &GrainTypePlural,
warnings: &mut Vec<String>,
) -> std::result::Result<Option<Condition>, CalError> {
match condition {
Condition::And { left, right, span } => {
let l = plan_residual_where(left, grain_type, warnings)?;
let r = plan_residual_where(right, grain_type, warnings)?;
Ok(match (l, r) {
(None, None) => None,
(Some(c), None) | (None, Some(c)) => Some(c),
(Some(a), Some(b)) => Some(Condition::And {
left: Box::new(a),
right: Box::new(b),
span: *span,
}),
})
}
Condition::Or { left, right, .. } => {
validate_residual_subtree(left, grain_type, "under NOT/OR", warnings)?;
validate_residual_subtree(right, grain_type, "under NOT/OR", warnings)?;
Ok(Some(condition.clone()))
}
Condition::Not { inner, .. } => {
validate_residual_subtree(inner, grain_type, "under NOT/OR", warnings)?;
Ok(Some(condition.clone()))
}
leaf => {
if leaf_pushdown_consumed(leaf) {
Ok(None)
} else {
validate_residual_leaf(leaf, grain_type, None, warnings)?;
Ok(Some(leaf.clone()))
}
}
}
}
fn validate_residual_subtree(
condition: &Condition,
grain_type: &GrainTypePlural,
context: &str,
warnings: &mut Vec<String>,
) -> std::result::Result<(), CalError> {
match condition {
Condition::And { left, right, .. } | Condition::Or { left, right, .. } => {
validate_residual_subtree(left, grain_type, context, warnings)?;
validate_residual_subtree(right, grain_type, context, warnings)
}
Condition::Not { inner, .. } => {
validate_residual_subtree(inner, grain_type, context, warnings)
}
leaf => validate_residual_leaf(leaf, grain_type, Some(context), warnings),
}
}
fn validate_residual_leaf(
leaf: &Condition,
grain_type: &GrainTypePlural,
forced_context: Option<&str>,
warnings: &mut Vec<String>,
) -> std::result::Result<(), CalError> {
let (field, span, own_context): (&str, Option<Span>, String) = match leaf {
Condition::Comparison {
field,
comparator,
span,
..
} => (field, *span, format!("with comparator {comparator}")),
Condition::In { field, span, .. } => (field, *span, "with IN".to_string()),
Condition::NotIn { field, span, .. } => (field, *span, "with NOT IN".to_string()),
Condition::Contains { field, span, .. } => (field, *span, "with CONTAINS".to_string()),
Condition::StartsWith { field, span, .. } => {
(field, *span, "with STARTS WITH".to_string())
}
Condition::IsNull { field, span, .. } | Condition::IsNotNull { field, span, .. } => {
(field, *span, "with IS [NOT] NULL".to_string())
}
Condition::IsCategory { field, span, .. } => (field, *span, "with IS".to_string()),
_ => return Ok(()),
};
if field.contains(':') {
return Ok(());
}
let (field, path_depth) = match field.split_once('.') {
Some((base, rest)) => (base, 1 + rest.split('.').count()),
None => (field, 1),
};
if path_depth > MAX_FIELD_PATH_DEPTH {
return Err(CalError::FieldNotOnGrainType {
field: field.to_string(),
grain_type: grain_type.as_str().to_string(),
span,
suggestion: Some(format!(
"a field path may be at most {MAX_FIELD_PATH_DEPTH} segments deep"
)),
});
}
if ENGINE_ONLY_FIELDS.contains(&field) {
return Err(CalError::EngineFieldNotFilterable {
field: field.to_string(),
context: forced_context.map(str::to_string).unwrap_or(own_context),
span,
});
}
if GRAIN_EVALUABLE_COMMON.contains(&field) {
return Ok(());
}
if grain_type.to_grain_type().is_some() {
if type_specific_fields(grain_type).contains(&field) {
return Ok(());
}
let suggestion = suggest_field(field, grain_type);
return Err(CalError::FieldNotOnGrainType {
field: field.to_string(),
grain_type: grain_type.as_str().to_string(),
span,
suggestion,
});
}
if !is_known_type_specific_field(field) {
warnings.push(
super::errors::CalWarning::UnrecognizedWhereField {
field: field.to_string(),
span,
}
.to_string(),
);
}
Ok(())
}
fn resolve_grain_field(grain: &CalGrainResult, field: &str) -> Option<serde_json::Value> {
match field {
"hash" => return Some(serde_json::Value::String(grain.hash.clone())),
"grain_type" | "type" => {
return Some(serde_json::Value::String(grain.grain_type.clone()))
}
"score" => return serde_json::Number::from_f64(grain.score).map(serde_json::Value::Number),
"kind" if grain.grain_type == "tool" && json_field(&grain.fields, "kind").is_none() => {
return Some(serde_json::Value::String("execution".into()))
}
"status" if grain.grain_type == "tool" && json_field(&grain.fields, "status").is_none() => {
return Some(serde_json::Value::String("completed".into()))
}
_ => {}
}
if let Some((base, path)) = field.split_once('.') {
let root = json_field(&grain.fields, base)?;
let parsed;
let root = match root {
serde_json::Value::String(text) => {
parsed = serde_json::from_str::<serde_json::Value>(text).ok()?;
&parsed
}
other => other,
};
return navigate_json_path(root, path).cloned();
}
json_field(&grain.fields, field).cloned()
}
const MAX_FIELD_PATH_DEPTH: usize = 8;
fn navigate_json_path<'a>(
root: &'a serde_json::Value,
path: &str,
) -> Option<&'a serde_json::Value> {
let mut cur = root;
for (depth, seg) in path.split('.').enumerate() {
if depth >= MAX_FIELD_PATH_DEPTH {
return None;
}
cur = match cur {
serde_json::Value::Object(m) => m.get(seg)?,
serde_json::Value::Array(a) => a.get(seg.parse::<usize>().ok()?)?,
_ => return None,
};
}
Some(cur)
}
fn grain_carries_field(grain: &CalGrainResult, field: &str) -> bool {
!matches!(
resolve_grain_field(grain, field),
None | Some(serde_json::Value::Null)
)
}
pub fn grain_matches_condition(
grain: &CalGrainResult,
field: &str,
comparator: &Comparator,
value: &Value,
) -> bool {
let resolved = resolve_grain_field(grain, field);
let grain_value = resolved.as_ref();
let value = &match (field, value) {
("hash", Value::String { value: v }) => Value::String {
value: v.strip_prefix("sha256:").unwrap_or(v).to_string(),
},
_ => value.clone(),
};
match comparator {
Comparator::Eq => match value {
Value::String { value: target } => grain_value
.and_then(|v| v.as_str())
.map(|s| s == target.as_str())
.unwrap_or(false),
Value::Number { value: target } => grain_value
.and_then(|v| v.as_f64())
.map(|n| (n - target).abs() < f64::EPSILON)
.unwrap_or(false),
Value::Boolean { value: target } => grain_value
.and_then(|v| v.as_bool())
.map(|b| b == *target)
.unwrap_or(false),
_ => false,
},
Comparator::NotEq => {
grain_carries_field(grain, field)
&& !grain_matches_condition(grain, field, &Comparator::Eq, value)
}
Comparator::Gte => match value {
Value::Number { value: target } => grain_value
.and_then(|v| v.as_f64())
.map(|n| n >= *target)
.unwrap_or(false),
_ => false,
},
Comparator::Gt => match value {
Value::Number { value: target } => grain_value
.and_then(|v| v.as_f64())
.map(|n| n > *target)
.unwrap_or(false),
_ => false,
},
Comparator::Lte => match value {
Value::Number { value: target } => grain_value
.and_then(|v| v.as_f64())
.map(|n| n <= *target)
.unwrap_or(false),
_ => false,
},
Comparator::Lt => match value {
Value::Number { value: target } => grain_value
.and_then(|v| v.as_f64())
.map(|n| n < *target)
.unwrap_or(false),
_ => false,
},
}
}
pub fn grain_matches_condition_tree(grain: &CalGrainResult, condition: &Condition) -> bool {
grain_condition_truth(grain, condition).unwrap_or(false)
}
fn grain_condition_truth(grain: &CalGrainResult, condition: &Condition) -> Option<bool> {
match condition {
Condition::Comparison {
field,
comparator,
value,
..
} => {
if !grain_carries_field(grain, field) {
return None;
}
Some(grain_matches_condition(grain, field, comparator, value))
}
Condition::And { left, right, .. } => {
match (
grain_condition_truth(grain, left),
grain_condition_truth(grain, right),
) {
(Some(false), _) | (_, Some(false)) => Some(false),
(Some(true), Some(true)) => Some(true),
_ => None,
}
}
Condition::Or { left, right, .. } => {
match (
grain_condition_truth(grain, left),
grain_condition_truth(grain, right),
) {
(Some(true), _) | (_, Some(true)) => Some(true),
(Some(false), Some(false)) => Some(false),
_ => None,
}
}
Condition::Not { inner, .. } => grain_condition_truth(grain, inner).map(|b| !b),
Condition::In { field, values, .. } => {
if !grain_carries_field(grain, field) {
return None;
}
Some(
values
.iter()
.any(|v| grain_matches_condition(grain, field, &Comparator::Eq, v)),
)
}
Condition::NotIn { field, values, .. } => {
if !grain_carries_field(grain, field) {
return None;
}
Some(
!values
.iter()
.any(|v| grain_matches_condition(grain, field, &Comparator::Eq, v)),
)
}
Condition::IsNull { field, .. } => Some(!grain_carries_field(grain, field)),
Condition::IsNotNull { field, .. } => Some(grain_carries_field(grain, field)),
Condition::Contains { field, value, .. } => {
let v = resolve_grain_field(grain, field)?;
Some(
v.as_str()
.map(|s| s.contains(value.as_str()))
.unwrap_or(false),
)
}
Condition::StartsWith { field, value, .. } => {
let v = resolve_grain_field(grain, field)?;
Some(
v.as_str()
.map(|s| s.starts_with(value.as_str()))
.unwrap_or(false),
)
}
Condition::IsCategory {
field, category, ..
} => {
let v = resolve_grain_field(grain, field)?;
Some(
v.as_str()
.map(|s| s.eq_ignore_ascii_case(category))
.unwrap_or(false),
)
}
}
}
fn collect_filter_names(condition: &Condition, names: &mut Vec<String>) {
match condition {
Condition::Comparison { field, .. } => {
if !names.contains(field) {
names.push(field.clone());
}
}
Condition::In { field, .. } | Condition::NotIn { field, .. } => {
if !names.contains(field) {
names.push(field.clone());
}
}
Condition::IsNull { field, .. } | Condition::IsNotNull { field, .. } => {
if !names.contains(field) {
names.push(field.clone());
}
}
Condition::Contains { field, .. } | Condition::StartsWith { field, .. } => {
if !names.contains(field) {
names.push(field.clone());
}
}
Condition::IsCategory { field, .. } => {
if !names.contains(field) {
names.push(field.clone());
}
}
Condition::And { left, right, .. } | Condition::Or { left, right, .. } => {
collect_filter_names(left, names);
collect_filter_names(right, names);
}
Condition::Not { inner, .. } => {
collect_filter_names(inner, names);
}
}
}
fn value_dedup_key(v: &serde_json::Value) -> String {
match v {
serde_json::Value::String(s) => s.clone(),
other => other.to_string(),
}
}
fn json_field<'a>(fields: &'a serde_json::Value, field: &str) -> Option<&'a serde_json::Value> {
if let serde_json::Value::Object(map) = fields {
map.get(field)
} else {
None
}
}
fn project_fields(fields: &serde_json::Value, selected: &[String]) -> serde_json::Value {
let mut out = serde_json::Map::new();
if let serde_json::Value::Object(map) = fields {
for key in selected {
if let Some(v) = map.get(key) {
out.insert(key.clone(), v.clone());
}
}
}
serde_json::Value::Object(out)
}
fn compare_json_values(a: Option<&serde_json::Value>, b: Option<&serde_json::Value>) -> Ordering {
match (a, b) {
(None, None) => Ordering::Equal,
(None, Some(_)) => Ordering::Less,
(Some(_), None) => Ordering::Greater,
(Some(va), Some(vb)) => {
if let (Some(fa), Some(fb)) = (va.as_f64(), vb.as_f64()) {
fa.partial_cmp(&fb).unwrap_or(Ordering::Equal)
} else if let (Some(sa), Some(sb)) = (va.as_str(), vb.as_str()) {
sa.cmp(sb)
} else if let (Some(ba), Some(bb)) = (va.as_bool(), vb.as_bool()) {
ba.cmp(&bb)
} else {
va.to_string().cmp(&vb.to_string())
}
}
}
}
fn union_grains(mut left: Vec<CalGrainResult>, right: Vec<CalGrainResult>) -> Vec<CalGrainResult> {
let left_hashes: std::collections::HashSet<String> =
left.iter().map(|g| g.hash.clone()).collect();
for g in right {
if !left_hashes.contains(&g.hash) {
left.push(g);
}
}
left
}
fn intersect_grains(left: Vec<CalGrainResult>, right: &[CalGrainResult]) -> Vec<CalGrainResult> {
let right_hashes: std::collections::HashSet<&str> =
right.iter().map(|g| g.hash.as_str()).collect();
left.into_iter()
.filter(|g| right_hashes.contains(g.hash.as_str()))
.collect()
}
fn except_grains(left: Vec<CalGrainResult>, right: &[CalGrainResult]) -> Vec<CalGrainResult> {
let right_hashes: std::collections::HashSet<&str> =
right.iter().map(|g| g.hash.as_str()).collect();
left.into_iter()
.filter(|g| !right_hashes.contains(g.hash.as_str()))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::facade::CalStoreFacade;
use crate::store_types::{RecallParams, SearchHit};
use crate::store_types::VersionEntry;
use areev_core::error::{AreevError, Hash};
use areev_core::format::deserialize::DeserializedGrain;
use areev_core::format::header::MgHeader;
use areev_core::types::GrainType;
use std::collections::HashMap;
#[test]
fn authz_store_code_reads_through_a_host_wrap() {
assert_eq!(
authz_store_code(&AreevError::AuthzUnknownPrincipal("p".into())),
Some("AUT-E002")
);
let wrapped = AreevError::Validation("LOP-E010: AUT-E001: authorization denied".into());
assert_eq!(authz_store_code(&wrapped), Some("AUT-E001"));
}
struct MockStore {
grains: Vec<(Hash, DeserializedGrain)>,
}
impl MockStore {
fn empty() -> Self {
Self { grains: Vec::new() }
}
fn with_grains(grains: Vec<(Hash, DeserializedGrain)>) -> Self {
Self { grains }
}
}
fn make_fact(subject: &str, relation: &str, object: &str) -> (Hash, DeserializedGrain) {
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("subject".into(), serde_json::json!(subject));
fields.insert("relation".into(), serde_json::json!(relation));
fields.insert("object".into(), serde_json::json!(object));
fields.insert("grain_type".into(), serde_json::json!("fact"));
fields.insert("confidence".into(), serde_json::json!(0.9));
let mut hash_bytes = [0u8; 32];
let key = format!("{}|{}|{}", subject, relation, object);
for (i, b) in key.as_bytes().iter().enumerate().take(32) {
hash_bytes[i] = *b;
}
let hash = Hash::from_bytes(&hash_bytes);
let grain = DeserializedGrain {
header: MgHeader {
version: 1,
flags: 0,
grain_type: 0x01,
ns_hash: 0,
created_at_sec: 0,
},
grain_type: GrainType::Fact,
fields,
hash,
};
(hash, grain)
}
impl CalStoreFacade for MockStore {
fn recall(&self, params: &RecallParams) -> areev_core::error::Result<Vec<SearchHit>> {
let mut hits: Vec<SearchHit> = self
.grains
.iter()
.filter(|(_, g)| {
if let Some(ref s) = params.subject {
if g.get_str("subject") != Some(s.as_str()) {
return false;
}
}
if let Some(ref r) = params.relation {
if g.get_str("relation") != Some(r.as_str()) {
return false;
}
}
true
})
.map(|(hash, grain)| SearchHit {
grain: grain.clone(),
score: 1.0,
hash: *hash,
score_breakdown: None,
explanation: None,
scope_depth: None,
source_namespace: None,
#[cfg(feature = "rerank")]
rerank_score: None,
#[cfg(feature = "llm-rerank")]
llm_rerank_score: None,
relative_time: None,
conflict_status: None,
supersession_status: None,
superseded_by_hash: None,
recall_source: None,
})
.collect();
if let Some(limit) = params.limit {
hits.truncate(limit);
}
Ok(hits)
}
fn exists(&self, hash: &Hash) -> areev_core::error::Result<bool> {
Ok(self.grains.iter().any(|(h, _)| h == hash))
}
fn get(&self, hash: &Hash) -> areev_core::error::Result<DeserializedGrain> {
self.grains
.iter()
.find(|(h, _)| h == hash)
.map(|(_, g)| g.clone())
.ok_or(AreevError::NotFound(*hash))
}
fn count(&self) -> areev_core::error::Result<usize> {
Ok(self.grains.len())
}
fn get_history(
&self,
_ns: &str,
_s: &str,
_r: &str,
) -> areev_core::error::Result<Vec<VersionEntry>> {
Ok(Vec::new())
}
fn default_namespace(&self) -> Option<&str> {
None
}
fn active_user(&self) -> Option<&str> {
None
}
fn cal_add(
&self,
_grain_type: &str,
_fields: &serde_json::Map<String, serde_json::Value>,
) -> areev_core::error::Result<Hash> {
Err(AreevError::Validation(
"mock: cal_add not implemented".into(),
))
}
fn cal_supersede(
&self,
_old_hash: &Hash,
_grain_type: &str,
_fields: &serde_json::Map<String, serde_json::Value>,
) -> areev_core::error::Result<Hash> {
Err(AreevError::Validation(
"mock: cal_supersede not implemented".into(),
))
}
fn list_templates(&self) -> Vec<crate::facade::TemplateInfo> {
let registry = crate::templates::TemplateRegistry::new();
registry.list()
}
}
fn exec() -> CalExecutor {
CalExecutor::with_defaults()
}
#[test]
fn test_execute_recall_empty_store() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("RECALL facts", &store).unwrap();
assert_eq!(result.metadata.statement_type, "recall");
match result.result {
CalResultPayload::Grains { grains, .. } => assert!(grains.is_empty()),
other => panic!("unexpected payload: {:?}", other),
}
}
#[test]
fn test_execute_recall_where_subject() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let (hash_b, grain_b) = make_fact("bob", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_a, grain_a), (hash_b, grain_b)]);
let ex = exec();
let result = ex
.execute(r#"RECALL facts WHERE subject = "john""#, &store)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].hash, hash_a.to_hex());
}
other => panic!("unexpected payload: {:?}", other),
}
}
#[test]
fn test_execute_recall_about_clause() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(r#"RECALL facts ABOUT "coffee preferences""#, &store)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
#[test]
fn test_execute_recall_pipeline_limit() {
let grains: Vec<_> = (0..10u8)
.map(|i| {
let key = format!("user{}", i);
make_fact(&key, "likes", "coffee")
})
.collect();
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex.execute("RECALL facts LIMIT 3", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => assert_eq!(grains.len(), 3),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_recall_pipeline_count() {
let grains: Vec<_> = (0..5u8)
.map(|i| make_fact(&format!("u{}", i), "likes", "tea"))
.collect();
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex.execute("RECALL facts COUNT", &store).unwrap();
match result.result {
CalResultPayload::Count { count } => assert_eq!(count, 5),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_recall_pipeline_order_by() {
let mut grains = vec![
make_fact("charlie", "likes", "rust"),
make_fact("john", "likes", "python"),
make_fact("bob", "likes", "go"),
];
for (i, (_, g)) in grains.iter_mut().enumerate() {
g.fields
.insert("created_at".into(), serde_json::json!(i as i64));
}
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex.execute("RECALL facts ORDER BY subject", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains[0].hash, make_fact("bob", "likes", "go").0.to_hex());
assert_eq!(
grains[1].hash,
make_fact("charlie", "likes", "rust").0.to_hex()
);
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_recall_pipeline_select() {
let (hash, _) = make_fact("john", "likes", "vim");
let store = MockStore::with_grains(vec![make_fact("john", "likes", "vim")]);
let ex = exec();
let result = ex
.execute("RECALL facts SELECT subject, object", &store)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
if let serde_json::Value::Object(map) = &grains[0].fields {
assert!(map.contains_key("subject"));
assert!(map.contains_key("object"));
assert!(!map.contains_key("relation"));
} else {
panic!("expected Object fields");
}
let _ = hash;
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_exists_known_hash() {
let (hash, grain) = make_fact("john", "is", "a developer");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let hex = hash.to_hex();
let query = format!("EXISTS sha256:{}", hex);
let result = ex.execute(&query, &store).unwrap();
match result.result {
CalResultPayload::Exists { exists, .. } => assert!(exists),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_exists_unknown_hash() {
let store = MockStore::empty();
let ex = exec();
let hex = "a".repeat(64);
let query = format!("EXISTS sha256:{}", hex);
let result = ex.execute(&query, &store).unwrap();
match result.result {
CalResultPayload::Exists { exists, .. } => assert!(!exists),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_describe_schema() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE SCHEMA", &store).unwrap();
assert_eq!(result.metadata.statement_type, "describe");
match result.result {
CalResultPayload::Describe { info } => {
assert!(info.get("grain_types").is_some());
assert!(info.get("common_fields").is_some());
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_explain_recall() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(r#"EXPLAIN RECALL facts ABOUT "preferences""#, &store)
.unwrap();
assert_eq!(result.metadata.statement_type, "explain");
match result.result {
CalResultPayload::Explain { plan } => {
assert_eq!(plan.statement_type, "recall");
assert!(!plan.index_usage.is_empty());
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_explain_reports_like_as_the_bm25_leg_it_runs() {
let store = MockStore::empty();
let ex = exec();
let plan_of = |q: &str| match ex.execute(q, &store).unwrap().result {
CalResultPayload::Explain { plan } => plan,
other => panic!("unexpected: {:?}", other),
};
let like = plan_of(r#"EXPLAIN RECALL facts LIKE "window""#);
assert_eq!(like.query_routing, "bm25");
assert!(like.index_usage.contains(&"bm25_fts".to_string()));
assert!(
like.estimated_cost.starts_with("O(log n)"),
"LIKE must not be reported as a full scan, got: {}",
like.estimated_cost
);
let about = plan_of(r#"EXPLAIN RECALL facts ABOUT "window""#);
assert_eq!(like.query_routing, about.query_routing);
assert_eq!(like.index_usage, about.index_usage);
let anchored = plan_of(r#"EXPLAIN RECALL facts LIKE "window" WHERE subject = "john""#);
assert_eq!(anchored.query_routing, "hybrid_rrf");
}
#[test]
fn test_execute_batch_two_queries() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute("BATCH { RECALL facts ; RECALL events }", &store)
.unwrap();
assert_eq!(result.metadata.statement_type, "batch");
match result.result {
CalResultPayload::Batch { results } => {
assert_eq!(results.len(), 2);
assert!(results.contains_key("0"));
assert!(results.contains_key("1"));
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_coalesce() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex
.execute(
r#"COALESCE(RECALL facts WHERE subject = "john", RECALL facts WHERE subject = "bob")"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => assert_eq!(grains.len(), 1),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_coalesce_fallback_to_second_branch() {
let (hash, grain) = make_fact("bob", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex
.execute(
r#"COALESCE(RECALL facts WHERE subject = "john", RECALL facts WHERE subject = "bob")"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1, "should fall back to second branch");
assert_eq!(
grains[0].fields.get("subject").and_then(|v| v.as_str()),
Some("bob"),
"result should be from the second branch (bob)"
);
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_execute_coalesce_all_branches_empty() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(
r#"COALESCE(RECALL facts WHERE subject = "john", RECALL facts WHERE subject = "bob")"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert!(grains.is_empty(), "all branches empty → empty result");
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_execute_batch_five_queries() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex
.execute(
r#"BATCH { RECALL facts ; RECALL events ; RECALL tools ; RECALL goals ; RECALL facts WHERE subject = "john" }"#,
&store,
)
.unwrap();
assert_eq!(result.metadata.statement_type, "batch");
match result.result {
CalResultPayload::Batch { results } => {
assert_eq!(results.len(), 5, "BATCH should execute all 5 sub-queries");
assert!(results.contains_key("0"));
assert!(results.contains_key("4"));
}
other => panic!("expected Batch, got: {:?}", other),
}
}
#[test]
fn test_execute_tier1_add_returns_e044() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
tier1_enabled: false,
..Default::default()
});
let err = ex
.execute(
r#"ADD fact SET subject = "john" SET relation = "likes" SET object = "rust" REASON "test""#,
&store,
)
.expect_err("Tier 1 disabled must error, not Unsupported-payload");
assert_eq!(err.code(), "CAL-E044");
assert!(err.to_string().contains("ADD"));
}
#[test]
fn test_execute_generic_add_rejects_unshapeable_type() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(r#"ADD event SET content = "test" REASON "test""#, &store)
.unwrap();
match result.result {
CalResultPayload::Unsupported { message, .. } => {
assert!(
message.contains("cannot be created via ADD"),
"expected grain type restriction message, got: {}",
message
);
assert!(
message.contains("event"),
"message should mention the rejected type, got: {}",
message
);
}
other => panic!("expected Unsupported, got: {:?}", other),
}
}
#[test]
fn test_execute_add_unresolved_parameter() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(
r#"ADD fact SET subject = $user SET relation = "likes" SET object = "rust" REASON "test""#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Unsupported { message, .. } => {
assert!(
message.contains("Unresolved parameter"),
"expected unresolved parameter message, got: {}",
message
);
assert!(
message.contains("$user"),
"message should mention the parameter name, got: {}",
message
);
}
other => panic!("expected Unsupported, got: {:?}", other),
}
}
#[test]
fn test_added_payload_has_extraction_fields() {
let payload = CalResultPayload::Added {
hash: "abc123".into(),
grain_type: "fact".into(),
extracted_count: Some(3),
extraction_warnings: vec!["warn1".into()],
};
match payload {
CalResultPayload::Added {
extracted_count,
extraction_warnings,
..
} => {
assert_eq!(extracted_count, Some(3));
assert_eq!(extraction_warnings.len(), 1);
}
_ => unreachable!(),
}
let payload2 = CalResultPayload::Added {
hash: "def456".into(),
grain_type: "observation".into(),
extracted_count: None,
extraction_warnings: vec![],
};
match payload2 {
CalResultPayload::Added {
extracted_count,
extraction_warnings,
..
} => {
assert_eq!(extracted_count, None);
assert!(extraction_warnings.is_empty());
}
_ => unreachable!(),
}
}
#[test]
fn test_execute_recall_with_score_breakdown() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute("RECALL facts WITH score_breakdown", &store)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
#[test]
fn test_query_hash_is_computed_c4() {
let store = MockStore::empty();
let ex = exec();
let input = "RECALL facts";
let result = ex.execute(input, &store).unwrap();
assert_eq!(result.query_hash.len(), 64);
assert!(result.query_hash.chars().all(|c| c.is_ascii_hexdigit()));
let result2 = ex.execute(input, &store).unwrap();
assert_eq!(result.query_hash, result2.query_hash);
}
#[test]
fn test_namespace_override_applied() {
let store = MockStore::empty();
let config = CalExecutorConfig {
namespace_override: Some("tenant_a".to_string()),
..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex
.execute(r#"RECALL facts WHERE namespace = "other""#, &store)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
#[test]
fn test_compute_query_hash_stable() {
let h1 = compute_query_hash("RECALL facts");
let h2 = compute_query_hash("RECALL facts");
assert_eq!(h1, h2);
}
#[test]
fn test_compute_query_hash_distinct() {
let h1 = compute_query_hash("RECALL facts");
let h2 = compute_query_hash("RECALL events");
assert_ne!(h1, h2);
}
#[test]
fn test_execute_pipeline_offset() {
let grains: Vec<_> = (0..5u8)
.map(|i| make_fact(&format!("u{}", i), "likes", "rust"))
.collect();
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex.execute("RECALL facts OFFSET 3", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 2);
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_pipeline_first() {
let grains: Vec<_> = (0..5u8)
.map(|i| make_fact(&format!("u{}", i), "likes", "rust"))
.collect();
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex.execute("RECALL facts FIRST", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => assert_eq!(grains.len(), 1),
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_pipeline_hashes() {
let (hash, grain) = make_fact("john", "is", "a developer");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex.execute("RECALL facts HASHES", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].grain_type, "extracted");
let value = grains[0].fields.get("value").unwrap();
assert_eq!(value.as_str().unwrap(), hash.to_hex());
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_describe_facts() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE facts", &store).unwrap();
match result.result {
CalResultPayload::Describe { info } => {
assert_eq!(info["grain_type"], "facts");
assert!(info.get("specific_fields").is_some());
}
other => panic!("unexpected: {:?}", other),
}
}
#[test]
fn test_execute_tier1_supersede_returns_e044() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
tier1_enabled: false,
..Default::default()
});
let err = ex
.execute(
r#"SUPERSEDE sha256:abc123def456 SET object = "new value" REASON "update""#,
&store,
)
.expect_err("Tier 1 disabled must error, not Unsupported-payload");
assert_eq!(err.code(), "CAL-E044");
assert!(err.to_string().contains("SUPERSEDE"));
}
#[test]
fn test_execute_tier1_revert_returns_unsupported_s2() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(r#"REVERT sha256:abc123def456 REASON "mistake""#, &store)
.unwrap();
match result.result {
CalResultPayload::Unsupported { message, .. } => {
assert!(
message.contains("Tier 1"),
"REVERT must return Tier 1 unsupported message (S-2), got: {}",
message
);
}
other => panic!("expected Unsupported for REVERT, got: {:?}", other),
}
}
#[test]
fn test_query_hash_format_c4() {
let store = MockStore::empty();
let ex = exec();
let queries = [
"RECALL facts",
"RECALL events",
"EXISTS sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"DESCRIBE SCHEMA",
"BATCH { RECALL facts ; RECALL events }",
];
for q in &queries {
let result = ex.execute(q, &store).unwrap();
assert_eq!(
result.query_hash.len(),
64,
"query_hash must be 64 hex chars for '{}', got {}",
q,
result.query_hash.len()
);
assert!(
result.query_hash.chars().all(|c| c.is_ascii_hexdigit()),
"query_hash must be hex for '{}', got '{}'",
q,
result.query_hash
);
}
}
#[test]
fn test_query_hash_nfc_equivalence_c4_s6() {
let store = MockStore::empty();
let ex = exec();
let decomposed = "RECALL facts WHERE subject = \"caf\u{0065}\u{0301}\"";
let precomposed = "RECALL facts WHERE subject = \"caf\u{00E9}\"";
let r1 = ex.execute(decomposed, &store).unwrap();
let r2 = ex.execute(precomposed, &store).unwrap();
assert_eq!(
r1.query_hash, r2.query_hash,
"NFC-equivalent inputs must produce the same query_hash (C-4 + S-6)"
);
}
#[test]
fn test_or_condition_warning_propagation() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(
r#"RECALL facts WHERE subject = "john" OR subject = "bob""#,
&store,
)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
#[test]
fn test_assemble_where_clause_applied() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(
r#"ASSEMBLE "summary" FROM (RECALL facts) WHERE confidence >= 0.8"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert!(grains.is_empty(), "empty store should return no grains");
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_execute_empty_input_error() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("", &store);
assert!(result.is_err(), "empty input must return an error");
}
#[test]
fn test_execute_whitespace_only_error() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute(" \t\n ", &store);
assert!(
result.is_err(),
"whitespace-only input must return an error"
);
}
#[test]
fn test_execute_query_too_long_error() {
let store = MockStore::empty();
let ex = exec();
let huge = format!("RECALL facts WHERE subject = \"{}\"", "a".repeat(66_000));
let result = ex.execute(&huge, &store);
assert!(result.is_err(), "oversized input must return an error");
}
#[test]
fn test_execute_malformed_inputs_no_panic() {
let store = MockStore::empty();
let ex = exec();
let malformed = [
"RECALL",
"WHERE subject",
"|||",
"RECALL facts |",
"RECALL facts WHERE",
"RECALL facts WHERE subject =",
"RECALL 123",
"; ; ;",
"RECALL facts beliefs beliefs",
"RECALL facts WHERE subject = \"unterminated",
];
for input in &malformed {
let _ = ex.execute(input, &store);
}
}
#[test]
fn test_history_diff_identical_grains() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let history_stmt = super::HistoryStmt {
hash: hash.to_hex(),
where_clause: None,
diff_target: Some(hash.to_hex()),
span: None,
};
let query = super::CalQuery {
version: super::super::ast::CalVersion(1),
statement: super::CalStatement::History(history_stmt),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let payload = ex
.execute_statement(&query.statement, &store, &query, &mut warnings)
.unwrap();
match payload {
CalResultPayload::Diff { changes, .. } => {
assert!(
changes.is_empty(),
"identical grains should have no differences"
);
}
other => panic!("expected Diff, got {:?}", other),
}
assert!(warnings.is_empty(), "no warnings for same subject/relation");
}
#[test]
fn test_history_diff_different_grains() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let (hash_b, grain_b) = make_fact("john", "likes", "tea");
let store = MockStore::with_grains(vec![(hash_a, grain_a), (hash_b, grain_b)]);
let ex = exec();
let history_stmt = super::HistoryStmt {
hash: hash_a.to_hex(),
where_clause: None,
diff_target: Some(hash_b.to_hex()),
span: None,
};
let query = super::CalQuery {
version: super::super::ast::CalVersion(1),
statement: super::CalStatement::History(history_stmt),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let payload = ex
.execute_statement(&query.statement, &store, &query, &mut warnings)
.unwrap();
match payload {
CalResultPayload::Diff {
source_hash,
target_hash,
changes,
} => {
assert_eq!(source_hash, hash_a.to_hex());
assert_eq!(target_hash, hash_b.to_hex());
let obj_change = changes.iter().find(|c| {
matches!(c,
super::super::ast::FieldDiff::Changed { field, .. } if field == "object"
)
});
assert!(
obj_change.is_some(),
"expected 'object' field to be Changed"
);
}
other => panic!("expected Diff, got {:?}", other),
}
assert!(
!warnings.iter().any(|w| w.contains("CAL-W005")),
"same subject+relation should not trigger CAL-W005"
);
}
#[test]
fn test_history_diff_w005_different_subject() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let (hash_b, grain_b) = make_fact("bob", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_a, grain_a), (hash_b, grain_b)]);
let ex = exec();
let history_stmt = super::HistoryStmt {
hash: hash_a.to_hex(),
where_clause: None,
diff_target: Some(hash_b.to_hex()),
span: None,
};
let query = super::CalQuery {
version: super::super::ast::CalVersion(1),
statement: super::CalStatement::History(history_stmt),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let _ = ex
.execute_statement(&query.statement, &store, &query, &mut warnings)
.unwrap();
assert!(
warnings.iter().any(|w| w.contains("CAL-W005")),
"different subject/relation should trigger CAL-W005, got: {:?}",
warnings
);
}
#[test]
fn test_history_diff_source_not_found() {
let (hash_b, grain_b) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_b, grain_b)]);
let ex = exec();
let fake_hash = "a".repeat(64);
let history_stmt = super::HistoryStmt {
hash: fake_hash,
where_clause: None,
diff_target: Some(hash_b.to_hex()),
span: None,
};
let query = super::CalQuery {
version: super::super::ast::CalVersion(1),
statement: super::CalStatement::History(history_stmt),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex.execute_statement(&query.statement, &store, &query, &mut warnings);
assert!(result.is_err(), "should error when source grain not found");
}
#[test]
fn test_history_diff_target_not_found() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_a, grain_a)]);
let ex = exec();
let fake_hash = "b".repeat(64);
let history_stmt = super::HistoryStmt {
hash: hash_a.to_hex(),
where_clause: None,
diff_target: Some(fake_hash),
span: None,
};
let query = super::CalQuery {
version: super::super::ast::CalVersion(1),
statement: super::CalStatement::History(history_stmt),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex.execute_statement(&query.statement, &store, &query, &mut warnings);
assert!(result.is_err(), "should error when target grain not found");
}
#[test]
fn test_diff_grains_field_added() {
let (_, grain_a) = make_fact("john", "likes", "coffee");
let (_, mut grain_b) = make_fact("john", "likes", "coffee");
grain_b
.fields
.insert("extra".into(), serde_json::json!("new_value"));
let diffs = super::diff_grains(&grain_a, &grain_b);
let added = diffs.iter().find(|d| {
matches!(d,
super::super::ast::FieldDiff::Added { field, .. } if field == "extra"
)
});
assert!(added.is_some(), "should detect added field 'extra'");
}
#[test]
fn test_diff_grains_field_removed() {
let (_, mut grain_a) = make_fact("john", "likes", "coffee");
let (_, grain_b) = make_fact("john", "likes", "coffee");
grain_a
.fields
.insert("old_field".into(), serde_json::json!("old_value"));
let diffs = super::diff_grains(&grain_a, &grain_b);
let removed = diffs.iter().find(|d| {
matches!(d,
super::super::ast::FieldDiff::Removed { field, .. } if field == "old_field"
)
});
assert!(removed.is_some(), "should detect removed field 'old_field'");
}
#[test]
fn test_diff_grains_field_changed() {
let (_, grain_a) = make_fact("john", "likes", "coffee");
let (_, grain_b) = make_fact("john", "likes", "tea");
let diffs = super::diff_grains(&grain_a, &grain_b);
let changed = diffs.iter().find(|d| {
matches!(d,
super::super::ast::FieldDiff::Changed { field, .. } if field == "object"
)
});
assert!(changed.is_some(), "should detect changed field 'object'");
if let Some(super::super::ast::FieldDiff::Changed { old, new, .. }) = changed {
assert_eq!(old.as_str().unwrap(), "coffee");
assert_eq!(new.as_str().unwrap(), "tea");
}
}
#[test]
fn test_diff_grains_identical() {
let (_, grain_a) = make_fact("john", "likes", "vim");
let (_, grain_b) = make_fact("john", "likes", "vim");
let diffs = super::diff_grains(&grain_a, &grain_b);
assert!(
diffs.is_empty(),
"identical grains should produce empty diff"
);
}
#[test]
fn test_execute_describe_capabilities() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE CAPABILITIES", &store).unwrap();
assert_eq!(result.metadata.statement_type, "describe");
match result.result {
CalResultPayload::Describe { info } => {
assert_eq!(info["cal_version"], 1);
assert_eq!(info["conformance_level"], 2);
assert!(!info["supported_statements"].as_array().unwrap().is_empty());
assert_eq!(info["max_sources"], 8);
assert_eq!(info["max_let_bindings"], 5);
}
other => panic!("expected Describe, got {:?}", other),
}
}
#[test]
fn test_execute_describe_server() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE SERVER", &store).unwrap();
match result.result {
CalResultPayload::Describe { info } => {
assert_eq!(info["name"], "areev");
assert!(info.get("version").is_some());
assert_eq!(info["oms_version"], "1.2");
assert!(info.get("build_features").is_some());
}
other => panic!("expected Describe, got {:?}", other),
}
}
#[test]
fn test_execute_describe_fields() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE FIELDS", &store).unwrap();
match result.result {
CalResultPayload::Describe { info } => {
assert!(info.get("fields").is_some());
let fields = info["fields"].as_array().unwrap();
assert!(!fields.is_empty());
let has_subject = fields.iter().any(|f| f["name"] == "subject");
assert!(has_subject, "fields should contain 'subject'");
}
other => panic!("expected Describe, got {:?}", other),
}
}
#[test]
fn test_execute_describe_templates() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE TEMPLATES", &store).unwrap();
match result.result {
CalResultPayload::Describe { info } => {
assert!(info.get("templates").is_some());
let templates = info["templates"].as_array().unwrap();
assert!(!templates.is_empty(), "templates list should not be empty");
for name in ["structured", "readable", "compact"] {
assert!(
templates.iter().any(|t| t["name"] == name),
"templates should include '{name}'"
);
}
assert!(
!templates.iter().any(|t| t["name"] == "toon"),
"no builtin template may shadow the FORMAT toon arm"
);
}
other => panic!("expected Describe, got {:?}", other),
}
}
#[test]
fn test_execute_describe_grammar() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute("DESCRIBE GRAMMAR", &store).unwrap();
match result.result {
CalResultPayload::Describe { info } => {
assert_eq!(info["version"], 1);
assert!(info.get("features").is_some());
let features = info["features"].as_array().unwrap();
assert!(!features.is_empty());
assert_eq!(info["conformance_level"], 2);
}
other => panic!("expected Describe, got {:?}", other),
}
}
#[test]
fn test_execute_explain_assemble() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute(r#"EXPLAIN ASSEMBLE "summary" FROM (RECALL facts)"#, &store)
.unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert_eq!(plan.statement_type, "assemble");
assert!(plan.query_routing.contains("assemble"));
assert!(plan.filters.iter().any(|f| f.contains("sources")));
}
other => panic!("expected Explain, got {:?}", other),
}
}
#[test]
fn test_execute_explain_batch() {
let store = MockStore::empty();
let ex = exec();
let result = ex
.execute("EXPLAIN BATCH { RECALL facts ; RECALL events }", &store)
.unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert_eq!(plan.statement_type, "batch");
assert_eq!(plan.query_routing, "parallel_batch");
assert!(plan
.filters
.iter()
.any(|f| f.contains("parallel_execution")));
}
other => panic!("expected Explain, got {:?}", other),
}
}
#[test]
fn test_execute_explain_coalesce() {
let store = MockStore::empty();
let ex = exec();
let result = ex.execute(
r#"EXPLAIN COALESCE(RECALL facts WHERE subject = "john", RECALL facts WHERE subject = "bob")"#,
&store,
).unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert_eq!(plan.statement_type, "coalesce");
assert_eq!(plan.query_routing, "coalesce_fallback");
assert!(plan.filters.iter().any(|f| f.contains("fallback_chain")));
}
other => panic!("expected Explain, got {:?}", other),
}
}
#[test]
fn test_execute_explain_describe_rejected() {
let store = MockStore::empty();
let ex = exec();
let err = ex
.execute("EXPLAIN DESCRIBE SCHEMA", &store)
.expect_err("EXPLAIN DESCRIBE should be a parse error per §8.5");
assert_eq!(err.code(), "CAL-E002");
assert!(
err.to_string().contains("DESCRIBE"),
"error should mention DESCRIBE: {}",
err
);
}
#[test]
fn test_execute_explain_history() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let hex = hash.to_hex();
let query_str = format!("EXPLAIN HISTORY OF sha256:{}", hex);
let result = ex.execute(&query_str, &store).unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert_eq!(plan.statement_type, "history");
assert_eq!(plan.query_routing, "entity_latest");
assert!(plan.index_usage.contains(&"entity_latest".to_string()));
}
other => panic!("expected Explain, got {:?}", other),
}
}
#[test]
fn test_field_diff_serialization() {
let diff = super::super::ast::FieldDiff::Changed {
field: "object".into(),
old: serde_json::json!("coffee"),
new: serde_json::json!("tea"),
};
let json = serde_json::to_value(&diff).unwrap();
assert_eq!(json["kind"], "changed");
assert_eq!(json["field"], "object");
assert_eq!(json["old"], "coffee");
assert_eq!(json["new"], "tea");
}
#[test]
fn test_diff_payload_serialization() {
let payload = CalResultPayload::Diff {
source_hash: "aaa".into(),
target_hash: "bbb".into(),
changes: vec![
super::super::ast::FieldDiff::Added {
field: "new_field".into(),
value: serde_json::json!("value"),
},
super::super::ast::FieldDiff::Removed {
field: "old_field".into(),
value: serde_json::json!(42),
},
],
};
let json = serde_json::to_value(&payload).unwrap();
assert_eq!(json["type"], "diff");
assert_eq!(json["source_hash"], "aaa");
assert_eq!(json["target_hash"], "bbb");
assert_eq!(json["changes"].as_array().unwrap().len(), 2);
}
#[test]
fn test_build_features_list() {
let features = super::build_features_list();
assert!(features.contains(&"cal"), "cal feature should be active");
}
#[test]
fn test_count_payload_results_diff() {
let payload = CalResultPayload::Diff {
source_hash: "a".into(),
target_hash: "b".into(),
changes: vec![super::super::ast::FieldDiff::Added {
field: "x".into(),
value: serde_json::json!(1),
}],
};
assert_eq!(super::count_payload_results(&payload), 1);
}
#[test]
fn test_i5_value_to_string_parameter_returns_error() {
let val = Value::Parameter {
name: "test".into(),
};
let result = super::value_to_string(&val);
assert!(
result.is_err(),
"Parameter must return error, not \"$test\""
);
match result.unwrap_err() {
CalError::UnboundParameter { name, .. } => {
assert_eq!(name, "test");
}
other => panic!("expected UnboundParameter, got: {:?}", other),
}
}
#[test]
fn test_i7_subjects_extractor_returns_grains() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex.execute("RECALL facts SUBJECTS", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].grain_type, "extracted");
let value = grains[0].fields.get("value").unwrap();
assert_eq!(value.as_str().unwrap(), "john");
}
other => panic!("SUBJECTS should return Grains (I-7 fix), got: {:?}", other),
}
}
#[test]
fn test_i7_objects_extractor_returns_grains() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex.execute("RECALL facts OBJECTS", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].grain_type, "extracted");
let value = grains[0].fields.get("value").unwrap();
assert_eq!(value.as_str().unwrap(), "coffee");
}
other => panic!("OBJECTS should return Grains (I-7 fix), got: {:?}", other),
}
}
#[test]
fn test_i7_hashes_extractor_returns_grains() {
let (hash, grain) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash, grain)]);
let ex = exec();
let result = ex.execute("RECALL facts HASHES", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].grain_type, "extracted");
let value = grains[0].fields.get("value").unwrap();
assert_eq!(value.as_str().unwrap(), hash.to_hex());
}
other => panic!("HASHES should return Grains (I-7 fix), got: {:?}", other),
}
}
#[test]
fn test_let_binding_subjects_extractor() {
let grains = vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "tea"),
];
let store = MockStore::with_grains(grains);
let ex = exec();
let result = ex
.execute(
r#"LET $users = RECALL facts SUBJECTS; RECALL facts WHERE subject = "john""#,
&store,
)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
#[test]
fn test_let_scope_evaluate_basic() {
let grains = vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
];
let store = MockStore::with_grains(grains);
let ex = exec();
let binding = super::super::ast::LetBinding {
name: "users".into(),
extractor: super::super::ast::Extractor::Subjects,
source: Box::new(CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: vec![binding.clone()],
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let scope =
super::LetScope::evaluate(&[binding], &ex, &store, &query, &mut warnings).unwrap();
match scope.resolve("users").unwrap() {
super::LetValue::Extracted(values) => {
assert!(values.contains(&"john".to_string()));
assert!(values.contains(&"bob".to_string()));
}
other => panic!("expected Extracted, got: {:?}", other),
}
assert!(scope.resolve("missing").is_err());
}
#[test]
fn test_let_scope_too_many_bindings_s06() {
let store = MockStore::empty();
let ex = exec();
let bindings: Vec<super::super::ast::LetBinding> = (0..6)
.map(|i| super::super::ast::LetBinding {
name: format!("var{}", i),
extractor: super::super::ast::Extractor::Subjects,
source: Box::new(CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
span: None,
})
.collect();
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: bindings.clone(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = super::LetScope::evaluate(&bindings, &ex, &store, &query, &mut warnings);
assert!(result.is_err(), "6 bindings must exceed S-06 limit of 5");
match result.unwrap_err() {
CalError::TooManyLetBindings { count, max, .. } => {
assert_eq!(count, 6);
assert_eq!(max, 5);
}
other => panic!("expected TooManyLetBindings, got: {:?}", other),
}
}
#[test]
fn test_redact_budget_metadata_config_default() {
let config = CalExecutorConfig::default();
assert!(
!config.redact_budget_metadata,
"default should be false (S-09)"
);
}
#[test]
fn test_redact_budget_metadata_config_override() {
let config = CalExecutorConfig {
redact_budget_metadata: true,
..Default::default()
};
assert!(config.redact_budget_metadata);
}
#[test]
fn test_count_payload_results_assembled() {
let payload = CalResultPayload::Assembled {
grains: vec![CalGrainResult {
hash: "abc".into(),
grain_type: "fact".into(),
score: 1.0,
fields: serde_json::json!({}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
}],
sources: vec![],
total_tokens: 100,
budget_limit: Some(500),
progressive: false,
total_available: Some(1),
};
assert_eq!(super::count_payload_results(&payload), 1);
}
#[test]
fn test_labeled_batch_returns_keyed_results() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let batch = super::super::ast::BatchStmt {
statements: Vec::new(),
labeled: Some(vec![
(
"prefs".to_string(),
super::super::ast::BatchEntry {
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "john".into(),
},
span: None,
},
span: None,
}),
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
user_vars: HashMap::new(),
},
),
(
"all".to_string(),
super::super::ast::BatchEntry {
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
user_vars: HashMap::new(),
},
),
]),
span: None,
};
let mut warnings = Vec::new();
let result = ex.execute_batch(&batch, &store, &mut warnings).unwrap();
match result {
CalResultPayload::Batch { results } => {
assert!(results.contains_key("prefs"), "should have 'prefs' label");
assert!(results.contains_key("all"), "should have 'all' label");
assert_eq!(results.len(), 2, "exactly 2 labeled results");
}
other => panic!("expected Batch, got: {:?}", other),
}
}
#[test]
fn test_labeled_batch_duplicate_label_errors() {
let store = MockStore::empty();
let ex = exec();
let batch = super::super::ast::BatchStmt {
statements: Vec::new(),
labeled: Some(vec![
(
"dup".to_string(),
super::super::ast::BatchEntry {
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
user_vars: HashMap::new(),
},
),
(
"dup".to_string(),
super::super::ast::BatchEntry {
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Events,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
user_vars: HashMap::new(),
},
),
]),
span: None,
};
let mut warnings = Vec::new();
let result = ex.execute_batch(&batch, &store, &mut warnings);
assert!(result.is_err());
match result.unwrap_err() {
CalError::AssembleDuplicateLabel { label, .. } => {
assert_eq!(label, "dup");
}
other => panic!("expected AssembleDuplicateLabel, got: {:?}", other),
}
}
#[test]
fn test_positional_batch_still_works() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let batch = super::super::ast::BatchStmt {
statements: vec![super::super::ast::BatchEntry {
statement: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
user_vars: HashMap::new(),
}],
labeled: None,
span: None,
};
let mut warnings = Vec::new();
let result = ex.execute_batch(&batch, &store, &mut warnings).unwrap();
match result {
CalResultPayload::Batch { results } => {
assert!(
results.contains_key("0"),
"positional batch uses index keys"
);
}
other => panic!("expected Batch, got: {:?}", other),
}
}
#[test]
fn test_coalesce_multibranch_first_hit_wins() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let coalesce = super::super::ast::CoalesceStmt {
grain_type: GrainTypePlural::Facts,
where_clause: None,
branches: vec![
super::super::ast::CoalesceBranch {
query: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "john".into(),
},
span: None,
},
span: None,
}),
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
span: None,
},
super::super::ast::CoalesceBranch {
query: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Events,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
span: None,
},
],
else_branch: None,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Coalesce(coalesce.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_coalesce(&coalesce, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1, "branch 1 should return john");
}
other => panic!("expected Grains, got: {:?}", other),
}
assert!(
warnings.iter().any(|w| w.contains("short-circuited")),
"should emit short-circuit warning"
);
}
#[test]
fn test_coalesce_multibranch_falls_through_to_else() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let coalesce = super::super::ast::CoalesceStmt {
grain_type: GrainTypePlural::Facts,
where_clause: None,
branches: vec![
super::super::ast::CoalesceBranch {
query: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "zzz".into(),
},
span: None,
},
span: None,
}),
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
span: None,
},
],
else_branch: Some(Box::new(CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}))),
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Coalesce(coalesce.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_coalesce(&coalesce, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1, "ELSE branch should return john");
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_coalesce_multibranch_all_empty_no_else() {
let store = MockStore::empty();
let ex = exec();
let coalesce = super::super::ast::CoalesceStmt {
grain_type: GrainTypePlural::Facts,
where_clause: None,
branches: vec![super::super::ast::CoalesceBranch {
query: CalStatement::Recall(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
}),
span: None,
}],
else_branch: None,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Coalesce(coalesce.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_coalesce(&coalesce, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert!(grains.is_empty(), "all empty, no ELSE → empty result");
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_explain_reports_tier1_disabled() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
tier1_enabled: false,
..Default::default()
});
let result = ex.execute("EXPLAIN RECALL facts", &store).unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert!(
plan.filters.iter().any(|f| f.contains("tier1_disabled")),
"should report tier1_disabled: {:?}",
plan.filters
);
}
other => panic!("expected Explain, got: {:?}", other),
}
}
#[test]
fn test_explain_reports_namespace_override() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
namespace_override: Some("test_ns".to_string()),
..Default::default()
});
let result = ex.execute("EXPLAIN RECALL facts", &store).unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert!(
plan.filters
.iter()
.any(|f| f.contains("namespace_override")),
"should report namespace_override: {:?}",
plan.filters
);
}
other => panic!("expected Explain, got: {:?}", other),
}
}
#[test]
fn test_explain_reports_user_id_override() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
user_id_override: Some("user123".to_string()),
..Default::default()
});
let result = ex.execute("EXPLAIN RECALL facts", &store).unwrap();
match result.result {
CalResultPayload::Explain { plan } => {
assert!(
plan.filters.iter().any(|f| f.contains("user_id_override")),
"should report user_id_override: {:?}",
plan.filters
);
}
other => panic!("expected Explain, got: {:?}", other),
}
}
#[test]
fn test_type_specific_fields_tools() {
let fields = super::type_specific_fields(&GrainTypePlural::Tools);
assert!(fields.contains(&"tool"), "Tools should have 'tool' field");
assert!(
fields.contains(&"is_error"),
"Tools should have 'is_error' field"
);
assert!(
fields.contains(&"duration_ms"),
"Tools should have 'duration_ms' field"
);
}
#[test]
fn test_type_specific_fields_goals() {
let fields = super::type_specific_fields(&GrainTypePlural::Goals);
assert!(fields.contains(&"title"), "Goals should have 'title' field");
assert!(super::COMMON_FIELDS.contains(&"priority"));
assert!(super::COMMON_FIELDS.contains(&"status"));
}
#[test]
fn test_type_specific_fields_all_returns_empty() {
let fields = super::type_specific_fields(&GrainTypePlural::All);
assert!(
fields.is_empty(),
"All should return empty (no type-specific validation)"
);
}
#[test]
fn test_is_known_type_specific_field() {
assert!(
super::is_known_type_specific_field("tool"),
"'tool' is known (on Tools)"
);
assert!(super::COMMON_FIELDS.contains(&"priority"));
assert!(
!super::is_known_type_specific_field("zzz_unknown"),
"'zzz_unknown' is not known"
);
}
#[test]
fn test_grain_matches_condition_string_eq() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({
"tool": "web_search",
"is_error": false,
"duration_ms": 150
}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "web_search".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "db_query".into()
}
));
}
#[test]
fn test_grain_matches_condition_number_comparisons() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({
"duration_ms": 150.0
}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"duration_ms",
&super::super::ast::Comparator::Gte,
&super::super::ast::Value::Number { value: 100.0 }
));
assert!(super::grain_matches_condition(
&grain,
"duration_ms",
&super::super::ast::Comparator::Gte,
&super::super::ast::Value::Number { value: 150.0 }
));
assert!(!super::grain_matches_condition(
&grain,
"duration_ms",
&super::super::ast::Comparator::Gte,
&super::super::ast::Value::Number { value: 200.0 }
));
assert!(super::grain_matches_condition(
&grain,
"duration_ms",
&super::super::ast::Comparator::Lt,
&super::super::ast::Value::Number { value: 200.0 }
));
assert!(!super::grain_matches_condition(
&grain,
"duration_ms",
&super::super::ast::Comparator::Lt,
&super::super::ast::Value::Number { value: 150.0 }
));
}
#[test]
fn test_grain_matches_condition_boolean() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({
"is_error": false
}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"is_error",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::Boolean { value: false }
));
assert!(!super::grain_matches_condition(
&grain,
"is_error",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::Boolean { value: true }
));
}
#[test]
fn test_grain_matches_condition_not_eq() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({
"tool": "web_search"
}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::NotEq,
&super::super::ast::Value::String {
value: "db_query".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::NotEq,
&super::super::ast::Value::String {
value: "web_search".into()
}
));
}
#[test]
fn test_grain_matches_condition_missing_field_returns_false() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(!super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "anything".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"tool",
&super::super::ast::Comparator::NotEq,
&super::super::ast::Value::String {
value: "anything".into()
}
));
}
#[test]
fn test_207_negations_fail_closed_on_a_field_the_grain_lacks() {
use super::super::ast::{Comparator, Condition, Value};
let skill = CalGrainResult {
hash: "abc".into(),
grain_type: "skill".into(),
score: 1.0,
fields: serde_json::json!({ "name": "beta", "description": "retired" }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let retired = || Value::String {
value: "retired".into(),
};
let object_ne = Condition::Comparison {
field: "object".into(),
comparator: Comparator::NotEq,
value: retired(),
span: None,
};
assert!(!super::grain_matches_condition_tree(&skill, &object_ne));
assert!(!super::grain_matches_condition_tree(
&skill,
&Condition::Not {
inner: Box::new(Condition::Comparison {
field: "object".into(),
comparator: Comparator::Eq,
value: retired(),
span: None,
}),
span: None,
}
));
assert!(!super::grain_matches_condition_tree(
&skill,
&Condition::NotIn {
field: "object".into(),
values: vec![retired()],
span: None,
}
));
assert!(super::grain_matches_condition_tree(
&skill,
&Condition::Comparison {
field: "description".into(),
comparator: Comparator::NotEq,
value: Value::String {
value: "active".into()
},
span: None,
}
));
}
#[test]
fn test_207_unknown_propagates_by_the_sql_truth_tables() {
use super::super::ast::{Comparator, Condition, Value};
let skill = CalGrainResult {
hash: "abc".into(),
grain_type: "skill".into(),
score: 1.0,
fields: serde_json::json!({ "name": "beta" }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let known_true = Condition::Comparison {
field: "name".into(),
comparator: Comparator::Eq,
value: Value::String {
value: "beta".into(),
},
span: None,
};
let unknown = Condition::Comparison {
field: "object".into(),
comparator: Comparator::Eq,
value: Value::String {
value: "anything".into(),
},
span: None,
};
assert!(super::grain_matches_condition_tree(
&skill,
&Condition::Or {
left: Box::new(known_true.clone()),
right: Box::new(unknown.clone()),
span: None,
}
));
assert!(!super::grain_matches_condition_tree(
&skill,
&Condition::And {
left: Box::new(known_true),
right: Box::new(unknown),
span: None,
}
));
assert!(super::grain_matches_condition_tree(
&skill,
&Condition::IsNull {
field: "object".into(),
span: None,
}
));
assert!(super::grain_matches_condition_tree(
&skill,
&Condition::IsNotNull {
field: "name".into(),
span: None,
}
));
}
#[test]
fn test_207_omit_default_discriminators_survive_fails_closed() {
use super::super::ast::{Comparator, Condition, Value};
let legacy_execution = CalGrainResult {
hash: "abc".into(),
grain_type: "tool".into(),
score: 1.0,
fields: serde_json::json!({ "tool_name": "db_query" }),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition_tree(
&legacy_execution,
&Condition::Comparison {
field: "kind".into(),
comparator: Comparator::NotEq,
value: Value::String {
value: "definition".into()
},
span: None,
}
));
}
#[test]
fn test_206_validity_fields_are_filterable_and_sortable() {
for f in [
"valid_from",
"valid_to",
"system_valid_from",
"system_valid_to",
] {
assert!(
super::COMMON_FIELDS.contains(&f),
"{f} must be pipeline-valid (ORDER BY / SELECT / GROUP BY)"
);
assert!(
super::GRAIN_EVALUABLE_COMMON.contains(&f),
"{f} must be evaluable per grain (WHERE residual)"
);
assert!(!super::ENGINE_ONLY_FIELDS.contains(&f));
}
}
#[test]
fn test_plan_residual_where_splits_pushdown_from_residual() {
let condition = super::super::ast::Condition::And {
left: Box::new(super::super::ast::Condition::Comparison {
field: "subject".into(), comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "john".into(),
},
span: None,
}),
right: Box::new(super::super::ast::Condition::Comparison {
field: "tool".into(), comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "web_search".into(),
},
span: None,
}),
span: None,
};
let mut warnings = Vec::new();
let residual =
super::plan_residual_where(&condition, &GrainTypePlural::Tools, &mut warnings)
.expect("plan must succeed")
.expect("the type-specific leaf must remain residual");
match residual {
super::super::ast::Condition::Comparison { field, .. } => {
assert_eq!(field, "tool");
}
other => panic!("expected the bare 'tool' leaf, got {other:?}"),
}
assert!(warnings.is_empty(), "no warnings for valid fields");
}
#[test]
fn test_plan_residual_where_unpushed_comparator_on_common_field() {
let condition = super::super::ast::Condition::Comparison {
field: "confidence".into(),
comparator: super::super::ast::Comparator::Lt,
value: super::super::ast::Value::Number { value: 0.5 },
span: None,
};
let mut warnings = Vec::new();
let residual =
super::plan_residual_where(&condition, &GrainTypePlural::Facts, &mut warnings)
.expect("plan must succeed");
assert!(
residual.is_some(),
"an unpushed comparator on a common field must be post-filtered"
);
}
#[test]
fn test_plan_residual_where_refuses_unfilterable_field_on_typed_recall() {
let condition = super::super::ast::Condition::Comparison {
field: "priority".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "high".into(),
},
span: None,
};
let mut warnings = Vec::new();
let err = super::plan_residual_where(&condition, &GrainTypePlural::Tools, &mut warnings)
.expect_err("a filter that cannot be honoured must refuse");
assert_eq!(err.code(), "CAL-E060");
}
#[test]
fn test_plan_residual_where_refuses_engine_field_under_not() {
let condition = super::super::ast::Condition::Not {
inner: Box::new(super::super::ast::Condition::Comparison {
field: "query".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String { value: "x".into() },
span: None,
}),
span: None,
};
let mut warnings = Vec::new();
let err = super::plan_residual_where(&condition, &GrainTypePlural::Facts, &mut warnings)
.expect_err("engine-only fields under NOT must refuse");
assert_eq!(err.code(), "CAL-E061");
}
#[test]
fn test_plan_residual_where_keeps_not_subtree_whole() {
let condition = super::super::ast::Condition::Not {
inner: Box::new(super::super::ast::Condition::Comparison {
field: "tool_name".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String { value: "x".into() },
span: None,
}),
span: None,
};
let mut warnings = Vec::new();
let residual =
super::plan_residual_where(&condition, &GrainTypePlural::Tools, &mut warnings)
.expect("plan must succeed")
.expect("NOT subtree must remain residual");
assert!(
matches!(residual, super::super::ast::Condition::Not { .. }),
"the negation must be preserved"
);
}
#[test]
fn test_pushdown_consumed_matches_apply_where_clause_arms() {
use super::super::ast::{Comparator, Condition, Value};
let cmp = |field: &str, comparator: Comparator| Condition::Comparison {
field: field.into(),
comparator,
value: Value::String { value: "v".into() },
span: None,
};
for f in [
"subject", "relation", "object", "namespace", "user_id", "query", "time", "entity",
"scope", "scope_path",
] {
assert!(super::leaf_pushdown_consumed(&cmp(f, Comparator::Eq)), "{f} Eq");
assert!(!super::leaf_pushdown_consumed(&cmp(f, Comparator::NotEq)), "{f} NotEq");
}
for f in ["confidence", "importance"] {
assert!(super::leaf_pushdown_consumed(&cmp(f, Comparator::Gte)), "{f} Gte");
assert!(!super::leaf_pushdown_consumed(&cmp(f, Comparator::Lt)), "{f} Lt");
}
assert!(super::leaf_pushdown_consumed(&cmp("contradicted", Comparator::Eq)));
assert!(!super::leaf_pushdown_consumed(&cmp("hash", Comparator::Eq)));
assert!(!super::leaf_pushdown_consumed(&cmp("session_id", Comparator::Eq)));
assert!(!super::leaf_pushdown_consumed(&cmp("tool_name", Comparator::Eq)));
assert!(!super::leaf_pushdown_consumed(&cmp("status", Comparator::Eq)));
let in_cond = |field: &str| Condition::In {
field: field.into(),
values: vec![Value::String { value: "v".into() }],
span: None,
};
for f in ["subject", "relation", "object", "tags", "namespace"] {
assert!(super::leaf_pushdown_consumed(&in_cond(f)), "{f} IN");
}
assert!(!super::leaf_pushdown_consumed(&in_cond("role")));
let not_in_tags = Condition::NotIn {
field: "tags".into(),
values: vec![Value::String { value: "v".into() }],
span: None,
};
assert!(super::leaf_pushdown_consumed(¬_in_tags));
}
#[test]
fn test_suggest_field_cross_type() {
let suggestion = super::suggest_field("tool", &GrainTypePlural::Goals);
assert!(
suggestion.is_some(),
"should suggest 'tool' exists on another type"
);
assert!(
suggestion.unwrap().contains("different grain type"),
"should mention different grain type"
);
}
#[test]
fn test_suggest_field_substring_match() {
let suggestion = super::suggest_field("stat", &GrainTypePlural::Goals);
assert!(
suggestion.is_some(),
"should suggest 'status' for substring 'stat'"
);
}
#[test]
fn test_type_specific_fields_events() {
let fields = super::type_specific_fields(&GrainTypePlural::Events);
assert!(fields.contains(&"session_id"));
assert!(fields.contains(&"content"));
assert!(fields.contains(&"created_at"));
}
#[test]
fn test_type_specific_fields_states() {
let fields = super::type_specific_fields(&GrainTypePlural::States);
assert!(fields.contains(&"context"));
assert!(fields.contains(&"plan"));
assert!(fields.contains(&"history"));
assert!(!fields.contains(&"checkpoint_data"));
assert!(!fields.contains(&"session_id"));
}
#[test]
fn test_type_specific_fields_consents() {
let fields = super::type_specific_fields(&GrainTypePlural::Consents);
assert!(super::COMMON_FIELDS.contains(&"scope"));
assert!(fields.contains(&"granted"));
assert!(fields.contains(&"subject_did"));
assert!(fields.contains(&"grantee_did"));
}
#[test]
fn test_type_specific_fields_observations() {
let fields = super::type_specific_fields(&GrainTypePlural::Observations);
assert!(fields.contains(&"sensor"));
assert!(fields.contains(&"value"));
assert!(fields.contains(&"unit"));
}
#[test]
fn test_type_specific_fields_workflows() {
let fields = super::type_specific_fields(&GrainTypePlural::Workflows);
assert!(fields.contains(&"name"));
assert!(super::COMMON_FIELDS.contains(&"status"));
assert!(fields.contains(&"nodes"));
}
#[test]
fn test_type_specific_fields_reasonings() {
let fields = super::type_specific_fields(&GrainTypePlural::Reasonings);
assert!(fields.contains(&"premises"));
assert!(fields.contains(&"conclusion"));
assert!(super::COMMON_FIELDS.contains(&"confidence"));
}
#[test]
fn test_session_id_not_in_common_fields() {
assert!(
!super::COMMON_FIELDS.contains(&"session_id"),
"session_id should not be in COMMON_FIELDS"
);
}
#[test]
fn test_session_id_stays_residual_despite_scan_pushdown() {
let condition = super::super::ast::Condition::And {
left: Box::new(super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "alice".into(),
},
span: None,
}),
right: Box::new(super::super::ast::Condition::Comparison {
field: "session_id".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "sess-001".into(),
},
span: None,
}),
span: None,
};
let mut warnings = Vec::new();
let residual =
super::plan_residual_where(&condition, &GrainTypePlural::Events, &mut warnings)
.expect("plan must succeed")
.expect("session_id must remain residual");
match residual {
super::super::ast::Condition::Comparison { field, .. } => {
assert_eq!(field, "session_id");
}
other => panic!("expected the session_id leaf, got {other:?}"),
}
}
#[test]
fn test_goal_state_post_filter() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "goal".into(),
score: 1.0,
fields: serde_json::json!({"goal_state": "active", "subject": "alice"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"goal_state",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "active".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"goal_state",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "failed".into()
}
));
}
#[test]
fn test_session_id_post_filter() {
let grain = CalGrainResult {
hash: "abc".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::json!({"session_id": "sess-001", "subject": "alice"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"session_id",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "sess-001".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"session_id",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "sess-999".into()
}
));
}
#[test]
fn event_role_listed_as_type_specific() {
let fields = super::type_specific_fields(&GrainTypePlural::Events);
assert!(fields.contains(&"role"));
assert!(fields.contains(&"parent_message_id"));
assert!(fields.contains(&"model_id"));
assert!(fields.contains(&"stop_reason"));
}
#[test]
fn event_role_not_in_common_fields() {
assert!(!super::COMMON_FIELDS.contains(&"role"));
assert!(!super::COMMON_FIELDS.contains(&"parent_message_id"));
assert!(!super::COMMON_FIELDS.contains(&"model_id"));
assert!(!super::COMMON_FIELDS.contains(&"stop_reason"));
}
#[test]
fn role_eq_post_filter() {
let grain = CalGrainResult {
hash: "h1".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::json!({"role": "user", "content": "hi"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"role",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "user".into()
}
));
assert!(!super::grain_matches_condition(
&grain,
"role",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "assistant".into()
}
));
}
#[test]
fn role_in_list_post_filter() {
use super::super::ast::{Condition, Value};
let condition = Condition::In {
field: "role".into(),
values: vec![
Value::String {
value: "user".into(),
},
Value::String {
value: "assistant".into(),
},
],
span: None,
};
let user_grain = CalGrainResult {
hash: "h1".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::json!({"role": "user"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let tool_grain = CalGrainResult {
hash: "h2".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::json!({"role": "tool"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition_tree(&user_grain, &condition));
assert!(!super::grain_matches_condition_tree(
&tool_grain,
&condition
));
}
#[test]
fn parent_message_id_eq_post_filter() {
let grain = CalGrainResult {
hash: "h1".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::json!({"parent_message_id": "deadbeef"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
assert!(super::grain_matches_condition(
&grain,
"parent_message_id",
&super::super::ast::Comparator::Eq,
&super::super::ast::Value::String {
value: "deadbeef".into()
}
));
}
#[test]
fn role_and_session_id_both_stay_residual() {
use super::super::ast::{Comparator, Condition, Value};
let condition = Condition::And {
left: Box::new(Condition::Comparison {
field: "role".into(),
comparator: Comparator::Eq,
value: Value::String {
value: "user".into(),
},
span: None,
}),
right: Box::new(Condition::Comparison {
field: "session_id".into(),
comparator: Comparator::Eq,
value: Value::String { value: "s1".into() },
span: None,
}),
span: None,
};
let mut warnings = Vec::new();
let residual =
super::plan_residual_where(&condition, &GrainTypePlural::Events, &mut warnings)
.expect("plan must succeed")
.expect("both leaves must remain residual");
match residual {
super::super::ast::Condition::And { .. } => {}
other => panic!("expected the full AND tree, got {other:?}"),
}
}
#[test]
fn test_assemble_where_filters_results() {
let store = MockStore::with_grains(vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "john".into(),
},
span: None,
},
span: None,
}),
context_name: None,
sources: None,
budget: None,
priority: None,
format: None,
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1, "WHERE subject='john' should filter to 1");
assert_eq!(
grains[0].fields.get("subject").and_then(|v| v.as_str()),
Some("john")
);
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_assemble_format_json() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: None,
context_name: None,
sources: None,
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Json,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
assert_eq!(grains[0].grain_type, "fact");
let subject = grains[0].fields.get("subject").and_then(|v| v.as_str());
assert_eq!(subject, Some("john"));
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_assemble_format_markdown() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: None,
context_name: None,
sources: None,
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Markdown,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Formatted {
text,
format,
grain_count,
..
} => {
assert_eq!(format, "markdown");
assert_eq!(grain_count, 1);
assert!(
!text.contains("###"),
"ungrouped markdown should not carry per-grain headings, got: {text}"
);
assert!(
text.starts_with("- **"),
"Markdown should render an assertion line, got: {text}"
);
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_assemble_format_triples() {
let store = MockStore::with_grains(vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: None,
context_name: None,
sources: None,
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Triples,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Formatted {
text,
format,
grain_count,
..
} => {
assert_eq!(format, "triples");
assert_eq!(grain_count, 2);
assert_eq!(text.lines().count(), 2, "should have 2 triple lines");
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_assemble_format_with_where_filters_then_formats() {
let store = MockStore::with_grains(vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "john".into(),
},
span: None,
},
span: None,
}),
context_name: None,
sources: None,
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Json,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1, "WHERE should filter to 1 grain");
let subject = grains[0].fields.get("subject").and_then(|v| v.as_str());
assert_eq!(subject, Some("john"));
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_assemble_without_where_returns_all() {
let store = MockStore::with_grains(vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
]);
let ex = exec();
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
})),
where_clause: None,
context_name: None,
sources: None,
budget: None,
priority: None,
format: None,
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 2, "no WHERE should return all");
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_multisource_assemble_format_json() {
let store = MockStore::with_grains(vec![
make_fact("john", "likes", "coffee"),
make_fact("bob", "likes", "coffee"),
]);
let ex = exec();
let recall_facts = RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
};
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(recall_facts.clone())),
where_clause: None,
context_name: None,
sources: Some(vec![super::super::ast::NamedSource {
label: "facts".into(),
literal: None,
pinned: false,
query: Box::new(CalStatement::Recall(recall_facts)),
with_options: vec![],
pipeline: vec![],
span: None,
}]),
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Json,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Grains { grains, .. } => {
assert!(!grains.is_empty(), "should have grains");
assert!(
grains
.iter()
.filter_map(|g| g.fields.get("subject").and_then(|v| v.as_str()))
.any(|x| x == "john"),
"should contain john"
);
}
other => panic!(
"expected Grains for multi-source FORMAT json, got: {:?}",
other
),
}
}
#[test]
fn test_multisource_assemble_format_sml() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let recall_facts = RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
};
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(recall_facts.clone())),
where_clause: None,
context_name: None,
sources: Some(vec![super::super::ast::NamedSource {
label: "events".into(),
literal: None,
pinned: false,
query: Box::new(CalStatement::Recall(recall_facts)),
with_options: vec![],
pipeline: vec![],
span: None,
}]),
budget: None,
priority: None,
format: Some(super::super::ast::FormatClause::Single(
super::super::ast::FormatSpec::Sml,
)),
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Formatted {
text,
format,
grain_count,
..
} => {
assert_eq!(format, "sml");
assert!(grain_count > 0, "should have grains");
assert!(text.contains("<grains>"), "SML should contain <grains> tag");
}
other => panic!(
"expected Formatted for multi-source FORMAT sml, got: {:?}",
other
),
}
}
#[test]
fn test_multisource_assemble_without_format_returns_assembled() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = exec();
let recall_facts = RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None,
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
};
let assemble = super::super::ast::AssembleStmt {
topic: "test".into(),
from: super::super::ast::Source::Query(Box::new(recall_facts.clone())),
where_clause: None,
context_name: None,
sources: Some(vec![super::super::ast::NamedSource {
label: "facts".into(),
literal: None,
pinned: false,
query: Box::new(CalStatement::Recall(recall_facts)),
with_options: vec![],
pipeline: vec![],
span: None,
}]),
budget: None,
priority: None,
format: None,
for_whom: None,
assemble_with: Vec::new(),
with_options: Vec::new(),
streaming: false,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Assemble(assemble.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let result = ex
.execute_assemble(&assemble, &store, &query, &mut warnings)
.unwrap();
match result {
CalResultPayload::Assembled { grains, .. } => {
assert!(!grains.is_empty(), "should have assembled grains");
}
other => panic!("expected Assembled (no FORMAT), got: {:?}", other),
}
}
#[test]
fn test_multi_format_rendering() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts FORMAT [json, markdown]", &store)
.unwrap();
match result.result {
CalResultPayload::MultiFormatted {
formats,
grain_count,
..
} => {
assert_eq!(grain_count, 1);
assert_eq!(formats.len(), 2);
assert!(formats.contains_key("json"), "should have json key");
assert!(formats.contains_key("markdown"), "should have markdown key");
assert!(formats["json"].contains("john"));
assert!(formats["markdown"].contains("john"));
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
#[test]
fn test_single_format_backward_compat() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts FORMAT json", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 1);
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_multi_format_single_element_list() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts FORMAT [json]", &store).unwrap();
match result.result {
CalResultPayload::MultiFormatted {
formats,
grain_count,
..
} => {
assert_eq!(grain_count, 1);
assert_eq!(formats.len(), 1);
assert!(formats.contains_key("json"));
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
#[test]
fn test_multi_format_all_seven_types() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
"RECALL facts FORMAT [json, markdown, yaml, text, sml]",
&store,
)
.unwrap();
match result.result {
CalResultPayload::MultiFormatted { formats, .. } => {
assert_eq!(formats.len(), 5);
for key in &["json", "markdown", "yaml", "text", "sml"] {
assert!(formats.contains_key(*key), "missing format: {}", key);
}
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
#[test]
fn test_no_format_returns_grains() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts", &store).unwrap();
assert!(matches!(result.result, CalResultPayload::Grains { .. }));
}
#[test]
fn test_multi_format_with_aliases() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
"RECALL facts FORMAT [json AS customers, markdown AS report]",
&store,
)
.unwrap();
match result.result {
CalResultPayload::MultiFormatted {
formats,
grain_count,
..
} => {
assert_eq!(grain_count, 1);
assert_eq!(formats.len(), 2);
assert!(
formats.contains_key("customers"),
"should have aliased key 'customers'"
);
assert!(
formats.contains_key("report"),
"should have aliased key 'report'"
);
assert!(
!formats.contains_key("json"),
"should NOT have canonical key 'json'"
);
assert!(
!formats.contains_key("markdown"),
"should NOT have canonical key 'markdown'"
);
assert!(formats["customers"].contains("john"));
assert!(formats["report"].contains("john"));
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
#[test]
fn test_multi_format_mixed_alias_and_no_alias() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts FORMAT [json AS customers, markdown]", &store)
.unwrap();
match result.result {
CalResultPayload::MultiFormatted {
formats,
grain_count,
..
} => {
assert_eq!(grain_count, 1);
assert_eq!(formats.len(), 2);
assert!(formats.contains_key("customers"), "aliased key");
assert!(
formats.contains_key("markdown"),
"canonical key for non-aliased"
);
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
#[test]
fn test_multi_format_template_alias() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
r#"RECALL facts FORMAT [TEMPLATE "{{subject}}: {{object}}" AS summary, json]"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::MultiFormatted {
formats,
grain_count,
..
} => {
assert_eq!(grain_count, 1);
assert_eq!(formats.len(), 2);
assert!(formats.contains_key("summary"), "template alias");
assert!(formats.contains_key("json"), "json canonical");
assert!(formats["summary"].contains("john: coffee"));
}
other => panic!("expected MultiFormatted, got: {:?}", other),
}
}
fn make_fact_with_time(
subject: &str,
relation: &str,
object: &str,
created_at_sec: u32,
) -> (Hash, DeserializedGrain) {
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("subject".into(), serde_json::json!(subject));
fields.insert("relation".into(), serde_json::json!(relation));
fields.insert("object".into(), serde_json::json!(object));
fields.insert("grain_type".into(), serde_json::json!("fact"));
fields.insert("confidence".into(), serde_json::json!(0.9));
fields.insert("created_at_sec".into(), serde_json::json!(created_at_sec));
let mut hash_bytes = [0u8; 32];
let key = format!("{}|{}|{}|{}", subject, relation, object, created_at_sec);
for (i, b) in key.as_bytes().iter().enumerate().take(32) {
hash_bytes[i] = *b;
}
let hash = Hash::from_bytes(&hash_bytes);
let grain = DeserializedGrain {
header: MgHeader {
version: 1,
flags: 0,
grain_type: 0x01,
ns_hash: 0,
created_at_sec,
},
grain_type: GrainType::Fact,
fields,
hash,
};
(hash, grain)
}
#[test]
fn test_group_by_parses() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts GROUP BY subject", &store);
assert!(result.is_ok(), "GROUP BY should parse and execute");
}
#[test]
fn test_group_by_reorders_grains() {
let store = MockStore::with_grains(vec![
make_fact_with_time("bob", "likes", "tea", 100),
make_fact_with_time("john", "likes", "coffee", 200),
make_fact_with_time("bob", "likes", "vim", 300),
make_fact_with_time("john", "likes", "rust", 400),
]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts GROUP BY subject", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
assert_eq!(grains.len(), 4);
let subjects: Vec<&str> = grains
.iter()
.filter_map(|g| g.fields.get("subject").and_then(|v| v.as_str()))
.collect();
assert_eq!(subjects, vec!["bob", "bob", "john", "john"]);
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_group_by_chronological_within_group() {
let store = MockStore::with_grains(vec![
make_fact_with_time("john", "likes", "vim", 500),
make_fact_with_time("john", "likes", "coffee", 100),
make_fact_with_time("john", "likes", "rust", 300),
]);
let ex = CalExecutor::with_defaults();
let result = ex.execute("RECALL facts GROUP BY subject", &store).unwrap();
match result.result {
CalResultPayload::Grains { grains, .. } => {
let times: Vec<u64> = grains
.iter()
.filter_map(|g| g.fields.get("created_at_sec").and_then(|v| v.as_u64()))
.collect();
assert_eq!(
times,
vec![100, 300, 500],
"should be chronological within group"
);
}
other => panic!("expected Grains, got: {:?}", other),
}
}
#[test]
fn test_group_by_sml_format() {
let store = MockStore::with_grains(vec![
make_fact_with_time("sess_1", "user", "hello", 100),
make_fact_with_time("sess_2", "user", "world", 200),
make_fact_with_time("sess_1", "assistant", "hi", 150),
]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts GROUP BY subject FORMAT sml", &store)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "sml");
assert!(text.contains("<group key=\"sess_1\" count=\"2\">"));
assert!(text.contains("<group key=\"sess_2\" count=\"1\">"));
assert!(text.contains("</group>"));
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_group_by_markdown_format() {
let store = MockStore::with_grains(vec![
make_fact_with_time("sess_1", "user", "hello", 100),
make_fact_with_time("sess_2", "user", "world", 200),
]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts GROUP BY subject FORMAT markdown", &store)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "markdown");
assert!(text.contains("### sess_1 (1 memory)"));
assert!(text.contains("### sess_2 (1 memory)"));
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_group_by_text_format() {
let store = MockStore::with_grains(vec![
make_fact_with_time("sess_1", "user", "hello", 100),
make_fact_with_time("sess_1", "assistant", "hi", 150),
make_fact_with_time("sess_2", "user", "world", 200),
]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts GROUP BY subject FORMAT text", &store)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "text");
assert!(text.contains("--- Group 1/2: sess_1 (2 memories) ---"));
assert!(text.contains("--- Group 2/2: sess_2 (1 memory) ---"));
assert!(text.contains("[1]"));
assert!(text.contains("[2]"));
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_group_by_json_format() {
let store = MockStore::with_grains(vec![
make_fact_with_time("sess_1", "user", "hello", 100),
make_fact_with_time("sess_2", "user", "world", 200),
]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts GROUP BY subject FORMAT json", &store)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "json");
assert!(text.contains("\"group_key\""));
assert!(text.contains("\"sess_1\""));
assert!(text.contains("\"sess_2\""));
assert!(text.contains("\"count\""));
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_group_by_missing_field_present_field() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute("RECALL facts GROUP BY relation FORMAT text", &store)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "text");
assert!(text.contains("--- Group 1/1:"));
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_group_by_unknown_field_rejected() {
let store = MockStore::with_grains(vec![make_fact("john", "likes", "coffee")]);
let ex = CalExecutor::with_defaults();
let err = ex
.execute("RECALL facts GROUP BY session_id FORMAT text", &store)
.expect_err("GROUP BY unknown field must be rejected per Bug 9");
assert_eq!(err.code(), "CAL-E060");
}
#[test]
fn test_with_vars_template_substitution() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_a, grain_a)]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
r#"RECALL facts FORMAT TEMPLATE "User: {{$user_name}} | {{subject}} {{relation}} {{object}}" WITH VARS { "user_name": "John" }"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, format, .. } => {
assert_eq!(format, "template");
assert!(
text.contains("User: John |"),
"user var not substituted: {}",
text
);
assert!(text.contains("john"), "subject not substituted: {}", text);
assert!(text.contains("coffee"), "object not substituted: {}", text);
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_with_vars_missing_var_renders_empty() {
let (hash_a, grain_a) = make_fact("john", "likes", "coffee");
let store = MockStore::with_grains(vec![(hash_a, grain_a)]);
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
r#"RECALL facts FORMAT TEMPLATE "{{$missing}} | {{subject}}" WITH VARS { "other": "val" }"#,
&store,
)
.unwrap();
match result.result {
CalResultPayload::Formatted { text, .. } => {
assert!(
text.contains("john"),
"subject should be resolved: {}",
text
);
}
other => panic!("expected Formatted, got: {:?}", other),
}
}
#[test]
fn test_with_vars_parsed_into_query() {
let store = MockStore::empty();
let ex = CalExecutor::with_defaults();
let result = ex
.execute(
r#"RECALL facts WITH VARS { "app": "test", "version": "1.0" }"#,
&store,
)
.unwrap();
assert_eq!(result.metadata.statement_type, "recall");
}
fn make_event_grain_result(content: &str, relation: &str, subject: &str) -> CalGrainResult {
let mut fields = serde_json::Map::new();
fields.insert("subject".into(), serde_json::json!(subject));
fields.insert("relation".into(), serde_json::json!(relation));
fields.insert("content".into(), serde_json::json!(content));
CalGrainResult {
hash: "aabbccdd00112233".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::Value::Object(fields),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
}
}
fn make_event_grain_result_no_relation(content: &str, subject: &str) -> CalGrainResult {
let mut fields = serde_json::Map::new();
fields.insert("subject".into(), serde_json::json!(subject));
fields.insert("content".into(), serde_json::json!(content));
CalGrainResult {
hash: "aabbccdd00112233".into(),
grain_type: "event".into(),
score: 1.0,
fields: serde_json::Value::Object(fields),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
}
}
fn view_of(grain: &CalGrainResult) -> crate::render::GrainView<'_> {
grain_view(grain)
}
#[test]
fn test_render_sml_event_carries_speaker_role() {
let grain = make_event_grain_result("I am a hair stylist.", "user", "s000");
let out = crate::render::render_grain_sml(&view_of(&grain), crate::render::MetadataDetail::Minimal);
assert!(
out.contains(r#"role="user""#),
"SML should carry the speaker as a role attribute, got: {out}"
);
assert!(
out.starts_with("<event"),
"events render as semantic <event> elements, got: {out}"
);
assert!(out.contains("I am a hair stylist."));
}
#[test]
fn test_render_sml_omits_role_attribute_when_absent() {
let grain = make_event_grain_result_no_relation("Hello world", "s000");
let out = crate::render::render_grain_sml(&view_of(&grain), crate::render::MetadataDetail::Minimal);
assert!(
!out.contains("role="),
"SML should not have a role attribute when the field is absent, got: {out}"
);
}
#[test]
fn test_render_sml_omits_role_attribute_when_empty() {
let grain = make_event_grain_result("Hello world", "", "s000");
let out = crate::render::render_grain_sml(&view_of(&grain), crate::render::MetadataDetail::Minimal);
assert!(
!out.contains("role=\"\""),
"SML should not have an empty role attribute, got: {out}"
);
}
#[test]
fn test_render_sml_escapes_role_attribute() {
let grain = make_event_grain_result("test", "user<script>", "s000");
let out = crate::render::render_grain_sml(&view_of(&grain), crate::render::MetadataDetail::Minimal);
assert!(
out.contains("role=\"user<script>\""),
"SML should escape the role attribute value, got: {out}"
);
}
#[test]
fn test_render_markdown_prefixes_content_with_relation() {
let grain = make_event_grain_result("I am a hair stylist.", "user", "s000");
let out = crate::render::render_grain_markdown(&view_of(&grain));
assert!(
out.contains("- **user**: I am a hair stylist."),
"Markdown should prefix content with relation, got: {out}"
);
}
#[test]
fn test_render_markdown_no_relation_names_the_subject() {
let grain = make_event_grain_result_no_relation("Hello world", "s000");
let out = crate::render::render_grain_markdown(&view_of(&grain));
assert!(
out.contains("- **s000**: Hello world"),
"Markdown should name the subject when there is no relation, got: {out}"
);
}
#[test]
fn test_render_markdown_speaker_disambiguation() {
let assistant = make_event_grain_result("I am a school teacher.", "assistant", "s000");
let user = make_event_grain_result("I am a hair stylist.", "user", "s000");
let mut out = String::new();
out.push_str(&crate::render::render_grain_markdown(&view_of(&assistant)));
out.push_str(&crate::render::render_grain_markdown(&view_of(&user)));
assert!(
out.contains("- **assistant**: I am a school teacher."),
"Markdown should show assistant role, got: {out}"
);
assert!(
out.contains("- **user**: I am a hair stylist."),
"Markdown should show user role, got: {out}"
);
}
#[test]
fn test_render_text_event_with_subject_relation_uses_triple_path() {
let grain = make_event_grain_result("I am a hair stylist.", "user", "s000");
let out = crate::render::render_grain_text_line(&view_of(&grain), None);
assert!(
out.contains("s000 user"),
"Text should render subject + relation via triple path, got: {out}"
);
}
#[test]
fn test_render_text_content_only_no_relation() {
let grain = make_event_grain_result_no_relation("Hello world", "");
let out = crate::render::render_grain_text_line(&view_of(&grain), None);
assert_eq!(
out.trim(),
"Hello world",
"Text should render content without prefix when no triple fields"
);
}
#[test]
fn test_render_text_speaker_disambiguation_via_triple() {
let assistant = make_event_grain_result("I am a school teacher.", "assistant", "s000");
let user = make_event_grain_result("I am a hair stylist.", "user", "s000");
let mut out = String::new();
out.push_str(&crate::render::render_grain_text_line(&view_of(&assistant), Some(1)));
out.push_str(&crate::render::render_grain_text_line(&view_of(&user), Some(2)));
assert!(
out.contains("[1] s000 assistant"),
"Text should show assistant relation in triple, got: {out}"
);
assert!(
out.contains("[2] s000 user"),
"Text should show user relation in triple, got: {out}"
);
}
#[test]
fn test_render_sml_fact_states_the_triple_inline() {
let (_, grain) = make_fact("john", "likes", "coffee");
let cgr = CalGrainResult {
hash: grain.hash.to_hex(),
grain_type: "fact".into(),
score: 1.0,
fields: serde_json::to_value(&grain.fields).unwrap(),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let out = crate::render::render_grain_sml(&view_of(&cgr), crate::render::MetadataDetail::Minimal);
assert!(
out.starts_with("<fact") && out.contains("john likes coffee"),
"SML fact should state the triple inline, got: {out}"
);
}
#[test]
fn test_scope_read_allows_recall() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec!["read".to_string()],
..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex.execute("RECALL facts LIMIT 1", &store);
assert!(result.is_ok(), "read scope should allow RECALL");
}
#[test]
fn test_scope_read_blocks_add() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec!["read".to_string()],
..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex.execute(
r#"ADD fact SET subject = "x" SET relation = "y" SET object = "z" REASON "test""#,
&store,
);
assert!(result.is_err(), "read scope should block ADD");
let err = result.unwrap_err();
assert_eq!(err.code(), "CAL-E114");
}
#[test]
fn test_scope_write_allows_add() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec!["read".to_string(), "write".to_string()],
..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex.execute(
r#"ADD fact SET subject = "x" SET relation = "y" SET object = "z" REASON "test""#,
&store,
);
if let Err(ref e) = result {
assert_ne!(e.code(), "CAL-E114", "write scope should not block ADD");
}
}
#[test]
fn test_forget_hash_parses_user_scope_rejected() {
let store = MockStore::empty();
let ex = CalExecutor::new(CalExecutorConfig {
allow_destructive_ops: true,
..Default::default()
});
let hash = format!("sha256:{}", "a".repeat(64));
if let Err(e) = ex.execute(&format!("FORGET {hash}"), &store) {
assert_eq!(e.code(), "CAL-E093", "FORGET <hash> must parse: {e}");
}
let err = ex
.execute("FORGET USER \"test\"", &store)
.expect_err("FORGET USER is not reachable from CAL text");
assert_eq!(err.code(), "CAL-E002");
}
#[test]
fn test_forget_requires_admin_scope_when_scoped() {
let store = MockStore::empty();
let hash = format!("sha256:{}", "a".repeat(64));
let ex = CalExecutor::new(CalExecutorConfig {
caller_scopes: vec!["read".to_string(), "write".to_string()],
allow_destructive_ops: true,
..Default::default()
});
let err = ex
.execute(&format!("FORGET {hash}"), &store)
.expect_err("FORGET needs admin scope");
assert!(
matches!(err, CalError::InsufficientScope { .. }),
"expected InsufficientScope, got {err:?}"
);
}
#[test]
fn test_forget_hash_gated_by_allow_destructive_ops() {
let store = MockStore::empty();
let hash = format!("sha256:{}", "a".repeat(64));
let ex = CalExecutor::new(CalExecutorConfig {
allow_destructive_ops: false,
..Default::default()
});
let res = ex
.execute(&format!("FORGET {hash}"), &store)
.expect("FORGET <hash> parses even when disabled");
match res.result {
CalResultPayload::Unsupported { statement, message } => {
assert_eq!(statement, "forget");
assert!(message.contains("disabled"), "unexpected message: {message}");
}
other => panic!("expected Unsupported when disabled, got {other:?}"),
}
}
#[test]
fn test_admin_forget_still_rejected_at_parse_time() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec!["read".to_string(), "write".to_string(), "admin".to_string()],
allow_destructive_ops: true,
..Default::default()
};
let ex = CalExecutor::new(config);
let err = ex
.execute("FORGET USER \"nonexistent\"", &store)
.expect_err("FORGET rejected unconditionally");
assert_eq!(err.code(), "CAL-E002");
}
#[test]
fn test_scope_empty_no_enforcement() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec![], ..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex.execute(
r#"ADD fact SET subject = "x" SET relation = "y" SET object = "z" REASON "test""#,
&store,
);
if let Err(ref e) = result {
assert_ne!(e.code(), "CAL-E114", "empty scopes should skip enforcement");
}
}
#[test]
fn test_scope_admin_bypasses_all() {
let store = MockStore::empty();
let config = CalExecutorConfig {
caller_scopes: vec!["admin".to_string()],
..Default::default()
};
let ex = CalExecutor::new(config);
let result = ex.execute(
r#"ADD fact SET subject = "x" SET relation = "y" SET object = "z" REASON "test""#,
&store,
);
if let Err(ref e) = result {
assert_ne!(e.code(), "CAL-E114", "admin scope should bypass all checks");
}
}
#[test]
fn test_assemble_post_merge_min_score_keeps_deterministic_grains() {
let ex = exec();
let store = MockStore::empty();
let semantic_low = CalGrainResult {
hash: "11".repeat(32),
grain_type: "fact".into(),
score: 0.10,
fields: serde_json::json!({"subject": "a", "relation": "r", "object": "o"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let semantic_high = CalGrainResult {
hash: "22".repeat(32),
grain_type: "fact".into(),
score: 0.90,
fields: serde_json::json!({"subject": "b", "relation": "r", "object": "o"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: false,
contested_by: None,
};
let deterministic = CalGrainResult {
hash: "33".repeat(32),
grain_type: "workflow".into(),
score: 0.0, fields: serde_json::json!({"name": "boot", "status": "active"}),
score_breakdown: None,
explanation: None,
relative_time: None,
is_deterministic: true,
contested_by: None,
};
let payload = CalResultPayload::Assembled {
grains: vec![
semantic_low.clone(),
semantic_high.clone(),
deterministic.clone(),
],
sources: vec![],
total_tokens: 0,
budget_limit: Some(4000),
progressive: false,
total_available: Some(3),
};
let mut warnings = Vec::new();
let out = ex
.apply_assemble_post_merge_options(
payload,
&[WithOption::MinScore { score: 0.5 }],
&mut warnings,
&store,
"test topic",
)
.unwrap();
match out {
CalResultPayload::Assembled { grains, .. } => {
let kept_hashes: Vec<&str> = grains.iter().map(|g| g.hash.as_str()).collect();
assert!(
kept_hashes.contains(&semantic_high.hash.as_str()),
"high-scoring semantic grain must be retained, got: {:?}",
kept_hashes
);
assert!(
!kept_hashes.contains(&semantic_low.hash.as_str()),
"low-scoring semantic grain must be dropped, got: {:?}",
kept_hashes
);
assert!(
kept_hashes.contains(&deterministic.hash.as_str()),
"deterministic-source grain must NOT be dropped by min_score, got: {:?}",
kept_hashes
);
}
other => panic!("expected Assembled payload, got: {:?}", other),
}
}
#[test]
fn test_execute_recall_marks_no_about_as_deterministic() {
let store = MockStore::with_grains(vec![
make_fact("alice", "knows", "bob"),
make_fact("alice", "knows", "carol"),
]);
let ex = exec();
let recall_stmt = RecallStmt {
grain_type: GrainTypePlural::Facts,
about: None, where_clause: Some(super::super::ast::WhereClause {
condition: super::super::ast::Condition::Comparison {
field: "subject".into(),
comparator: super::super::ast::Comparator::Eq,
value: super::super::ast::Value::String {
value: "alice".into(),
},
span: None,
},
span: None,
}),
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Recall(recall_stmt.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let payload = ex
.execute_recall(&recall_stmt, &store, &query, &mut warnings)
.unwrap();
match payload {
CalResultPayload::Grains { grains, .. } => {
assert!(!grains.is_empty(), "expected at least one grain");
assert!(
grains.iter().all(|g| g.is_deterministic),
"every grain from a no-ABOUT RECALL must be flagged deterministic"
);
}
other => panic!("expected Grains payload, got: {:?}", other),
}
}
#[test]
fn test_execute_recall_with_about_not_deterministic() {
let store = MockStore::with_grains(vec![make_fact("alice", "likes", "coffee")]);
let ex = exec();
let recall_stmt = RecallStmt {
grain_type: GrainTypePlural::Facts,
about: Some(super::super::ast::AboutClause {
text: "coffee preferences".into(),
span: None,
}),
where_clause: None,
recent: None,
since: None,
until: None,
like: None,
between: None,
contradictions: None,
limit: None,
as_format: None,
span: None,
};
let query = CalQuery {
version: super::super::ast::CalVersion(1),
statement: CalStatement::Recall(recall_stmt.clone()),
pipeline: Vec::new(),
with_options: Vec::new(),
format: None,
let_bindings: Vec::new(),
let_values: Default::default(),
user_vars: HashMap::new(),
warnings: Vec::new(),
};
let mut warnings = Vec::new();
let payload = ex
.execute_recall(&recall_stmt, &store, &query, &mut warnings)
.unwrap();
match payload {
CalResultPayload::Grains { grains, .. } => {
assert!(
grains.iter().all(|g| !g.is_deterministic),
"grains from an ABOUT RECALL must NOT be flagged deterministic"
);
}
other => panic!("expected Grains payload, got: {:?}", other),
}
}
}