use std::collections::HashMap;
use std::str::FromStr;
use serde::Serialize;
use uuid::Uuid;
use khive_score::DeterministicScore;
use khive_storage::note::Note;
use khive_storage::types::{
BatchWriteSummary, DeleteMode, DirectedNeighborHit, Direction, EdgeSortField, GraphPath,
LinkId, NeighborHit, NeighborQuery, Page, PageRequest, SortOrder, SqlRow, SqlStatement,
SqlValue, TextFilter, TextQueryMode, TextSearchRequest, TraversalRequest,
};
use khive_storage::{Edge, EdgeRelation, Entity, EntityFilter, Event, EventFilter};
use khive_types::{EdgeEndpointRule, EndpointKind, EventKind, SubstrateKind};
use khive_db::stores::entity::entity_hard_delete_statement;
use khive_db::stores::graph::{edge_hard_delete_statement, purge_incident_edges_statement};
use khive_db::stores::note::note_hard_delete_statement;
use khive_db::SqliteError;
use rusqlite::OptionalExtension;
use crate::atomic_plan::{AffectedRowGuard, DeletePlan, PlanStatement, PostCommitEffect};
use crate::atomic_runner::{run_atomic_unit, AtomicOpFailure, AtomicOpPlan, AtomicRunOutcome};
use crate::curation::{entity_fts_document, note_fts_document};
use crate::error::{GuardedWriteFailure, RuntimeError, RuntimeResult};
use crate::runtime::{KhiveRuntime, NamespaceToken};
#[cfg(test)]
std::thread_local! {
static LINK_FAIL_AFTER: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(any(test, feature = "fault-injection"))]
std::thread_local! {
static VECTOR_FAIL_AFTER: std::cell::Cell<Option<usize>> =
const { std::cell::Cell::new(None) };
}
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_vector_fail_after(n: usize) {
VECTOR_FAIL_AFTER.with(|cell| cell.set(Some(n)));
}
#[cfg(any(test, feature = "fault-injection"))]
static FTS_FAIL_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
static VECTOR_FAIL_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
static FTS_FAIL_MANY_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
static FTS_FAIL_MANY_PARTIAL_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
static FTS_SEARCH_FAIL_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_fts_fail(ns: &str) {
*FTS_FAIL_NS.lock().unwrap() = Some(ns.to_string());
}
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_fts_fail_many(ns: &str) {
*FTS_FAIL_MANY_NS.lock().unwrap() = Some(ns.to_string());
}
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_fts_fail_many_partial(ns: &str) {
*FTS_FAIL_MANY_PARTIAL_NS.lock().unwrap() = Some(ns.to_string());
}
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_fts_search_fail(ns: &str) {
*FTS_SEARCH_FAIL_NS.lock().unwrap() = Some(ns.to_string());
}
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_vector_fail(ns: &str) {
*VECTOR_FAIL_NS.lock().unwrap() = Some(ns.to_string());
}
#[cfg(any(test, feature = "fault-injection"))]
static ROLLBACK_CLEANUP_FAIL_NS: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
#[cfg(any(test, feature = "fault-injection"))]
pub fn arm_rollback_cleanup_fail(ns: &str) {
*ROLLBACK_CLEANUP_FAIL_NS.lock().unwrap() = Some(ns.to_string());
}
#[derive(Clone, Debug)]
pub struct NoteSearchHit {
pub note_id: Uuid,
pub score: DeterministicScore,
pub title: Option<String>,
pub snippet: Option<String>,
}
pub fn hex_prefix_to_uuid_pattern(prefix: &str) -> String {
if prefix.contains('-') {
return prefix.to_string();
}
const BOUNDARIES: [usize; 4] = [8, 13, 18, 23]; let mut out = String::with_capacity(36);
for c in prefix.chars() {
if BOUNDARIES.contains(&out.len()) {
out.push('-');
}
out.push(c);
}
out
}
fn text_preview(text: &str, max_chars: usize) -> Option<String> {
let trimmed = text.trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed.chars().take(max_chars).collect())
}
}
fn normalize_symmetric_direction(
direction: Direction,
relations: Option<&[EdgeRelation]>,
) -> Direction {
let Some(rels) = relations else {
return direction;
};
if rels.is_empty() {
return direction;
}
let all_symmetric = rels
.iter()
.all(|r| matches!(r, EdgeRelation::CompetesWith | EdgeRelation::ComposedWith));
if all_symmetric {
Direction::Both
} else {
direction
}
}
fn direction_sort_rank(direction: &Direction) -> u8 {
match direction {
Direction::Out => 0,
Direction::In => 1,
Direction::Both => 2,
}
}
fn note_title(note: &Note) -> Option<String> {
note.name
.clone()
.filter(|s| !s.trim().is_empty())
.or_else(|| Some(format!("[{}]", note.kind.as_str())))
}
fn note_snippet(note: &Note) -> Option<String> {
text_preview(¬e.content, 200)
}
#[derive(Clone, Debug)]
pub enum Resolved {
Entity(Entity),
Note(Note),
Event(Event),
PackRecord {
pack: String,
kind: String,
data: serde_json::Value,
},
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum EdgeEndpointKind {
Entity,
Note,
Event,
Edge,
}
fn resolved_pair(r: Option<&Resolved>) -> Option<(&'static str, &str, Option<&str>)> {
match r? {
Resolved::Entity(e) => Some(("entity", e.kind.as_str(), e.entity_type.as_deref())),
Resolved::Note(n) => Some(("note", n.kind.as_str(), None)),
Resolved::Event(_) => None,
Resolved::PackRecord { .. } => None,
}
}
pub fn endpoint_matches(
spec: &EndpointKind,
substrate: &str,
kind: &str,
entity_type: Option<&str>,
) -> bool {
match spec {
EndpointKind::EntityOfKind(k) => substrate == "entity" && *k == kind,
EndpointKind::NoteOfKind(k) => substrate == "note" && *k == kind,
EndpointKind::EntityOfType {
kind: k,
entity_type: t,
} => substrate == "entity" && *k == kind && entity_type == Some(*t),
}
}
pub fn accepted_pack_relations_for_entities(
rules: &[EdgeEndpointRule],
src_kind: &str,
src_entity_type: Option<&str>,
tgt_kind: &str,
tgt_entity_type: Option<&str>,
) -> Vec<EdgeRelation> {
let mut relations: Vec<EdgeRelation> = rules
.iter()
.filter(|r| {
endpoint_matches(&r.source, "entity", src_kind, src_entity_type)
&& endpoint_matches(&r.target, "entity", tgt_kind, tgt_entity_type)
})
.map(|r| r.relation)
.collect();
relations.sort_by_key(|r| r.as_str());
relations.dedup();
relations
}
fn pack_rule_allows(
rules: &[EdgeEndpointRule],
relation: EdgeRelation,
src: Option<&Resolved>,
tgt: Option<&Resolved>,
) -> bool {
let Some((src_sub, src_kind, src_type)) = resolved_pair(src) else {
return false;
};
let Some((tgt_sub, tgt_kind, tgt_type)) = resolved_pair(tgt) else {
return false;
};
rules.iter().any(|r| {
r.relation == relation
&& endpoint_matches(&r.source, src_sub, src_kind, src_type)
&& endpoint_matches(&r.target, tgt_sub, tgt_kind, tgt_type)
})
}
pub const BASE_ENTITY_ENDPOINT_RULES: &[(&str, EdgeRelation, &str)] = &[
("concept", EdgeRelation::Contains, "concept"),
("project", EdgeRelation::Contains, "project"),
("project", EdgeRelation::Contains, "artifact"),
("org", EdgeRelation::Contains, "project"),
("org", EdgeRelation::Contains, "service"),
("concept", EdgeRelation::PartOf, "concept"),
("project", EdgeRelation::PartOf, "project"),
("project", EdgeRelation::PartOf, "org"),
("*", EdgeRelation::InstanceOf, "concept"),
("service", EdgeRelation::InstanceOf, "project"),
("concept", EdgeRelation::Extends, "concept"),
("concept", EdgeRelation::VariantOf, "concept"),
("artifact", EdgeRelation::VariantOf, "artifact"),
("concept", EdgeRelation::IntroducedBy, "document"),
("concept", EdgeRelation::IntroducedBy, "person"),
("artifact", EdgeRelation::IntroducedBy, "document"),
("document", EdgeRelation::IntroducedBy, "person"),
("document", EdgeRelation::IntroducedBy, "org"),
("concept", EdgeRelation::IntroducedBy, "org"),
("artifact", EdgeRelation::DerivedFrom, "dataset"),
("artifact", EdgeRelation::DerivedFrom, "document"),
("artifact", EdgeRelation::DerivedFrom, "project"),
("artifact", EdgeRelation::DerivedFrom, "artifact"),
("document", EdgeRelation::Precedes, "document"),
("dataset", EdgeRelation::Precedes, "dataset"),
("artifact", EdgeRelation::Precedes, "artifact"),
("service", EdgeRelation::Precedes, "service"),
("project", EdgeRelation::Precedes, "project"),
("project", EdgeRelation::DependsOn, "project"),
("service", EdgeRelation::DependsOn, "project"),
("service", EdgeRelation::DependsOn, "service"),
("service", EdgeRelation::DependsOn, "artifact"),
("service", EdgeRelation::DependsOn, "dataset"),
("artifact", EdgeRelation::DependsOn, "project"),
("artifact", EdgeRelation::DependsOn, "service"),
("document", EdgeRelation::DependsOn, "document"),
("concept", EdgeRelation::Enables, "concept"),
("service", EdgeRelation::Enables, "concept"),
("dataset", EdgeRelation::Enables, "concept"),
("project", EdgeRelation::Implements, "concept"),
("service", EdgeRelation::Implements, "concept"),
("concept", EdgeRelation::CompetesWith, "concept"),
("project", EdgeRelation::CompetesWith, "project"),
("service", EdgeRelation::CompetesWith, "service"),
("concept", EdgeRelation::ComposedWith, "concept"),
("project", EdgeRelation::ComposedWith, "project"),
("concept", EdgeRelation::Supersedes, "concept"),
("document", EdgeRelation::Supersedes, "document"),
("artifact", EdgeRelation::Supersedes, "artifact"),
("service", EdgeRelation::Supersedes, "service"),
("dataset", EdgeRelation::Supersedes, "dataset"),
("concept", EdgeRelation::Supports, "concept"),
("document", EdgeRelation::Supports, "concept"),
("dataset", EdgeRelation::Supports, "concept"),
("artifact", EdgeRelation::Supports, "concept"),
("concept", EdgeRelation::Refutes, "concept"),
("document", EdgeRelation::Refutes, "concept"),
("dataset", EdgeRelation::Refutes, "concept"),
("artifact", EdgeRelation::Refutes, "concept"),
];
pub fn base_entity_endpoint_rules() -> &'static [(&'static str, EdgeRelation, &'static str)] {
BASE_ENTITY_ENDPOINT_RULES
}
pub fn base_entity_rule_allows(src_kind: &str, relation: EdgeRelation, tgt_kind: &str) -> bool {
BASE_ENTITY_ENDPOINT_RULES.iter().any(|(src, rel, tgt)| {
*rel == relation && (*src == "*" || *src == src_kind) && *tgt == tgt_kind
})
}
pub(crate) fn canonical_edge_endpoints(
relation: EdgeRelation,
source_id: Uuid,
target_id: Uuid,
) -> (Uuid, Uuid) {
if relation.is_symmetric() && target_id < source_id {
(target_id, source_id)
} else {
(source_id, target_id)
}
}
pub(crate) fn infer_dependency_kind(src_kind: &str, tgt_kind: &str) -> Option<&'static str> {
match (src_kind, tgt_kind) {
("project", "project") => Some("build"),
("service", "service") => Some("runtime"),
("service", "dataset") => Some("data"),
("service", "artifact") => Some("artifact"),
("artifact", "project") | ("artifact", "service") => Some("tooling"),
("document", "document") => Some("normative"),
_ => None,
}
}
pub(crate) fn merge_dependency_kind(
src_kind: &str,
tgt_kind: &str,
metadata: Option<serde_json::Value>,
) -> Option<serde_json::Value> {
if let Some(ref m) = metadata {
if m.get("dependency_kind").is_some() {
return metadata;
}
}
let inferred = infer_dependency_kind(src_kind, tgt_kind)?;
let mut obj = metadata.unwrap_or_else(|| serde_json::json!({}));
if let Some(o) = obj.as_object_mut() {
o.insert("dependency_kind".to_string(), serde_json::json!(inferred));
}
Some(obj)
}
pub fn merge_entry_metadata(
metadata: Option<serde_json::Value>,
dependency_kind: Option<String>,
) -> RuntimeResult<Option<serde_json::Value>> {
let Some(dk) = dependency_kind else {
return Ok(metadata);
};
let mut obj = metadata.unwrap_or_else(|| serde_json::json!({}));
let map = obj
.as_object_mut()
.ok_or_else(|| RuntimeError::InvalidInput("metadata must be a JSON object".into()))?;
map.entry("dependency_kind".to_string())
.or_insert_with(|| serde_json::json!(dk));
Ok(Some(obj))
}
const VALID_DEPENDENCY_KINDS: &[&str] = &[
"build",
"runtime",
"data",
"artifact",
"tooling",
"normative",
];
pub(crate) fn validate_edge_weight(weight: f64) -> RuntimeResult<()> {
if !weight.is_finite() || !(0.0..=1.0).contains(&weight) {
return Err(RuntimeError::InvalidInput(format!(
"edge weight must be finite and in [0.0, 1.0], got {weight}"
)));
}
Ok(())
}
pub(crate) fn validate_edge_metadata(
relation: EdgeRelation,
metadata: Option<&serde_json::Value>,
) -> RuntimeResult<()> {
let Some(meta) = metadata else {
return Ok(());
};
if let Some(dk) = meta.get("dependency_kind") {
if relation != EdgeRelation::DependsOn {
return Err(RuntimeError::InvalidInput(format!(
"dependency_kind is only valid on depends_on edges (got {})",
relation.as_str()
)));
}
let dk_str = dk
.as_str()
.ok_or_else(|| RuntimeError::InvalidInput("dependency_kind must be a string".into()))?;
if !VALID_DEPENDENCY_KINDS.contains(&dk_str) {
return Err(RuntimeError::InvalidInput(format!(
"unknown dependency_kind {dk_str:?}; valid: {}",
VALID_DEPENDENCY_KINDS.join(" | ")
)));
}
}
Ok(())
}
fn note_props_match(note_props: Option<&serde_json::Value>, filter: &serde_json::Value) -> bool {
let required = match filter.as_object() {
Some(obj) if !obj.is_empty() => obj,
_ => return true,
};
let actual = match note_props.and_then(serde_json::Value::as_object) {
Some(obj) => obj,
None => return false,
};
required
.iter()
.all(|(k, v)| actual.get(k).is_some_and(|av| av == v))
}
impl KhiveRuntime {
#[allow(clippy::too_many_arguments)]
pub async fn create_entity(
&self,
token: &NamespaceToken,
kind: &str,
entity_type: Option<&str>,
name: &str,
description: Option<&str>,
properties: Option<serde_json::Value>,
tags: Vec<String>,
) -> RuntimeResult<Entity> {
self.validate_entity_kind(kind)?;
crate::secret_gate::check(name)?;
if let Some(d) = description {
crate::secret_gate::check(d)?;
}
if let Some(ref p) = properties {
crate::secret_gate::check_json(p)?;
}
crate::secret_gate::check_tags(&tags)?;
let ns = token.namespace().as_str();
let mut entity = Entity::new(ns, kind, name).with_entity_type(entity_type);
if let Some(d) = description {
entity = entity.with_description(d);
}
if let Some(p) = properties {
entity = entity.with_properties(p);
}
if !tags.is_empty() {
entity = entity.with_tags(tags);
}
self.entities(token)?.upsert_entity(entity.clone()).await?;
let doc = entity_fts_document(&entity);
let embed_body = doc.body.clone();
{
#[cfg(any(test, feature = "fault-injection"))]
let fts_inject = {
let mut g = FTS_FAIL_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let fts_inject = false;
let fts_result: RuntimeResult<()> = if fts_inject {
Err(RuntimeError::Internal("injected FTS failure".to_string()))
} else {
match self.text(token) {
Ok(fts) => fts.upsert_document(doc).await.map_err(RuntimeError::from),
Err(e) => Err(e),
}
};
if let Err(e) = fts_result {
if let Ok(store) = self.entities(token) {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back entity row after FTS failure"
);
}
}
return Err(e);
}
}
let embed_model_names = {
let names = self.registered_embedding_model_names();
if names.is_empty() {
vec![]
} else {
names
}
};
if embed_model_names.len() == 1 {
let model_name = &embed_model_names[0];
let vec_result = self
.embed_document_with_model(model_name, &embed_body)
.await;
#[cfg(any(test, feature = "fault-injection"))]
let vec_inject = {
let mut g = VECTOR_FAIL_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let vec_inject = false;
let vec_result: RuntimeResult<Vec<f32>> = if vec_inject {
Err(RuntimeError::Internal(
"injected vector failure".to_string(),
))
} else {
vec_result
};
let single_result: RuntimeResult<()> = match vec_result {
Ok(vector) => match self.vectors_for_model(token, model_name) {
Ok(vs) => vs
.insert(
entity.id,
SubstrateKind::Entity,
ns,
"entity.body",
vec![vector],
)
.await
.map_err(RuntimeError::from),
Err(e) => Err(e),
},
Err(e) => Err(e),
};
if let Err(e) = single_result {
if let Ok(store) = self.entities(token) {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back entity row after vector failure"
);
}
}
if let Ok(fts) = self.text(token) {
if let Err(ce) = fts.delete_document(ns, entity.id).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back FTS document after vector failure"
);
}
}
return Err(e);
}
} else if !embed_model_names.is_empty() {
let rt_clone = self.clone();
let body_owned = embed_body.clone();
let mut handles = Vec::with_capacity(embed_model_names.len());
for model_name in &embed_model_names {
let rt = rt_clone.clone();
let text = body_owned.clone();
let name = model_name.clone();
handles.push(tokio::spawn(async move {
rt.embed_document_with_model(&name, &text).await
}));
}
let mut vectors: Vec<Vec<f32>> = Vec::with_capacity(embed_model_names.len());
for handle in handles {
let join_result = handle
.await
.map_err(|e| RuntimeError::Internal(format!("embed task panicked: {e}")));
match join_result {
Err(e) => {
if let Ok(store) = self.entities(token) {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await
{
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back entity row after embed task panic"
);
}
}
if let Ok(fts) = self.text(token) {
if let Err(ce) = fts.delete_document(ns, entity.id).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back FTS document after embed task panic"
);
}
}
return Err(e);
}
Ok(Err(e)) => {
if let Ok(store) = self.entities(token) {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await
{
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back entity row after embed failure"
);
}
}
if let Ok(fts) = self.text(token) {
if let Err(ce) = fts.delete_document(ns, entity.id).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back FTS document after embed failure"
);
}
}
return Err(e);
}
Ok(Ok(vec)) => vectors.push(vec),
}
}
let mut inserted_models: Vec<String> = Vec::with_capacity(embed_model_names.len());
for (model_name, vector) in embed_model_names.iter().zip(vectors) {
#[cfg(any(test, feature = "fault-injection"))]
let count_inject = VECTOR_FAIL_AFTER.with(|cell| match cell.get() {
Some(0) => {
cell.set(None);
true
}
Some(n) => {
cell.set(Some(n - 1));
false
}
None => false,
});
#[cfg(not(any(test, feature = "fault-injection")))]
let count_inject = false;
let insert_result = if count_inject {
Err(RuntimeError::Internal(
"injected vector insert failure".to_string(),
))
} else {
match self.vectors_for_model(token, model_name) {
Ok(vs) => vs
.insert(
entity.id,
SubstrateKind::Entity,
ns,
"entity.body",
vec![vector],
)
.await
.map_err(RuntimeError::from),
Err(e) => Err(e),
}
};
if let Err(e) = insert_result {
if let Ok(store) = self.entities(token) {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back entity row after vector insert failure"
);
}
}
if let Ok(fts) = self.text(token) {
if let Err(ce) = fts.delete_document(ns, entity.id).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_entity: failed to roll back FTS document after vector insert failure"
);
}
}
for m in &inserted_models {
if let Ok(vs) = self.vectors_for_model(token, m) {
if let Err(ce) = vs.delete(entity.id).await {
tracing::error!(
error = %ce,
model = m,
id = %entity.id,
"create_entity: failed to roll back vector for model after insert failure"
);
}
}
}
return Err(e);
}
inserted_models.push(model_name.clone());
}
}
Ok(entity)
}
pub async fn get_entity(&self, token: &NamespaceToken, id: Uuid) -> RuntimeResult<Entity> {
self.entities(token)?
.get_entity(id)
.await?
.ok_or_else(|| RuntimeError::NotFound(format!("entity {id}")))
}
pub async fn get_entity_including_deleted(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Entity>> {
self.entities(token)?
.get_entity_including_deleted(id)
.await
.map_err(Into::into)
}
pub async fn get_note_including_deleted(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<khive_storage::note::Note>> {
self.notes(token)?
.get_note_including_deleted(id)
.await
.map_err(Into::into)
}
pub async fn get_entities_by_ids(
&self,
token: &NamespaceToken,
ids: &[Uuid],
) -> RuntimeResult<Vec<Entity>> {
if ids.is_empty() {
return Ok(vec![]);
}
let filter = EntityFilter {
ids: ids.to_vec(),
..Default::default()
};
let page = self
.entities(token)?
.query_entities(
token.namespace().as_str(),
filter,
PageRequest {
offset: 0,
limit: ids.len() as u32,
},
)
.await?;
Ok(page.items)
}
async fn get_entities_by_ids_visible(
&self,
token: &NamespaceToken,
ids: &[Uuid],
) -> RuntimeResult<Vec<Entity>> {
if ids.is_empty() {
return Ok(vec![]);
}
let namespaces: Vec<String> = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
let filter = EntityFilter {
ids: ids.to_vec(),
namespaces,
..Default::default()
};
let page = self
.entities(token)?
.query_entities(
token.namespace().as_str(),
filter,
PageRequest {
offset: 0,
limit: ids.len() as u32,
},
)
.await?;
Ok(page.items)
}
pub(crate) fn ensure_namespace(record_ns: &str, caller_primary_ns: &str) -> RuntimeResult<()> {
if record_ns == caller_primary_ns {
return Ok(());
}
Err(RuntimeError::NotFound("not found in this namespace".into()))
}
pub(crate) fn ensure_namespace_visible(
record_ns: &str,
token: &NamespaceToken,
) -> RuntimeResult<()> {
for ns in token.visible_namespaces() {
if record_ns == ns.as_str() {
return Ok(());
}
}
Err(RuntimeError::NotFound("not found in this namespace".into()))
}
pub async fn list_entities(
&self,
token: &NamespaceToken,
kind: Option<&str>,
entity_type: Option<&str>,
limit: u32,
offset: u32,
) -> RuntimeResult<Vec<Entity>> {
let ns_strs: Vec<String> = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
let filter = EntityFilter {
kinds: match kind {
Some(k) => vec![k.to_string()],
None => vec![],
},
entity_types: match entity_type {
Some(t) => vec![t.to_string()],
None => vec![],
},
namespaces: ns_strs,
..Default::default()
};
let page = self
.entities(token)?
.query_entities(
token.namespace().as_str(),
filter,
PageRequest {
offset: offset.into(),
limit,
},
)
.await?;
Ok(page.items)
}
pub async fn list_entities_tagged(
&self,
token: &NamespaceToken,
kind: Option<&str>,
domain_tag: Option<&str>,
limit: u32,
offset: u32,
) -> RuntimeResult<Vec<Entity>> {
let ns_strs: Vec<String> = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
let filter = EntityFilter {
kinds: match kind {
Some(k) => vec![k.to_string()],
None => vec![],
},
tags_any: match domain_tag {
Some(t) if !t.is_empty() => vec![t.to_string()],
_ => vec![],
},
namespaces: ns_strs,
..Default::default()
};
let page = self
.entities(token)?
.query_entities(
token.namespace().as_str(),
filter,
PageRequest {
offset: offset.into(),
limit,
},
)
.await?;
Ok(page.items)
}
pub async fn count_entities_tagged(
&self,
token: &NamespaceToken,
kind: Option<&str>,
domain_tag: Option<&str>,
) -> RuntimeResult<u64> {
let ns_strs: Vec<String> = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
let filter = EntityFilter {
kinds: match kind {
Some(k) => vec![k.to_string()],
None => vec![],
},
tags_any: match domain_tag {
Some(t) if !t.is_empty() => vec![t.to_string()],
_ => vec![],
},
namespaces: ns_strs,
..Default::default()
};
Ok(self
.entities(token)?
.count_entities(token.namespace().as_str(), filter)
.await?)
}
pub async fn list_events(
&self,
token: &NamespaceToken,
filter: EventFilter,
page: PageRequest,
) -> RuntimeResult<Page<Event>> {
self.events(token)?
.query_events(filter, page)
.await
.map_err(Into::into)
}
pub(crate) async fn validate_edge_relation_endpoints(
&self,
token: &NamespaceToken,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
) -> RuntimeResult<()> {
if source_id == target_id {
return Err(RuntimeError::InvalidInput(
"self-loop edges are not allowed: source_id and target_id must be different".into(),
));
}
if relation == EdgeRelation::Annotates {
match self.resolve_edge_endpoint(token, source_id).await? {
Some(Resolved::Note(_)) => {}
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"annotates source {source_id} must be a note"
)));
}
None => {
if self.get_edge(token, source_id).await?.is_some() {
return Err(RuntimeError::InvalidInput(format!(
"annotates source {source_id} must be a note"
)));
}
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
}
if !self.substrate_exists_by_id(token, target_id).await? {
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
} else if matches!(
relation,
EdgeRelation::Supersedes | EdgeRelation::Supports | EdgeRelation::Refutes
) {
let rel_name = relation.as_str();
let src = match self.resolve_edge_endpoint(token, source_id).await? {
Some(r) => r,
None => {
if self.get_edge(token, source_id).await?.is_some() {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} source {source_id} must be a note or entity (got edge)"
)));
}
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
};
let tgt = match self.resolve_edge_endpoint(token, target_id).await? {
Some(r) => r,
None => {
if self.get_edge(token, target_id).await?.is_some() {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} target {target_id} must be a note or entity (got edge)"
)));
}
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
};
match (&src, &tgt) {
(Resolved::Entity(src_e), Resolved::Entity(tgt_e)) => {
if !base_entity_rule_allows(&src_e.kind, relation, &tgt_e.kind) {
let rule_hint = match relation {
EdgeRelation::Supports | EdgeRelation::Refutes => {
"requires concept|document|dataset|artifact -> concept \
(or same-substrate note -> note)"
}
_ => "requires same-kind entity endpoints",
};
return Err(RuntimeError::InvalidInput(format!(
"({}) -[{rel_name}]-> ({}) is not in the base endpoint \
allowlist; {rel_name} {rule_hint}",
src_e.kind, tgt_e.kind
)));
}
}
(Resolved::Note(_), Resolved::Note(_)) => {}
(Resolved::Event(_), _) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} does not apply to events; source {source_id} is an event"
)));
}
(_, Resolved::Event(_)) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} does not apply to events; target {target_id} is an event"
)));
}
(Resolved::Entity(_), Resolved::Note(_)) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} endpoints must be the same substrate (note→note or entity→entity); \
got source={source_id} (entity) target={target_id} (note)"
)));
}
(Resolved::Note(_), Resolved::Entity(_)) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} endpoints must be the same substrate (note→note or entity→entity); \
got source={source_id} (note) target={target_id} (entity)"
)));
}
(Resolved::PackRecord { .. }, _) | (_, Resolved::PackRecord { .. }) => {
return Err(RuntimeError::InvalidInput(format!(
"pack-private record is not a valid edge endpoint for {rel_name}"
)));
}
}
} else {
let src_res = self.resolve_edge_endpoint(token, source_id).await?;
let tgt_res = self.resolve_edge_endpoint(token, target_id).await?;
if pack_rule_allows(
&self.pack_edge_rules(),
relation,
src_res.as_ref(),
tgt_res.as_ref(),
) {
return Ok(());
}
let src_kind = match src_res {
Some(Resolved::Entity(e)) => e.kind,
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"link source {source_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
None => {
if self.get_edge(token, source_id).await?.is_some() {
return Err(RuntimeError::InvalidInput(format!(
"link source {source_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
};
let tgt_kind = match tgt_res {
Some(Resolved::Entity(e)) => e.kind,
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"link target {target_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
None => {
if self.get_edge(token, target_id).await?.is_some() {
return Err(RuntimeError::InvalidInput(format!(
"link target {target_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
};
if !base_entity_rule_allows(&src_kind, relation, &tgt_kind) {
return Err(RuntimeError::InvalidInput(format!(
"({src_kind}) -[{}]-> ({tgt_kind}) is not in the base endpoint \
allowlist; use pack EDGE_RULES to extend the allowlist",
relation.as_str()
)));
}
}
Ok(())
}
pub async fn validate_link_endpoints(
&self,
token: &NamespaceToken,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
) -> RuntimeResult<()> {
self.validate_edge_relation_endpoints(token, source_id, target_id, relation)
.await
}
pub fn validate_link_endpoints_by_resolved(
&self,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
src: Option<&Resolved>,
tgt: Option<&Resolved>,
) -> RuntimeResult<()> {
if source_id == target_id {
return Err(RuntimeError::InvalidInput(
"self-loop edges are not allowed: source_id and target_id must be different".into(),
));
}
if relation == EdgeRelation::Annotates {
match src {
Some(Resolved::Note(_)) => {}
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"annotates source {source_id} must be a note"
)));
}
None => {
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
}
if tgt.is_none() {
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
return Ok(());
}
if matches!(
relation,
EdgeRelation::Supersedes | EdgeRelation::Supports | EdgeRelation::Refutes
) {
let rel_name = relation.as_str();
let src = src.ok_or_else(|| {
RuntimeError::NotFound(format!("link source {source_id} not found"))
})?;
let tgt = tgt.ok_or_else(|| {
RuntimeError::NotFound(format!("link target {target_id} not found"))
})?;
match (src, tgt) {
(Resolved::Entity(src_e), Resolved::Entity(tgt_e)) => {
if !base_entity_rule_allows(&src_e.kind, relation, &tgt_e.kind) {
let rule_hint = match relation {
EdgeRelation::Supports | EdgeRelation::Refutes => {
"requires concept|document|dataset|artifact -> concept \
(or same-substrate note -> note)"
}
_ => "requires same-kind entity endpoints",
};
return Err(RuntimeError::InvalidInput(format!(
"({}) -[{rel_name}]-> ({}) is not in the base endpoint \
allowlist; {rel_name} {rule_hint}",
src_e.kind, tgt_e.kind
)));
}
}
(Resolved::Note(_), Resolved::Note(_)) => {}
(Resolved::Entity(_), Resolved::Note(_)) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} endpoints must be the same substrate \
(note→note or entity→entity); got source={source_id} (entity) \
target={target_id} (note)"
)));
}
(Resolved::Note(_), Resolved::Entity(_)) => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} endpoints must be the same substrate \
(note→note or entity→entity); got source={source_id} (note) \
target={target_id} (entity)"
)));
}
(Resolved::PackRecord { .. }, _) | (_, Resolved::PackRecord { .. }) => {
return Err(RuntimeError::InvalidInput(format!(
"pack-private record is not a valid edge endpoint for {rel_name}"
)));
}
_ => {
return Err(RuntimeError::InvalidInput(format!(
"{rel_name} endpoints must be notes or entities (not events)"
)));
}
}
return Ok(());
}
if pack_rule_allows(&self.pack_edge_rules(), relation, src, tgt) {
return Ok(());
}
let src_kind = match src {
Some(Resolved::Entity(e)) => &e.kind,
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"link source {source_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
None => {
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
};
let tgt_kind = match tgt {
Some(Resolved::Entity(e)) => &e.kind,
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"link target {target_id} must be an entity for relation {relation:?} \
(only `annotates` crosses substrates)"
)));
}
None => {
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
};
if !base_entity_rule_allows(src_kind, relation, tgt_kind) {
return Err(RuntimeError::InvalidInput(format!(
"({src_kind}) -[{}]-> ({tgt_kind}) is not in the base endpoint \
allowlist; use pack EDGE_RULES to extend the allowlist",
relation.as_str()
)));
}
Ok(())
}
pub fn validate_annotates_endpoint_kinds(
&self,
source_id: Uuid,
target_id: Uuid,
source: Option<EdgeEndpointKind>,
target: Option<EdgeEndpointKind>,
) -> RuntimeResult<()> {
if source_id == target_id {
return Err(RuntimeError::InvalidInput(
"self-loop edges are not allowed: source_id and target_id must be different".into(),
));
}
match source {
Some(EdgeEndpointKind::Note) => {}
Some(_) => {
return Err(RuntimeError::InvalidInput(format!(
"annotates source {source_id} must be a note"
)));
}
None => {
return Err(RuntimeError::NotFound(format!(
"link source {source_id} not found"
)));
}
}
if target.is_none() {
return Err(RuntimeError::NotFound(format!(
"link target {target_id} not found"
)));
}
Ok(())
}
pub async fn link(
&self,
token: &NamespaceToken,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
weight: f64,
metadata: Option<serde_json::Value>,
) -> RuntimeResult<Edge> {
validate_edge_weight(weight)?;
self.validate_edge_relation_endpoints(token, source_id, target_id, relation)
.await?;
let (source_id, target_id) = canonical_edge_endpoints(relation, source_id, target_id);
let metadata = if relation == EdgeRelation::DependsOn {
match (
self.resolve_edge_endpoint(token, source_id).await?,
self.resolve_edge_endpoint(token, target_id).await?,
) {
(Some(Resolved::Entity(src_e)), Some(Resolved::Entity(tgt_e))) => {
merge_dependency_kind(&src_e.kind, &tgt_e.kind, metadata)
}
_ => metadata,
}
} else {
metadata
};
validate_edge_metadata(relation, metadata.as_ref())?;
let now = chrono::Utc::now();
let ns = token.namespace().as_str();
let edge = Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: ns.to_string(),
source_id,
target_id,
relation,
weight,
created_at: now,
updated_at: now,
deleted_at: None,
metadata,
target_backend: None,
};
match self.graph(token)?.upsert_edge_guarded(edge).await? {
khive_storage::GuardedWriteOutcome::Written => {}
khive_storage::GuardedWriteOutcome::Refused(missing) => {
return Err(RuntimeError::GuardedWriteFailed(GuardedWriteFailure {
entry_index: None,
missing_source: missing.source.then_some(source_id),
missing_target: missing.target.then_some(target_id),
}));
}
}
let persisted = self
.list_edges(
token,
crate::curation::EdgeListFilter {
source_id: Some(source_id),
target_id: Some(target_id),
relations: vec![relation],
..Default::default()
},
1,
0,
)
.await?
.into_iter()
.next()
.ok_or_else(|| {
crate::RuntimeError::Internal(format!(
"upsert_edge succeeded but natural-key lookup for ({source_id}, {target_id}, {relation}) returned nothing"
))
})?;
Ok(persisted)
}
#[allow(clippy::too_many_arguments)]
pub async fn link_with_target_backend(
&self,
token: &NamespaceToken,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
weight: f64,
metadata: Option<serde_json::Value>,
target_backend: Option<String>,
) -> RuntimeResult<Edge> {
validate_edge_weight(weight)?;
let (source_id, target_id) = canonical_edge_endpoints(relation, source_id, target_id);
validate_edge_metadata(relation, metadata.as_ref())?;
let now = chrono::Utc::now();
let ns = token.namespace().as_str();
let edge = Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: ns.to_string(),
source_id,
target_id,
relation,
weight,
created_at: now,
updated_at: now,
deleted_at: None,
metadata,
target_backend,
};
self.graph(token)?.upsert_edge(edge).await?;
let persisted = self
.list_edges(
token,
crate::curation::EdgeListFilter {
source_id: Some(source_id),
target_id: Some(target_id),
relations: vec![relation],
..Default::default()
},
1,
0,
)
.await?
.into_iter()
.next()
.ok_or_else(|| {
crate::RuntimeError::Internal(format!(
"upsert_edge succeeded but natural-key lookup for ({source_id}, {target_id}, {relation}) returned nothing"
))
})?;
Ok(persisted)
}
pub(crate) async fn substrate_exists_in_ns(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<bool> {
if self.resolve(token, id).await?.is_some() {
return Ok(true);
}
match self.get_edge_visible(token, id).await {
Ok(Some(_)) => Ok(true),
Ok(None) | Err(RuntimeError::NotFound(_)) => Ok(false),
Err(err) => Err(err),
}
}
pub(crate) async fn substrate_exists_by_id(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<bool> {
if self.resolve_edge_endpoint(token, id).await?.is_some() {
return Ok(true);
}
match self.get_edge(token, id).await {
Ok(Some(_)) => Ok(true),
Ok(None) | Err(RuntimeError::NotFound(_)) => Ok(false),
Err(err) => Err(err),
}
}
pub async fn neighbors(
&self,
token: &NamespaceToken,
node_id: Uuid,
direction: Direction,
limit: Option<u32>,
relations: Option<Vec<EdgeRelation>>,
) -> RuntimeResult<Vec<NeighborHit>> {
self.neighbors_with_query(
token,
node_id,
NeighborQuery {
direction,
relations,
limit,
min_weight: None,
},
)
.await
}
pub async fn neighbors_with_query(
&self,
token: &NamespaceToken,
node_id: Uuid,
mut query: NeighborQuery,
) -> RuntimeResult<Vec<NeighborHit>> {
if !self.substrate_exists_in_ns(token, node_id).await? {
return Ok(Vec::new());
}
query.direction =
normalize_symmetric_direction(query.direction, query.relations.as_deref());
let mut hits = Vec::new();
for ns in token.visible_namespaces() {
let temp = NamespaceToken::for_namespace(ns.clone());
let mut ns_hits = self.graph(&temp)?.neighbors(node_id, query.clone()).await?;
hits.append(&mut ns_hits);
}
hits.sort_by_key(|h| (h.node_id, h.edge_id));
hits.dedup_by_key(|h| (h.node_id, h.edge_id));
self.enrich_neighbor_hits(token, &mut hits).await;
let candidate_ids: Vec<Uuid> = hits.iter().map(|h| h.node_id).collect();
let deleted = self.deleted_entity_ids(candidate_ids).await;
if !deleted.is_empty() {
hits.retain(|h| !deleted.contains(&h.node_id));
}
hits.sort_by(|a, b| {
b.weight
.partial_cmp(&a.weight)
.unwrap_or(std::cmp::Ordering::Equal)
.then(a.node_id.cmp(&b.node_id))
});
Ok(hits)
}
pub async fn neighbors_with_query_directed(
&self,
token: &NamespaceToken,
node_id: Uuid,
query: NeighborQuery,
) -> RuntimeResult<Vec<(NeighborHit, Direction)>> {
if !self.substrate_exists_in_ns(token, node_id).await? {
return Ok(Vec::new());
}
let mut hits: Vec<DirectedNeighborHit> = Vec::new();
for ns in token.visible_namespaces() {
let temp = NamespaceToken::for_namespace(ns.clone());
let mut ns_hits = self
.graph(&temp)?
.neighbors_both_directions(node_id, query.clone())
.await?;
hits.append(&mut ns_hits);
}
hits.sort_by_key(|h| {
(
h.hit.node_id,
h.hit.edge_id,
direction_sort_rank(&h.direction),
)
});
hits.dedup_by_key(|h| {
(
h.hit.node_id,
h.hit.edge_id,
direction_sort_rank(&h.direction),
)
});
let mut plain_hits: Vec<NeighborHit> = hits.iter().map(|h| h.hit.clone()).collect();
self.enrich_neighbor_hits(token, &mut plain_hits).await;
for (dh, enriched) in hits.iter_mut().zip(plain_hits) {
dh.hit = enriched;
}
let candidate_ids: Vec<Uuid> = hits.iter().map(|h| h.hit.node_id).collect();
let deleted = self.deleted_entity_ids(candidate_ids).await;
if !deleted.is_empty() {
hits.retain(|h| !deleted.contains(&h.hit.node_id));
}
hits.sort_by(|a, b| {
b.hit
.weight
.partial_cmp(&a.hit.weight)
.unwrap_or(std::cmp::Ordering::Equal)
.then(a.hit.node_id.cmp(&b.hit.node_id))
});
Ok(hits.into_iter().map(|h| (h.hit, h.direction)).collect())
}
pub async fn traverse(
&self,
token: &NamespaceToken,
request: TraversalRequest,
) -> RuntimeResult<Vec<GraphPath>> {
let mut request = request;
let mut visible_roots = Vec::with_capacity(request.roots.len());
for root in request.roots.drain(..) {
if self.substrate_exists_in_ns(token, root).await? {
visible_roots.push(root);
}
}
request.roots = visible_roots;
if request.roots.is_empty() {
return Ok(Vec::new());
}
let mut paths = Vec::new();
for ns in token.visible_namespaces() {
let temp = NamespaceToken::for_namespace(ns.clone());
let mut ns_paths = self.graph(&temp)?.traverse(request.clone()).await?;
paths.append(&mut ns_paths);
}
self.enrich_path_nodes(token, &mut paths, request.include_properties)
.await;
let all_node_ids: Vec<Uuid> = paths
.iter()
.flat_map(|p| p.nodes.iter().map(|n| n.node_id))
.collect();
let deleted = self.deleted_entity_ids(all_node_ids).await;
if !deleted.is_empty() {
for path in paths.iter_mut() {
path.nodes.retain(|n| !deleted.contains(&n.node_id));
}
paths.retain(|p| !p.nodes.is_empty());
}
Ok(paths)
}
async fn deleted_entity_ids(&self, ids: Vec<Uuid>) -> std::collections::HashSet<Uuid> {
if ids.is_empty() {
return std::collections::HashSet::new();
}
let id_strs: Vec<String> = ids.iter().map(|u| u.to_string()).collect();
let n = id_strs.len();
let entities_placeholders = (0..n)
.map(|i| format!("?{}", i + 1))
.collect::<Vec<_>>()
.join(",");
let notes_placeholders = (0..n)
.map(|i| format!("?{}", n + i + 1))
.collect::<Vec<_>>()
.join(",");
let sql_str = format!(
"SELECT id FROM entities WHERE id IN ({entities_placeholders}) AND deleted_at IS NOT NULL \
UNION \
SELECT id FROM notes WHERE id IN ({notes_placeholders}) AND deleted_at IS NOT NULL"
);
let params: Vec<SqlValue> = id_strs
.iter()
.chain(id_strs.iter())
.cloned()
.map(SqlValue::Text)
.collect();
let stmt = SqlStatement {
sql: sql_str,
params,
label: Some("deleted_entity_ids".into()),
};
let mut out = std::collections::HashSet::new();
let sql = self.sql();
if let Ok(mut reader) = sql.reader().await {
if let Ok(rows) = reader.query_all(stmt).await {
for row in rows {
if let Some(col) = row.columns.first() {
if let SqlValue::Text(s) = &col.value {
if let Ok(u) = s.parse::<Uuid>() {
out.insert(u);
}
}
}
}
}
}
out
}
async fn enrich_neighbor_hits(&self, token: &NamespaceToken, hits: &mut [NeighborHit]) {
if hits.is_empty() {
return;
}
let unique_ids: Vec<Uuid> = {
let mut seen = std::collections::HashSet::new();
hits.iter()
.filter_map(|h| {
if seen.insert(h.node_id) {
Some(h.node_id)
} else {
None
}
})
.collect()
};
let entity_map: HashMap<Uuid, Entity> = self
.get_entities_by_ids_visible(token, &unique_ids)
.await
.unwrap_or_default()
.into_iter()
.map(|e| (e.id, e))
.collect();
let residual_ids: Vec<Uuid> = unique_ids
.iter()
.filter(|id| !entity_map.contains_key(id))
.copied()
.collect();
let note_map: HashMap<Uuid, Note> = if !residual_ids.is_empty() {
if let Ok(store) = self.notes(token) {
store
.get_notes_batch(&residual_ids)
.await
.unwrap_or_default()
.into_iter()
.map(|n| (n.id, n))
.collect()
} else {
HashMap::new()
}
} else {
HashMap::new()
};
for hit in hits.iter_mut() {
if let Some(entity) = entity_map.get(&hit.node_id) {
hit.name = Some(entity.name.clone());
hit.kind = Some(entity.kind.clone());
hit.entity_type = entity.entity_type.clone();
} else if let Some(note) = note_map.get(&hit.node_id) {
let kind = note.kind.clone();
let name = note
.name
.as_deref()
.filter(|s| !s.trim().is_empty())
.map(|s| s.to_owned())
.unwrap_or_else(|| format!("[{kind}]"));
hit.name = Some(name);
hit.kind = Some(kind);
}
}
}
async fn enrich_path_nodes(
&self,
token: &NamespaceToken,
paths: &mut [GraphPath],
include_properties: bool,
) {
if paths.is_empty() {
return;
}
let unique_ids: Vec<Uuid> = {
let mut seen = std::collections::HashSet::new();
paths
.iter()
.flat_map(|p| p.nodes.iter())
.filter_map(|n| {
if seen.insert(n.node_id) {
Some(n.node_id)
} else {
None
}
})
.collect()
};
let entity_map: HashMap<Uuid, Entity> = self
.get_entities_by_ids_visible(token, &unique_ids)
.await
.unwrap_or_default()
.into_iter()
.map(|e| (e.id, e))
.collect();
for path in paths.iter_mut() {
for node in path.nodes.iter_mut() {
if let Some(entity) = entity_map.get(&node.node_id) {
node.name = Some(entity.name.clone());
node.kind = Some(entity.kind.clone());
if include_properties {
node.properties = entity.properties.clone();
}
}
}
}
}
#[allow(clippy::too_many_arguments)]
pub async fn create_note(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
salience: Option<f64>,
properties: Option<serde_json::Value>,
annotates: Vec<Uuid>,
) -> RuntimeResult<Note> {
self.create_note_inner(
token, kind, name, content, None, salience, None, properties, annotates, None,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn create_note_with_embedding_content(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
embedding_content: Option<&str>,
salience: Option<f64>,
properties: Option<serde_json::Value>,
annotates: Vec<Uuid>,
) -> RuntimeResult<Note> {
self.create_note_inner(
token,
kind,
name,
content,
embedding_content,
salience,
None,
properties,
annotates,
None,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn create_note_with_decay(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
salience: Option<f64>,
decay_factor: f64,
properties: Option<serde_json::Value>,
annotates: Vec<Uuid>,
) -> RuntimeResult<Note> {
self.create_note_with_decay_for_embedding_model(
token,
kind,
name,
content,
salience,
decay_factor,
properties,
annotates,
None,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn create_note_with_decay_for_embedding_model(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
salience: Option<f64>,
decay_factor: f64,
properties: Option<serde_json::Value>,
annotates: Vec<Uuid>,
embedding_model: Option<&str>,
) -> RuntimeResult<Note> {
self.create_note_inner(
token,
kind,
name,
content,
None,
salience,
Some(decay_factor),
properties,
annotates,
embedding_model,
)
.await
}
pub async fn try_create_note(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
properties: Option<serde_json::Value>,
) -> RuntimeResult<Option<Note>> {
self.validate_note_kind(kind)?;
crate::secret_gate::check(content)?;
if let Some(n) = name {
crate::secret_gate::check(n)?;
}
if let Some(ref p) = properties {
crate::secret_gate::check_json(p)?;
}
let ns = token.namespace().as_str();
let mut note = Note::new(ns, kind, content);
if let Some(n) = name {
note = note.with_name(n);
}
if let Some(p) = properties {
note = note.with_properties(p);
}
let inserted = self.notes(token)?.try_insert_note(note.clone()).await?;
if !inserted {
return Ok(None);
}
if let Ok(fts) = self.text_for_notes(token) {
if let Err(e) = fts.upsert_document(note_fts_document(¬e)).await {
tracing::warn!(
note_id = %note.id,
error = %e,
"try_create_note: FTS indexing failed (non-fatal)"
);
}
}
let embed_model_names = self.registered_embedding_model_names();
for model_name in &embed_model_names {
match self
.embed_document_with_model(model_name, ¬e.content)
.await
{
Ok(vector) => {
if let Ok(vs) = self.vectors_for_model(token, model_name) {
if let Err(e) = vs
.insert(
note.id,
SubstrateKind::Note,
ns,
"note.content",
vec![vector],
)
.await
{
tracing::warn!(
note_id = %note.id,
model = %model_name,
error = %e,
"try_create_note: vector insert failed (non-fatal)"
);
}
}
}
Err(e) => {
tracing::warn!(
note_id = %note.id,
model = %model_name,
error = %e,
"try_create_note: embedding failed (non-fatal)"
);
}
}
}
Ok(Some(note))
}
#[allow(clippy::too_many_arguments)]
async fn create_note_inner(
&self,
token: &NamespaceToken,
kind: &str,
name: Option<&str>,
content: &str,
embedding_content: Option<&str>,
salience: Option<f64>,
decay_factor: Option<f64>,
properties: Option<serde_json::Value>,
annotates: Vec<Uuid>,
embedding_model: Option<&str>,
) -> RuntimeResult<Note> {
self.validate_note_kind(kind)?;
crate::secret_gate::check(content)?;
if let Some(n) = name {
crate::secret_gate::check(n)?;
}
if let Some(ref p) = properties {
crate::secret_gate::check_json(p)?;
}
if let Some(ec) = embedding_content {
if ec.is_empty() {
return Err(RuntimeError::InvalidInput(
"embedding_content must not be empty".into(),
));
}
if ec.len() >= content.len() || !content.starts_with(ec) {
return Err(RuntimeError::InvalidInput(
"embedding_content must be a proper prefix of content".into(),
));
}
crate::secret_gate::check(ec)?;
}
let ns = token.namespace().as_str();
for &target_id in &annotates {
if !self.substrate_exists_by_id(token, target_id).await? {
return Err(RuntimeError::NotFound(format!(
"create_note annotates target {target_id} not found"
)));
}
}
if let Some(s) = salience {
if !s.is_finite() || !(0.0..=1.0).contains(&s) {
return Err(RuntimeError::InvalidInput(format!(
"salience must be a finite value in [0.0, 1.0]; got {s}"
)));
}
}
if let Some(d) = decay_factor {
if !d.is_finite() || d < 0.0 {
return Err(RuntimeError::InvalidInput(format!(
"decay_factor must be a finite value >= 0.0; got {d}"
)));
}
}
if let Some(model_name) = embedding_model {
self.resolve_embedding_model(Some(model_name))?;
}
let mut note = Note::new(ns, kind, content);
if let Some(s) = salience {
note = note.with_salience(s);
}
if let Some(df) = decay_factor {
note = note.with_decay(df);
}
if let Some(n) = name {
note = note.with_name(n);
}
if let Some(p) = properties {
note = note.with_properties(p);
}
self.notes(token)?.upsert_note(note.clone()).await?;
let embed_model_names: Vec<String> = if let Some(m) = embedding_model {
vec![m.to_string()]
} else {
let names = self.registered_embedding_model_names();
if names.is_empty() {
vec![]
} else {
names
}
};
{
#[cfg(any(test, feature = "fault-injection"))]
let fts_inject = {
let mut g = FTS_FAIL_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let fts_inject = false;
let fts_result: RuntimeResult<()> = if fts_inject {
Err(RuntimeError::Internal("injected FTS failure".to_string()))
} else {
match self.text_for_notes(token) {
Ok(fts) => fts
.upsert_document(note_fts_document(¬e))
.await
.map_err(RuntimeError::from),
Err(e) => Err(e),
}
};
if let Err(e) = fts_result {
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
return Err(e);
}
}
let embed_text: &str = embedding_content.unwrap_or(content);
if embed_model_names.len() == 1 {
let model_name = &embed_model_names[0];
let vec_result = self.embed_document_with_model(model_name, embed_text).await;
#[cfg(any(test, feature = "fault-injection"))]
let vec_inject = {
let ns_inject = {
let mut g = VECTOR_FAIL_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
let count_inject = VECTOR_FAIL_AFTER.with(|cell| match cell.get() {
Some(0) => {
cell.set(None);
true
}
Some(n) => {
cell.set(Some(n - 1));
false
}
None => false,
});
ns_inject || count_inject
};
#[cfg(not(any(test, feature = "fault-injection")))]
let vec_inject = false;
let vec_result: RuntimeResult<Vec<f32>> = if vec_inject {
Err(RuntimeError::Internal(
"injected vector failure".to_string(),
))
} else {
vec_result
};
let single_model_result: RuntimeResult<()> = match vec_result {
Ok(vector) => match self.vectors_for_model(token, model_name) {
Ok(vs) => vs
.insert(
note.id,
SubstrateKind::Note,
ns,
"note.content",
vec![vector],
)
.await
.map_err(RuntimeError::from),
Err(e) => Err(e),
},
Err(e) => Err(e),
};
if let Err(e) = single_model_result {
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
if let Ok(fts) = self.text_for_notes(token) {
let _ = fts.delete_document(ns, note.id).await;
}
return Err(e);
}
} else if !embed_model_names.is_empty() {
let rt_clone = self.clone();
let content_owned = embed_text.to_string();
let mut handles = Vec::with_capacity(embed_model_names.len());
for model_name in &embed_model_names {
let rt = rt_clone.clone();
let text = content_owned.clone();
let name = model_name.clone();
handles.push(tokio::spawn(async move {
rt.embed_document_with_model(&name, &text).await
}));
}
let mut vectors: Vec<Vec<f32>> = Vec::with_capacity(embed_model_names.len());
for handle in handles {
let join_result = handle
.await
.map_err(|e| RuntimeError::Internal(format!("embed task panicked: {e}")));
match join_result {
Err(e) => {
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
if let Ok(fts) = self.text_for_notes(token) {
let _ = fts.delete_document(ns, note.id).await;
}
return Err(e);
}
Ok(Err(e)) => {
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
if let Ok(fts) = self.text_for_notes(token) {
let _ = fts.delete_document(ns, note.id).await;
}
return Err(e);
}
Ok(Ok(vec)) => vectors.push(vec),
}
}
let mut inserted_models: Vec<String> = Vec::with_capacity(embed_model_names.len());
for (model_name, vector) in embed_model_names.iter().zip(vectors) {
let insert_result = match self.vectors_for_model(token, model_name) {
Ok(vs) => vs
.insert(
note.id,
SubstrateKind::Note,
ns,
"note.content",
vec![vector],
)
.await
.map_err(RuntimeError::from),
Err(e) => Err(e),
};
if let Err(e) = insert_result {
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
if let Ok(fts) = self.text_for_notes(token) {
let _ = fts.delete_document(ns, note.id).await;
}
for m in &inserted_models {
if let Ok(vs) = self.vectors_for_model(token, m) {
let _ = vs.delete(note.id).await;
}
}
return Err(e);
}
inserted_models.push(model_name.clone());
}
}
let mut created_edges: Vec<Uuid> = Vec::with_capacity(annotates.len());
#[cfg(test)]
let annotates_iter: Vec<(usize, Uuid)> = annotates
.iter()
.enumerate()
.map(|(i, &id)| (i, id))
.collect();
#[cfg(test)]
macro_rules! next_target {
($pair:expr) => {
$pair.1
};
}
#[cfg(not(test))]
let annotates_iter: Vec<Uuid> = annotates.to_vec();
#[cfg(not(test))]
macro_rules! next_target {
($pair:expr) => {
$pair
};
}
for pair in annotates_iter {
let target_id = next_target!(pair);
#[cfg(test)]
let injected_err: Option<RuntimeError> = {
let call_idx = pair.0;
LINK_FAIL_AFTER.with(|cell| {
let n = cell.get();
if n > 0 && call_idx + 1 == n {
cell.set(0); Some(RuntimeError::Internal("injected link failure".to_string()))
} else {
None
}
})
};
#[cfg(not(test))]
let injected_err: Option<RuntimeError> = None;
let link_result = if let Some(e) = injected_err {
Err(e)
} else {
self.link(
token,
note.id,
target_id,
EdgeRelation::Annotates,
1.0,
None,
)
.await
};
match link_result {
Ok(edge) => created_edges.push(edge.id.into()),
Err(e) => {
for edge_id in created_edges {
let _ = self.delete_edge(token, edge_id, true).await;
}
if let Ok(store) = self.notes(token) {
let _ = store.delete_note(note.id, DeleteMode::Hard).await;
}
if let Ok(fts) = self.text_for_notes(token) {
let _ = fts.delete_document(ns, note.id).await;
}
for model_name in &embed_model_names {
if let Ok(vs) = self.vectors_for_model(token, model_name) {
let _ = vs.delete(note.id).await;
}
}
return Err(e);
}
}
}
Ok(note)
}
pub async fn list_notes(
&self,
token: &NamespaceToken,
kind: Option<&str>,
limit: u32,
offset: u32,
) -> RuntimeResult<Vec<Note>> {
let visible = token.visible_namespaces();
if visible.len() == 1 {
let page = self
.notes(token)?
.query_notes(
token.namespace().as_str(),
kind,
PageRequest {
offset: offset.into(),
limit,
},
)
.await?;
return Ok(page.items);
}
use khive_storage::note::NoteFilter;
let ns_strs: Vec<String> = visible.iter().map(|ns| ns.as_str().to_owned()).collect();
let filter = NoteFilter {
kind: kind.map(|k| k.to_string()),
namespaces: ns_strs,
..Default::default()
};
let page = self
.notes(token)?
.query_notes_filtered(
token.namespace().as_str(),
&filter,
PageRequest {
offset: offset.into(),
limit,
},
)
.await?;
Ok(page.items)
}
#[allow(clippy::too_many_arguments)]
pub async fn search_notes(
&self,
token: &NamespaceToken,
query_text: &str,
query_vector: Option<Vec<f32>>,
limit: u32,
note_kind: Option<&str>,
include_superseded: bool,
tags_any: &[String],
properties_filter: Option<&serde_json::Value>,
) -> RuntimeResult<Vec<NoteSearchHit>> {
const RRF_K: usize = 60;
let candidates = limit.saturating_mul(4).max(limit);
let visible_ns: Vec<String> = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_owned())
.collect();
#[cfg(any(test, feature = "fault-injection"))]
let fts_search_inject = {
let mut g = FTS_SEARCH_FAIL_NS.lock().unwrap();
match g.as_deref() {
Some(armed) if visible_ns.iter().any(|ns| ns == armed) => {
*g = None;
true
}
_ => false,
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let fts_search_inject = false;
let text_search_result = if fts_search_inject {
Err(khive_storage::StorageError::Timeout {
operation: "fts_search".into(),
})
} else {
self.text_for_notes(token)?
.search(TextSearchRequest {
query: query_text.to_string(),
mode: TextQueryMode::Plain,
filter: Some(TextFilter {
namespaces: visible_ns.clone(),
..TextFilter::default()
}),
top_k: candidates,
snippet_chars: 200,
})
.await
};
let text_hits = crate::error::fts_text_leg_or_err(
text_search_result.map_err(RuntimeError::from),
"search_notes",
query_text,
)?;
let vector_hits = if query_vector.is_some() || self.config().embedding_model.is_some() {
self.vector_search(
token,
query_vector,
Some(query_text),
candidates,
Some(SubstrateKind::Note),
)
.await?
} else {
vec![]
};
let fuse_k = text_hits.len() + vector_hits.len();
let fused = crate::fusion::rrf_fuse_k(text_hits, vector_hits, RRF_K, fuse_k)?;
let candidate_ids: Vec<Uuid> = fused.iter().map(|hit| hit.entity_id).collect();
if candidate_ids.is_empty() {
return Ok(vec![]);
}
let note_store = self.notes(token)?;
let mut alive_notes: HashMap<Uuid, Note> = HashMap::new();
for id in &candidate_ids {
if let Some(note) = note_store.get_note(*id).await? {
if note.deleted_at.is_some() {
continue;
}
if let Some(want_kind) = note_kind {
if note.kind != want_kind {
continue;
}
}
if !tags_any.is_empty() {
let note_tags: Vec<String> = note
.properties
.as_ref()
.and_then(|p| p.get("tags"))
.and_then(serde_json::Value::as_array)
.map(|arr| {
arr.iter()
.filter_map(serde_json::Value::as_str)
.map(str::to_owned)
.collect()
})
.unwrap_or_default();
if !note_tags
.iter()
.any(|t| tags_any.iter().any(|w| t.eq_ignore_ascii_case(w)))
{
continue;
}
}
if let Some(pf) = properties_filter {
if !note_props_match(note.properties.as_ref(), pf) {
continue;
}
}
alive_notes.insert(*id, note);
}
}
if !include_superseded && !alive_notes.is_empty() {
let graph = self.graph(token)?;
let mut superseded: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
for ¬e_id in alive_notes.keys() {
let inbound = graph
.neighbors(
note_id,
NeighborQuery {
direction: Direction::In,
relations: Some(vec![EdgeRelation::Supersedes]),
limit: Some(1),
min_weight: None,
},
)
.await?;
if !inbound.is_empty() {
superseded.insert(note_id);
}
}
alive_notes.retain(|id, _| !superseded.contains(id));
}
let mut hits: Vec<NoteSearchHit> = fused
.into_iter()
.filter_map(|hit| {
let note = alive_notes.get(&hit.entity_id)?;
let salience = note.salience.unwrap_or(0.5);
let weight = 0.5 + 0.5 * salience;
let weighted = DeterministicScore::from_f64(hit.score.to_f64() * weight);
Some(NoteSearchHit {
note_id: hit.entity_id,
score: weighted,
title: hit.title.or_else(|| note_title(note)),
snippet: hit.snippet.or_else(|| note_snippet(note)),
})
})
.collect();
hits.sort_by(|a, b| b.score.cmp(&a.score).then(a.note_id.cmp(&b.note_id)));
hits.truncate(limit as usize);
Ok(hits)
}
pub async fn resolve_prefix(
&self,
token: &NamespaceToken,
prefix: &str,
) -> RuntimeResult<Option<Uuid>> {
let namespaces = [token.namespace().as_str().to_owned()];
self.resolve_prefix_inner(Some(&namespaces), prefix, false)
.await
}
pub async fn resolve_prefix_including_deleted(
&self,
token: &NamespaceToken,
prefix: &str,
) -> RuntimeResult<Option<Uuid>> {
let namespaces = [token.namespace().as_str().to_owned()];
self.resolve_prefix_inner(Some(&namespaces), prefix, true)
.await
}
pub async fn resolve_prefix_unfiltered(&self, prefix: &str) -> RuntimeResult<Option<Uuid>> {
self.resolve_prefix_inner(None, prefix, false).await
}
pub async fn resolve_prefix_unfiltered_including_deleted(
&self,
prefix: &str,
) -> RuntimeResult<Option<Uuid>> {
self.resolve_prefix_inner(None, prefix, true).await
}
async fn resolve_prefix_inner(
&self,
namespaces: Option<&[String]>,
prefix: &str,
include_deleted: bool,
) -> RuntimeResult<Option<Uuid>> {
use khive_storage::types::{SqlStatement, SqlValue};
if !prefix.chars().all(|c| c.is_ascii_hexdigit() || c == '-') {
return Ok(None);
}
let pattern = format!("{}%", hex_prefix_to_uuid_pattern(prefix));
let tables = [
("entities", true),
("notes", true),
("events", false),
("graph_edges", false),
];
let ns_clause = namespaces.map(|ns| {
let placeholders: Vec<String> = (0..ns.len()).map(|i| format!("?{}", i + 2)).collect();
format!(" AND namespace IN ({})", placeholders.join(", "))
});
let mut matches: Vec<String> = Vec::new();
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut reader = self.sql().reader().await.map_err(RuntimeError::Storage)?;
for (table, has_deleted_at) in tables {
let deleted_filter = if has_deleted_at && !include_deleted {
" AND deleted_at IS NULL"
} else {
""
};
let mut params = vec![SqlValue::Text(pattern.clone())];
if let Some(ns) = namespaces {
params.extend(ns.iter().map(|n| SqlValue::Text(n.clone())));
}
let sql = SqlStatement {
sql: format!(
"SELECT id FROM {table} WHERE id LIKE ?1{ns_clause}{deleted_filter} LIMIT 2",
ns_clause = ns_clause.as_deref().unwrap_or("")
),
params,
label: Some("resolve_prefix".into()),
};
match reader.query_all(sql).await {
Ok(rows) => {
for row in rows {
if let Some(col) = row.columns.first() {
if let SqlValue::Text(s) = &col.value {
if seen.insert(s.clone()) {
matches.push(s.clone());
}
}
}
}
}
Err(e) => {
let msg = e.to_string();
if msg.contains("no such table") {
continue;
}
return Err(RuntimeError::Storage(e));
}
}
if matches.len() > 1 {
break;
}
}
match matches.len() {
0 => Ok(None),
1 => {
let uuid = Uuid::from_str(&matches[0])
.map_err(|e| RuntimeError::Internal(format!("stored UUID is invalid: {e}")))?;
Ok(Some(uuid))
}
_ => {
let uuids: Vec<uuid::Uuid> = matches
.iter()
.filter_map(|s| Uuid::from_str(s).ok())
.collect();
Err(RuntimeError::AmbiguousPrefix {
prefix: prefix.to_string(),
matches: uuids,
})
}
}
}
pub async fn resolve_by_id(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
if let Some(entity) = self.entities(token)?.get_entity(id).await? {
return Ok(Some(Resolved::Entity(entity)));
}
if let Some(note) = self.notes(token)?.get_note(id).await? {
return Ok(Some(Resolved::Note(note)));
}
Ok(None)
}
pub async fn resolve_by_id_including_deleted(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
if let Some(entity) = self
.entities(token)?
.get_entity_including_deleted(id)
.await?
{
return Ok(Some(Resolved::Entity(entity)));
}
if let Some(note) = self.notes(token)?.get_note_including_deleted(id).await? {
return Ok(Some(Resolved::Note(note)));
}
Ok(None)
}
pub async fn resolve(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
match self.get_entity(token, id).await {
Ok(entity) => return Ok(Some(Resolved::Entity(entity))),
Err(RuntimeError::NotFound(_) | RuntimeError::NamespaceMismatch { .. }) => {}
Err(e) => return Err(e),
}
if let Some(note) = self.notes(token)?.get_note(id).await? {
if Self::ensure_namespace_visible(¬e.namespace, token).is_ok() {
return Ok(Some(Resolved::Note(note)));
}
}
if let Some(event) = self.events(token)?.get_event(id).await? {
if Self::ensure_namespace_visible(&event.namespace, token).is_ok() {
return Ok(Some(Resolved::Event(event)));
}
}
Ok(None)
}
pub async fn resolve_edge_endpoint(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
if let Some(resolved) = self.resolve_by_id(token, id).await? {
return Ok(Some(resolved));
}
if let Some(event) = self.events(token)?.get_event(id).await? {
return Ok(Some(Resolved::Event(event)));
}
Ok(None)
}
pub async fn resolve_primary(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
let ns = token.namespace().as_str();
if let Some(entity) = self.entities(token)?.get_entity(id).await? {
if Self::ensure_namespace(&entity.namespace, ns).is_ok() {
return Ok(Some(Resolved::Entity(entity)));
}
}
if let Some(note) = self.notes(token)?.get_note(id).await? {
if Self::ensure_namespace(¬e.namespace, ns).is_ok() {
return Ok(Some(Resolved::Note(note)));
}
}
if let Some(event) = self.events(token)?.get_event(id).await? {
if Self::ensure_namespace(&event.namespace, ns).is_ok() {
return Ok(Some(Resolved::Event(event)));
}
}
Ok(None)
}
pub async fn resolve_including_deleted(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<Option<Resolved>> {
let ns = token.namespace().as_str();
if let Some(entity) = self
.entities(token)?
.get_entity_including_deleted(id)
.await?
{
if Self::ensure_namespace(&entity.namespace, ns).is_ok() {
return Ok(Some(Resolved::Entity(entity)));
}
}
if let Some(note) = self.notes(token)?.get_note_including_deleted(id).await? {
if Self::ensure_namespace(¬e.namespace, ns).is_ok() {
return Ok(Some(Resolved::Note(note)));
}
}
if let Some(event) = self.events(token)?.get_event(id).await? {
if Self::ensure_namespace(&event.namespace, ns).is_ok() {
return Ok(Some(Resolved::Event(event)));
}
}
Ok(None)
}
async fn atomic_hard_delete_with_edge_purge(
&self,
row_statement: SqlStatement,
node_id: Uuid,
) -> RuntimeResult<bool> {
let plan = AtomicOpPlan::Delete(DeletePlan {
target_id: node_id,
statements: vec![
PlanStatement {
statement: row_statement,
guard: Some(AffectedRowGuard::exactly(1)),
},
PlanStatement {
statement: purge_incident_edges_statement(node_id),
guard: None,
},
],
post_commit: PostCommitEffect::None,
});
match run_atomic_unit(self.sql().as_ref(), vec![plan]).await {
Ok(AtomicRunOutcome::Committed { .. }) => Ok(true),
Ok(AtomicRunOutcome::RolledBack {
failure: AtomicOpFailure::GuardFailed { .. },
..
}) => Ok(false),
Ok(AtomicRunOutcome::RolledBack {
failure: AtomicOpFailure::SqlError { message, .. },
..
}) => Err(RuntimeError::Internal(format!(
"hard delete + edge purge for {node_id} failed: {message}"
))),
Err(e) => Err(RuntimeError::Internal(format!(
"hard delete + edge purge for {node_id}: atomic unit seam failure: {}",
e.0
))),
}
}
pub async fn delete_note(
&self,
token: &NamespaceToken,
id: Uuid,
hard: bool,
) -> RuntimeResult<bool> {
let note_store = self.notes(token)?;
let note = if hard {
match note_store.get_note_including_deleted(id).await? {
Some(n) => n,
None => return Ok(false),
}
} else {
match note_store.get_note(id).await? {
Some(n) => n,
None => return Ok(false),
}
};
let mode = if hard {
DeleteMode::Hard
} else {
DeleteMode::Soft
};
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(¬e.namespace)
.map_err(|e| RuntimeError::Internal(format!("note namespace invalid: {e}")))?,
);
let record_ns = note.namespace.clone();
let deleted = if hard {
let deleted = self
.atomic_hard_delete_with_edge_purge(note_hard_delete_statement(id), id)
.await?;
self.text_for_notes(&record_tok)?
.delete_document(&record_ns, id)
.await?;
for model_name in self.registered_embedding_model_names() {
self.vectors_for_model(&record_tok, &model_name)?
.delete(id)
.await?;
}
deleted
} else {
let deleted = note_store.delete_note(id, mode).await?;
if deleted {
self.text_for_notes(&record_tok)?
.delete_document(&record_ns, id)
.await?;
for model_name in self.registered_embedding_model_names() {
self.vectors_for_model(&record_tok, &model_name)?
.delete(id)
.await?;
}
}
deleted
};
if deleted {
let event_store = self.events(token)?;
let event = khive_storage::event::Event::new(
record_ns.clone(),
"delete",
EventKind::NoteDeleted,
SubstrateKind::Note,
"",
)
.with_target(id)
.with_payload(serde_json::json!({"id": id, "namespace": record_ns, "hard": hard}));
event_store.append_event(event).await.map_err(|e| {
RuntimeError::Internal(format!("delete_note: event store write failed: {e}"))
})?;
self.fire_note_mutation_hook(¬e.kind, id).await;
}
Ok(deleted)
}
pub async fn delete_note_row_first_for_compensation(
&self,
token: &NamespaceToken,
id: Uuid,
) -> RuntimeResult<()> {
let note_store = self.notes(token)?;
let Some(note) = note_store.get_note_including_deleted(id).await? else {
return Ok(());
};
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(¬e.namespace)
.map_err(|e| RuntimeError::Internal(format!("note namespace invalid: {e}")))?,
);
let record_ns = note.namespace.clone();
note_store.delete_note(id, DeleteMode::Hard).await?;
#[cfg(any(test, feature = "fault-injection"))]
{
let armed = ROLLBACK_CLEANUP_FAIL_NS.lock().unwrap().take();
if armed.as_deref() == Some(record_ns.as_str()) {
return Err(RuntimeError::Internal(
"row removed but compensation cleanup failed: injected=true".to_string(),
));
}
}
let mut cleanup_errors = Vec::new();
if let Err(e) = self.graph(&record_tok)?.purge_incident_edges(id).await {
cleanup_errors.push(format!("graph={e}"));
}
if let Err(e) = self
.text_for_notes(&record_tok)?
.delete_document(&record_ns, id)
.await
{
cleanup_errors.push(format!("fts={e}"));
}
for model_name in self.registered_embedding_model_names() {
if let Err(e) = self
.vectors_for_model(&record_tok, &model_name)?
.delete(id)
.await
{
cleanup_errors.push(format!("vector[{model_name}]={e}"));
}
}
if cleanup_errors.is_empty() {
Ok(())
} else {
Err(RuntimeError::Internal(format!(
"row removed but compensation cleanup failed: {}",
cleanup_errors.join("; ")
)))
}
}
}
#[derive(Clone, Debug, Serialize)]
pub struct QueryResult {
pub rows: Vec<SqlRow>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub warnings: Vec<String>,
}
impl KhiveRuntime {
pub async fn query(&self, token: &NamespaceToken, query: &str) -> RuntimeResult<Vec<SqlRow>> {
Ok(self
.query_with_metadata(token, query, khive_query::CompileOptions::default())
.await?
.rows)
}
pub async fn query_with_metadata(
&self,
token: &NamespaceToken,
query: &str,
mut opts: khive_query::CompileOptions,
) -> RuntimeResult<QueryResult> {
use khive_query::QueryValue;
use khive_storage::types::SqlValue;
let ast = khive_query::parse_auto(query)?;
opts.scopes = token
.visible_namespaces()
.iter()
.map(|ns| ns.as_str().to_string())
.collect();
let compiled = khive_query::compile(&ast, &opts)?;
let mut warnings = compiled.warnings;
let truncation_check = compiled.truncation_check;
let params: Vec<SqlValue> = compiled
.params
.into_iter()
.map(|qv| match qv {
QueryValue::Null => SqlValue::Null,
QueryValue::Integer(n) => SqlValue::Integer(n),
QueryValue::Float(f) => SqlValue::Float(f),
QueryValue::Text(s) => SqlValue::Text(s),
QueryValue::Blob(b) => SqlValue::Blob(b),
})
.collect();
let mut reader = self.sql().reader().await?;
let stmt = SqlStatement {
sql: compiled.sql,
params,
label: None,
};
let mut rows = reader.query_all(stmt).await?;
if let Some(check) = truncation_check {
if rows.len() > check.max_limit {
rows.truncate(check.max_limit);
warnings.push(match check.requested_limit {
Some(requested) => format!(
"result set capped at {} rows; requested limit {requested} exceeds the \
cap — use LIMIT/OFFSET to page through the remaining results",
check.max_limit
),
None => format!(
"result set capped at {} rows; more than {} rows matched with no LIMIT \
clause — use LIMIT/OFFSET to page through the remaining results",
check.max_limit, check.max_limit
),
});
}
}
Ok(QueryResult { rows, warnings })
}
pub async fn delete_entity(
&self,
token: &NamespaceToken,
id: Uuid,
hard: bool,
) -> RuntimeResult<bool> {
let entity = if hard {
match self
.entities(token)?
.get_entity_including_deleted(id)
.await?
{
Some(e) => e,
None => return Ok(false),
}
} else {
match self.entities(token)?.get_entity(id).await? {
Some(e) => e,
None => return Ok(false),
}
};
let mode = if hard {
DeleteMode::Hard
} else {
DeleteMode::Soft
};
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(&entity.namespace)
.map_err(|e| RuntimeError::Internal(format!("entity namespace invalid: {e}")))?,
);
let deleted = if hard {
let deleted = self
.atomic_hard_delete_with_edge_purge(entity_hard_delete_statement(id), id)
.await?;
self.remove_from_indexes(&record_tok, id).await?;
deleted
} else {
let deleted = self.entities(token)?.delete_entity(id, mode).await?;
if deleted {
self.remove_from_indexes(&record_tok, id).await?;
}
deleted
};
if deleted {
let event_store = self.events(token)?;
let ns = entity.namespace.clone();
let event = khive_storage::event::Event::new(
ns.clone(),
"delete",
EventKind::EntityDeleted,
SubstrateKind::Entity,
"",
)
.with_target(id)
.with_payload(serde_json::json!({"id": id, "namespace": ns, "hard": hard}));
event_store.append_event(event).await.map_err(|e| {
RuntimeError::Internal(format!("delete_entity: event store write failed: {e}"))
})?;
}
Ok(deleted)
}
pub async fn count_entities(
&self,
token: &NamespaceToken,
kind: Option<&str>,
) -> RuntimeResult<u64> {
let filter = EntityFilter {
kinds: match kind {
Some(k) => vec![k.to_string()],
None => vec![],
},
..Default::default()
};
Ok(self
.entities(token)?
.count_entities(token.namespace().as_str(), filter)
.await?)
}
pub async fn get_edge(
&self,
_token: &NamespaceToken,
edge_id: Uuid,
) -> RuntimeResult<Option<Edge>> {
let mut reader = self.sql().reader().await?;
let record_ns = reader
.query_scalar(SqlStatement {
sql: "SELECT namespace FROM graph_edges \
WHERE id = ?1 AND deleted_at IS NULL LIMIT 1"
.into(),
params: vec![SqlValue::Text(edge_id.to_string())],
label: Some("get_edge_namespace".into()),
})
.await?;
let Some(SqlValue::Text(record_ns)) = record_ns else {
return Ok(None);
};
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(&record_ns)
.map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
);
Ok(self
.graph(&record_tok)?
.get_edge(LinkId::from(edge_id))
.await?)
}
pub async fn get_edge_visible(
&self,
token: &NamespaceToken,
edge_id: Uuid,
) -> RuntimeResult<Option<Edge>> {
self.get_edge(token, edge_id).await
}
pub async fn get_edge_including_deleted(
&self,
_token: &NamespaceToken,
edge_id: Uuid,
) -> RuntimeResult<Option<Edge>> {
let mut reader = self.sql().reader().await?;
let record_ns = reader
.query_scalar(SqlStatement {
sql: "SELECT namespace FROM graph_edges WHERE id = ?1 LIMIT 1".into(),
params: vec![SqlValue::Text(edge_id.to_string())],
label: Some("get_edge_including_deleted_namespace".into()),
})
.await?;
let Some(SqlValue::Text(record_ns)) = record_ns else {
return Ok(None);
};
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(&record_ns)
.map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
);
Ok(self
.graph(&record_tok)?
.get_edge_including_deleted(LinkId::from(edge_id))
.await?)
}
pub const EDGE_LIST_MAX_LIMIT: u32 = 1000;
pub async fn list_edges(
&self,
token: &NamespaceToken,
filter: crate::curation::EdgeListFilter,
limit: u32,
offset: u32,
) -> RuntimeResult<Vec<Edge>> {
let limit = limit.clamp(1, Self::EDGE_LIST_MAX_LIMIT);
let visible = token.visible_namespaces();
if let [ns] = visible {
let temp = NamespaceToken::for_namespace(ns.clone());
let page = self
.graph(&temp)?
.query_edges(
filter.into(),
vec![SortOrder {
field: EdgeSortField::CreatedAt,
direction: khive_storage::types::SortDirection::Asc,
}],
PageRequest {
offset: offset.into(),
limit,
},
)
.await?;
return Ok(page.items);
}
let fetch_limit = offset.saturating_add(limit);
let mut results = Vec::new();
for ns in visible {
let temp = NamespaceToken::for_namespace(ns.clone());
let page = self
.graph(&temp)?
.query_edges(
filter.clone().into(),
vec![SortOrder {
field: EdgeSortField::CreatedAt,
direction: khive_storage::types::SortDirection::Asc,
}],
PageRequest {
offset: 0,
limit: fetch_limit,
},
)
.await?;
results.extend(page.items);
}
results.sort_by_key(|e| Uuid::from(e.id));
results.dedup_by_key(|e| Uuid::from(e.id));
let start = (offset as usize).min(results.len());
let end = (start + limit as usize).min(results.len());
Ok(results[start..end].to_vec())
}
pub async fn list_edges_after(
&self,
token: &NamespaceToken,
filter: crate::curation::EdgeListFilter,
after: Option<Uuid>,
limit: u32,
) -> RuntimeResult<(Vec<Edge>, Option<Uuid>)> {
let limit = limit.clamp(1, Self::EDGE_LIST_MAX_LIMIT);
let visible = token.visible_namespaces();
let limit_usize = limit as usize;
if let [ns] = visible {
let temp = NamespaceToken::for_namespace(ns.clone());
let page = self
.graph(&temp)?
.query_edges_after(filter.into(), after, limit)
.await?;
return Ok((page.items, page.next_after));
}
let probe_limit = limit + 1;
let mut results = Vec::new();
for ns in visible {
let temp = NamespaceToken::for_namespace(ns.clone());
let page = self
.graph(&temp)?
.query_edges_after(filter.clone().into(), after, probe_limit)
.await?;
results.extend(page.items);
}
results.sort_by_key(|e| Uuid::from(e.id));
results.dedup_by_key(|e| Uuid::from(e.id));
let has_more = results.len() > limit_usize;
if has_more {
results.truncate(limit_usize);
}
let next_after = if has_more {
results.last().map(|e| Uuid::from(e.id))
} else {
None
};
Ok((results, next_after))
}
pub async fn count_edges_by_relation(
&self,
token: &NamespaceToken,
) -> RuntimeResult<std::collections::HashMap<String, u64>> {
let mut totals: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
for ns in token.visible_namespaces() {
let temp = NamespaceToken::for_namespace(ns.clone());
for (relation, count) in self.graph(&temp)?.count_edges_by_relation().await? {
*totals.entry(relation.to_string()).or_insert(0) += count;
}
}
Ok(totals)
}
#[allow(clippy::too_many_arguments)]
fn update_edge_symmetric_dml(
conn: &rusqlite::Connection,
ns: &str,
edge_id_str: &str,
canon_src_str: &str,
canon_tgt_str: &str,
relation_str: &str,
weight: f64,
metadata: Option<String>,
target_backend: Option<String>,
) -> Result<Option<String>, SqliteError> {
let now_ts = chrono::Utc::now().timestamp_micros();
let conflict_id: Option<String> = conn
.query_row(
khive_db::stores::graph::EDGE_SYMMETRIC_CONFLICT_PROBE_SQL,
rusqlite::params![
&ns,
&canon_src_str,
&canon_tgt_str,
&relation_str,
&edge_id_str
],
|row| row.get(0),
)
.optional()
.map_err(SqliteError::Rusqlite)?;
if let Some(existing_id) = conflict_id {
conn.execute(
khive_db::stores::graph::EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL,
rusqlite::params![&ns, &edge_id_str],
)
.map_err(SqliteError::Rusqlite)?;
let affected = conn
.execute(
khive_db::stores::graph::EDGE_SYMMETRIC_REFRESH_CANONICAL_SQL,
rusqlite::params![weight, now_ts, target_backend, metadata, &ns, &existing_id],
)
.map_err(SqliteError::Rusqlite)?;
if affected == 0 {
return Err(SqliteError::InvalidData(format!(
"update_edge: surviving canonical row {existing_id} vanished during update"
)));
}
Ok(Some(existing_id))
} else {
let affected = conn
.execute(
khive_db::stores::graph::EDGE_SYMMETRIC_UPDATE_INPLACE_SQL,
rusqlite::params![
&canon_src_str,
&canon_tgt_str,
&relation_str,
weight,
now_ts,
metadata,
&ns,
&edge_id_str,
],
)
.map_err(SqliteError::Rusqlite)?;
if affected == 0 {
return Err(SqliteError::InvalidData(format!(
"update_edge: zero rows affected updating edge {edge_id_str} \
in namespace {ns} — row vanished between fetch and update"
)));
}
Ok(None)
}
}
pub async fn update_edge(
&self,
token: &NamespaceToken,
edge_id: Uuid,
patch: crate::curation::EdgePatch,
) -> RuntimeResult<Edge> {
let graph_for_fetch = self.graph(token)?;
let mut edge = graph_for_fetch
.get_edge(LinkId::from(edge_id))
.await?
.ok_or_else(|| crate::RuntimeError::NotFound(format!("edge {edge_id}")))?;
let record_ns: String = edge.namespace.clone();
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(&record_ns)
.map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
);
let graph = self.graph(&record_tok)?;
let mut changed_fields: Vec<&'static str> = Vec::new();
if let Some(r) = patch.relation {
self.validate_edge_relation_endpoints(&record_tok, edge.source_id, edge.target_id, r)
.await?;
edge.relation = r;
changed_fields.push("relation");
}
if let Some(w) = patch.weight {
if !w.is_finite() || !(0.0..=1.0).contains(&w) {
return Err(RuntimeError::InvalidInput(format!(
"edge weight must be a finite value in [0.0, 1.0]; got {w}"
)));
}
edge.weight = w;
changed_fields.push("weight");
}
if let Some(props) = patch.properties {
edge.metadata = Some(props);
}
let (canon_src, canon_tgt) =
canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
if edge.relation.is_symmetric() {
let ns = record_ns.clone();
let edge_id_str = edge_id.to_string();
let relation_str = edge.relation.to_string();
let canon_src_str = canon_src.to_string();
let canon_tgt_str = canon_tgt.to_string();
let weight = edge.weight;
let metadata = edge
.metadata
.as_ref()
.map(|v| serde_json::to_string(v).unwrap_or_default());
let target_backend = edge.target_backend.clone();
let pool = self.backend().pool_arc();
let writer_task = pool.writer_task_handle().ok().flatten();
let surviving_id: Option<String> = if let Some(writer_task) = writer_task {
writer_task
.send(move |conn| {
Self::update_edge_symmetric_dml(
conn,
&ns,
&edge_id_str,
&canon_src_str,
&canon_tgt_str,
&relation_str,
weight,
metadata,
target_backend,
)
.map_err(|e| {
khive_storage::StorageError::driver(
khive_storage::StorageCapability::Graph,
"update_edge",
e,
)
})
})
.await
.map_err(RuntimeError::Storage)?
} else {
tokio::task::spawn_blocking(move || {
let guard = pool.writer()?;
guard.transaction(|conn| {
Self::update_edge_symmetric_dml(
conn,
&ns,
&edge_id_str,
&canon_src_str,
&canon_tgt_str,
&relation_str,
weight,
metadata,
target_backend,
)
})
})
.await
.map_err(|e| {
RuntimeError::Internal(format!("update_edge: spawn_blocking join: {e}"))
})?
.map_err(RuntimeError::Sqlite)?
};
if let Some(sid) = surviving_id {
let surviving_uuid = Uuid::parse_str(&sid).map_err(|e| {
RuntimeError::Internal(format!("update_edge: surviving id parse failed: {e}"))
})?;
edge = self
.get_edge(&record_tok, surviving_uuid)
.await?
.ok_or_else(|| {
RuntimeError::Internal(format!(
"update_edge: surviving canonical row {surviving_uuid} vanished after update"
))
})?;
} else {
edge.source_id = canon_src;
edge.target_id = canon_tgt;
}
} else {
graph.upsert_edge(edge.clone()).await?;
}
let event_store = self.events(&record_tok)?;
let event = khive_storage::event::Event::new(
record_ns.clone(),
"update",
EventKind::EdgeUpdated,
SubstrateKind::Entity,
"",
)
.with_target(edge_id)
.with_payload(
serde_json::json!({"id": edge_id, "namespace": record_ns, "changed_fields": changed_fields}),
);
event_store.append_event(event).await.map_err(|e| {
RuntimeError::Internal(format!("update_edge: event store write failed: {e}"))
})?;
Ok(edge)
}
pub async fn delete_edge(
&self,
token: &NamespaceToken,
edge_id: Uuid,
hard: bool,
) -> RuntimeResult<bool> {
let mode = if hard {
DeleteMode::Hard
} else {
DeleteMode::Soft
};
let edge = if hard {
self.get_edge_including_deleted(token, edge_id).await?
} else {
self.get_edge(token, edge_id).await?
};
let Some(edge) = edge else {
return Ok(false);
};
let record_ns: String = edge.namespace.clone();
let record_tok = NamespaceToken::for_namespace(
khive_types::Namespace::parse(&record_ns)
.map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
);
let graph = self.graph(&record_tok)?;
let deleted = if hard {
self.atomic_hard_delete_with_edge_purge(edge_hard_delete_statement(edge_id), edge_id)
.await?
} else {
graph.delete_edge(LinkId::from(edge_id), mode).await?
};
if deleted {
let event_store = self.events(&record_tok)?;
let event = khive_storage::event::Event::new(
record_ns.clone(),
"delete",
EventKind::EdgeDeleted,
SubstrateKind::Entity,
"",
)
.with_target(edge_id)
.with_payload(serde_json::json!({"id": edge_id, "namespace": record_ns, "hard": hard}));
event_store.append_event(event).await.map_err(|e| {
RuntimeError::Internal(format!("delete_edge: event store write failed: {e}"))
})?;
}
Ok(deleted)
}
pub async fn count_edges(
&self,
token: &NamespaceToken,
filter: crate::curation::EdgeListFilter,
) -> RuntimeResult<u64> {
Ok(self.graph(token)?.count_edges(filter.into()).await?)
}
pub async fn build_edge(&self, token: &NamespaceToken, spec: &LinkSpec) -> RuntimeResult<Edge> {
let ns_str = match &spec.namespace {
Some(s) => {
let spec_ns = crate::Namespace::parse(s)
.map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}")))?;
if &spec_ns != token.namespace() {
return Err(RuntimeError::InvalidInput(
"LinkSpec namespace does not match token namespace".into(),
));
}
s.as_str()
}
None => token.namespace().as_str(),
};
self.validate_edge_relation_endpoints(token, spec.source_id, spec.target_id, spec.relation)
.await?;
let (source_id, target_id) =
canonical_edge_endpoints(spec.relation, spec.source_id, spec.target_id);
let metadata = if spec.relation == EdgeRelation::DependsOn {
match (
self.resolve_edge_endpoint(token, source_id).await?,
self.resolve_edge_endpoint(token, target_id).await?,
) {
(Some(Resolved::Entity(src_e)), Some(Resolved::Entity(tgt_e))) => {
merge_dependency_kind(&src_e.kind, &tgt_e.kind, spec.metadata.clone())
}
_ => spec.metadata.clone(),
}
} else {
spec.metadata.clone()
};
validate_edge_metadata(spec.relation, metadata.as_ref())?;
let now = chrono::Utc::now();
Ok(Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: ns_str.to_string(),
source_id,
target_id,
relation: spec.relation,
weight: spec.weight,
created_at: now,
updated_at: now,
deleted_at: None,
metadata,
target_backend: None,
})
}
pub async fn link_many(
&self,
token: &NamespaceToken,
specs: Vec<LinkSpec>,
) -> RuntimeResult<Vec<Edge>> {
if specs.is_empty() {
return Ok(vec![]);
}
let mut edges = Vec::with_capacity(specs.len());
for spec in &specs {
edges.push(self.build_edge(token, spec).await?);
}
let outcome = self
.graph(token)?
.upsert_edges_guarded(edges.clone())
.await?;
if let Some(refusal) = outcome.refused {
return Err(RuntimeError::GuardedWriteFailed(GuardedWriteFailure {
entry_index: Some(refusal.entry_index),
missing_source: refusal
.missing
.source
.then_some(edges[refusal.entry_index].source_id),
missing_target: refusal
.missing
.target
.then_some(edges[refusal.entry_index].target_id),
}));
}
if outcome.summary.affected != edges.len() as u64 {
return Err(RuntimeError::NotFound(format!(
"link_many: one or more edge endpoints no longer exist at write time: {}",
outcome.summary.first_error
)));
}
let mut persisted = Vec::with_capacity(edges.len());
for edge in &edges {
let row = self
.list_edges(
token,
crate::curation::EdgeListFilter {
source_id: Some(edge.source_id),
target_id: Some(edge.target_id),
relations: vec![edge.relation],
..Default::default()
},
1,
0,
)
.await?
.into_iter()
.next()
.ok_or_else(|| {
crate::RuntimeError::Internal(format!(
"upsert_edges succeeded but natural-key lookup for ({}, {}, {}) returned nothing",
edge.source_id, edge.target_id, edge.relation.as_str()
))
})?;
persisted.push(row);
}
Ok(persisted)
}
pub async fn create_many(
&self,
token: &NamespaceToken,
specs: Vec<EntityCreateSpec>,
) -> RuntimeResult<Vec<Entity>> {
if specs.is_empty() {
return Ok(vec![]);
}
let ns = token.namespace().as_str();
let mut entities = Vec::with_capacity(specs.len());
for spec in &specs {
self.validate_entity_kind(&spec.kind)?;
let validated_type =
self.validate_entity_type_for_kind(&spec.kind, spec.entity_type.as_deref())?;
if spec.name.trim().is_empty() {
return Err(RuntimeError::InvalidInput("name must not be empty".into()));
}
crate::secret_gate::check(&spec.name)?;
if let Some(d) = &spec.description {
crate::secret_gate::check(d)?;
}
if let Some(ref p) = spec.properties {
crate::secret_gate::check_json(p)?;
}
crate::secret_gate::check_tags(&spec.tags)?;
let mut entity =
Entity::new(ns, &spec.kind, &spec.name).with_entity_type(validated_type.as_deref());
if let Some(d) = &spec.description {
entity = entity.with_description(d);
}
if let Some(p) = spec.properties.clone() {
entity = entity.with_properties(p);
}
if !spec.tags.is_empty() {
entity = entity.with_tags(spec.tags.clone());
}
entities.push(entity);
}
let entity_summary = self
.entities(token)?
.upsert_entities(entities.clone())
.await?;
if entity_summary.failed > 0 {
if let Ok(store) = self.entities(token) {
for entity in &entities {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_many: failed to roll back entity row after partial entity write"
);
}
}
}
return Err(RuntimeError::Internal(format!(
"create_many: {}/{} entity rows failed to write (first error: {}); \
all rows rolled back",
entity_summary.failed, entity_summary.attempted, entity_summary.first_error
)));
}
let docs: Vec<_> = entities.iter().map(entity_fts_document).collect();
#[cfg(any(test, feature = "fault-injection"))]
let fts_many_inject = {
let mut g = FTS_FAIL_MANY_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let fts_many_inject = false;
#[cfg(any(test, feature = "fault-injection"))]
let fts_many_inject_partial = {
let mut g = FTS_FAIL_MANY_PARTIAL_NS.lock().unwrap();
if g.as_deref() == Some(ns) {
*g = None;
true
} else {
false
}
};
#[cfg(not(any(test, feature = "fault-injection")))]
let fts_many_inject_partial = false;
let fts_summary_result: RuntimeResult<BatchWriteSummary> = if fts_many_inject {
Err(RuntimeError::Internal(
"injected FTS failure for create_many".to_string(),
))
} else if fts_many_inject_partial {
Ok(BatchWriteSummary {
attempted: docs.len() as u64,
affected: docs.len().saturating_sub(1) as u64,
failed: 1,
first_error: "injected partial FTS failure for create_many".to_string(),
})
} else {
match self.text(token) {
Ok(fts) => fts.upsert_documents(docs).await.map_err(RuntimeError::from),
Err(e) => Err(e),
}
};
let fts_err: Option<RuntimeError> = match fts_summary_result {
Err(e) => Some(e),
Ok(summary) if summary.failed > 0 => Some(RuntimeError::Internal(format!(
"create_many: {}/{} FTS rows failed to index (first error: {}); \
all rows rolled back",
summary.failed, summary.attempted, summary.first_error
))),
Ok(_) => None,
};
if let Some(e) = fts_err {
if let Ok(fts) = self.text(token) {
for entity in &entities {
if let Err(ce) = fts.delete_document(ns, entity.id).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_many: failed to remove FTS doc during rollback"
);
}
}
}
if let Ok(store) = self.entities(token) {
for entity in &entities {
if let Err(ce) = store.delete_entity(entity.id, DeleteMode::Hard).await {
tracing::error!(
error = %ce,
id = %entity.id,
"create_many: failed to roll back entity row after FTS failure"
);
}
}
}
return Err(e);
}
Ok(entities)
}
}
#[derive(Clone, Debug)]
pub struct LinkSpec {
pub namespace: Option<String>,
pub source_id: Uuid,
pub target_id: Uuid,
pub relation: EdgeRelation,
pub weight: f64,
pub metadata: Option<serde_json::Value>,
}
#[derive(Clone, Debug)]
pub struct EntityCreateSpec {
pub kind: String,
pub entity_type: Option<String>,
pub name: String,
pub description: Option<String>,
pub properties: Option<serde_json::Value>,
pub tags: Vec<String>,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::curation::EdgeListFilter;
use crate::embedder_registry::EmbedderProvider;
use crate::error::RuntimeError;
use crate::runtime::{KhiveRuntime, NamespaceToken};
use crate::{ActorRef, Namespace};
use async_trait::async_trait;
use khive_storage::types::PathNode;
use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
fn rt() -> KhiveRuntime {
KhiveRuntime::memory().unwrap()
}
struct ConstVecService {
dims: usize,
}
#[async_trait]
impl EmbeddingService for ConstVecService {
async fn embed(
&self,
texts: &[String],
_model: EmbeddingModel,
) -> std::result::Result<Vec<Vec<f32>>, EmbedError> {
Ok(texts.iter().map(|_| vec![1.0_f32; self.dims]).collect())
}
fn supports_model(&self, _model: EmbeddingModel) -> bool {
true
}
fn name(&self) -> &'static str {
"const-vec"
}
}
struct ConstVecProvider {
provider_name: String,
dims: usize,
pub build_count: Arc<AtomicUsize>,
}
impl ConstVecProvider {
fn new(name: &str, dims: usize) -> (Self, Arc<AtomicUsize>) {
let counter = Arc::new(AtomicUsize::new(0));
let provider = Self {
provider_name: name.to_owned(),
dims,
build_count: Arc::clone(&counter),
};
(provider, counter)
}
}
#[async_trait]
impl EmbedderProvider for ConstVecProvider {
fn name(&self) -> &str {
&self.provider_name
}
fn dimensions(&self) -> usize {
self.dims
}
async fn build(&self) -> crate::error::RuntimeResult<Arc<dyn EmbeddingService>> {
self.build_count.fetch_add(1, Ordering::SeqCst);
Ok(Arc::new(ConstVecService { dims: self.dims }))
}
}
#[tokio::test]
async fn custom_embedder_only_runtime_fanout_stores_vector() {
const MODEL_NAME: &str = "test-custom-encoder";
const DIMS: usize = 8;
let rt = KhiveRuntime::memory().unwrap();
let (provider, _counter) = ConstVecProvider::new(MODEL_NAME, DIMS);
rt.register_embedder(provider);
assert!(rt.config().embedding_model.is_none());
assert_eq!(rt.registered_embedding_model_names(), vec![MODEL_NAME]);
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"memory",
None,
"custom embedder integration test content",
Some(0.7),
None,
vec![],
)
.await
.expect("create_note with custom-only embedder must succeed");
use khive_storage::types::VectorSearchRequest;
let query_vec = vec![1.0_f32; DIMS];
let hits = rt
.vectors_for_model(&tok, MODEL_NAME)
.expect("vector store for custom model must be accessible")
.search(VectorSearchRequest {
query_vectors: vec![query_vec],
top_k: 5,
namespace: Some(tok.namespace().as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Note),
embedding_model: Some(MODEL_NAME.to_string()),
filter: None,
backend_hints: None,
})
.await
.expect("vector search succeeds");
assert!(
hits.iter().any(|h| h.subject_id == note.id),
"custom embedder must have written a vector for note {}: hits={hits:?}",
note.id
);
}
#[tokio::test]
async fn embed_with_model_accepts_custom_provider_name() {
const MODEL_NAME: &str = "my-custom-enc";
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider, _counter) = ConstVecProvider::new(MODEL_NAME, DIMS);
rt.register_embedder(provider);
let result = rt
.embed_with_model(MODEL_NAME, "hello world")
.await
.expect("embed_with_model must accept custom provider names");
assert_eq!(
result.len(),
DIMS,
"embedding dimension must match provider"
);
assert!(
result.iter().all(|&v| (v - 1.0_f32).abs() < 1e-6),
"ConstVecService must produce all-ones vector; got: {result:?}"
);
}
#[tokio::test]
async fn embed_with_model_rejects_unregistered_name() {
let rt = KhiveRuntime::memory().unwrap();
let result = rt.embed_with_model("nonexistent-model", "hello").await;
assert!(
matches!(result.unwrap_err(), RuntimeError::UnknownModel(ref n) if n == "nonexistent-model"),
"unregistered model name must return UnknownModel"
);
}
#[tokio::test]
async fn no_embeddings_config_registers_zero_embedders() {
let config = crate::config::RuntimeConfig {
db_path: None,
packs: vec!["kg".to_string()],
..crate::config::RuntimeConfig::no_embeddings()
};
let rt = KhiveRuntime::new(config).expect("runtime construction must succeed");
assert!(rt.config().embedding_model.is_none());
assert!(
rt.registered_embedding_model_names().is_empty(),
"no_embeddings() runtime must register zero embedders"
);
}
#[tokio::test]
async fn no_embeddings_runtime_create_note_succeeds_without_model_fanout() {
let config = crate::config::RuntimeConfig {
db_path: None,
packs: vec!["kg".to_string()],
..crate::config::RuntimeConfig::no_embeddings()
};
let rt = KhiveRuntime::new(config).expect("runtime construction must succeed");
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"memory",
None,
"issue-396 regression: model-less remember must succeed",
Some(0.7),
None,
vec![],
)
.await
.expect("create_note must succeed with zero registered embedders");
assert_eq!(
note.content,
"issue-396 regression: model-less remember must succeed"
);
}
#[tokio::test]
async fn update_edge_changes_weight() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
weight: Some(0.5),
..Default::default()
},
)
.await
.unwrap();
assert!((updated.weight - 0.5).abs() < 0.001);
}
#[tokio::test]
async fn update_edge_symmetric_relation_stores_microsecond_updated_at() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::CompetesWith, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let before = chrono::Utc::now();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
weight: Some(0.5),
..Default::default()
},
)
.await
.unwrap();
let drift = (updated.updated_at - before).num_seconds().abs();
assert!(
drift < 60,
"updated_at must round-trip as a recent timestamp (micros, not \
seconds); got {:?}, expected within 60s of {:?}",
updated.updated_at,
before
);
}
#[tokio::test]
async fn update_edge_changes_relation() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
relation: Some(EdgeRelation::VariantOf),
..Default::default()
},
)
.await
.unwrap();
assert_eq!(updated.relation, EdgeRelation::VariantOf);
}
#[tokio::test]
async fn update_edge_annotates_note_to_entity_set_supersedes_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(&tok, "observation", None, "a note", Some(0.5), None, vec![])
.await
.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "E", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, note.id, entity.id, EdgeRelation::Annotates, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let result = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
relation: Some(EdgeRelation::Supersedes),
..Default::default()
},
)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"update to Supersedes on note→entity edge must return InvalidInput, got {result:?}"
);
let fetched = rt.get_edge(&tok, edge_id).await.unwrap().unwrap();
assert_eq!(
fetched.relation,
EdgeRelation::Annotates,
"edge relation must be unchanged after failed update"
);
}
#[tokio::test]
async fn update_edge_entity_to_entity_set_annotates_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let result = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
relation: Some(EdgeRelation::Annotates),
..Default::default()
},
)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"update to Annotates on entity→entity edge must return InvalidInput, got {result:?}"
);
}
#[tokio::test]
async fn update_edge_entity_to_entity_set_supersedes_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
relation: Some(EdgeRelation::Supersedes),
..Default::default()
},
)
.await
.unwrap();
assert_eq!(updated.relation, EdgeRelation::Supersedes);
let fetched = rt.get_edge(&tok, edge_id).await.unwrap().unwrap();
assert_eq!(fetched.relation, EdgeRelation::Supersedes);
}
#[tokio::test]
async fn update_edge_weight_only_skips_validation() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
weight: Some(0.3),
..Default::default()
},
)
.await
.unwrap();
assert_eq!(updated.relation, EdgeRelation::Extends);
assert!((updated.weight - 0.3).abs() < 0.001);
}
#[tokio::test]
async fn update_edge_same_class_relation_change_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let updated = rt
.update_edge(
&tok,
edge_id,
crate::curation::EdgePatch {
relation: Some(EdgeRelation::VariantOf),
..Default::default()
},
)
.await
.unwrap();
assert_eq!(updated.relation, EdgeRelation::VariantOf);
}
#[tokio::test]
async fn list_edges_filters_by_relation() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, a.id, c.id, EdgeRelation::Enables, 1.0, None)
.await
.unwrap();
let filter = EdgeListFilter {
relations: vec![EdgeRelation::Extends],
..Default::default()
};
let edges = rt.list_edges(&tok, filter, 100, 0).await.unwrap();
assert_eq!(edges.len(), 1);
assert_eq!(edges[0].relation, EdgeRelation::Extends);
}
#[tokio::test]
async fn list_edges_filters_by_source() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
let d = rt
.create_entity(&tok, "concept", None, "D", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, c.id, d.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let filter = EdgeListFilter {
source_id: Some(a.id),
..Default::default()
};
let edges = rt.list_edges(&tok, filter, 100, 0).await.unwrap();
assert_eq!(edges.len(), 1);
let src: Uuid = edges[0].source_id;
assert_eq!(src, a.id);
}
#[tokio::test]
async fn list_edges_offset_pages_through_full_set() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
for i in 0..5 {
let t = rt
.create_entity(&tok, "concept", None, &format!("T{i}"), None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, t.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
}
let filter = EdgeListFilter {
source_id: Some(a.id),
relations: vec![EdgeRelation::Extends],
..Default::default()
};
let page0 = rt.list_edges(&tok, filter.clone(), 2, 0).await.unwrap();
let page1 = rt.list_edges(&tok, filter.clone(), 2, 2).await.unwrap();
let page2 = rt.list_edges(&tok, filter.clone(), 2, 4).await.unwrap();
assert_eq!(page0.len(), 2);
assert_eq!(page1.len(), 2);
assert_eq!(page2.len(), 1);
let ids = |p: &[Edge]| p.iter().map(|e| Uuid::from(e.id)).collect::<Vec<_>>();
assert_ne!(ids(&page0), ids(&page1), "page 2 must differ from page 1");
let mut all_ids: Vec<Uuid> = ids(&page0)
.into_iter()
.chain(ids(&page1))
.chain(ids(&page2))
.collect();
all_ids.sort();
all_ids.dedup();
assert_eq!(all_ids.len(), 5, "pages must tile the full edge set");
let empty = rt.list_edges(&tok, filter.clone(), 2, 100).await.unwrap();
assert!(
empty.is_empty(),
"offset past the end must return empty, not page 1"
);
}
#[tokio::test]
async fn list_edges_after_keyset_tiles_full_set() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
for i in 0..5 {
let t = rt
.create_entity(&tok, "concept", None, &format!("K{i}"), None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, t.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
}
let filter = EdgeListFilter {
source_id: Some(a.id),
relations: vec![EdgeRelation::Extends],
..Default::default()
};
let mut seen = Vec::new();
let mut cursor: Option<Uuid> = None;
for _ in 0..20 {
let (page, next) = rt
.list_edges_after(&tok, filter.clone(), cursor, 2)
.await
.unwrap();
if page.is_empty() {
break;
}
seen.extend(page.iter().map(|e| Uuid::from(e.id)));
if next.is_none() {
break;
}
cursor = next;
}
seen.sort();
seen.dedup();
assert_eq!(seen.len(), 5, "keyset walk must tile the full edge set");
let (first_a, next_a) = rt
.list_edges_after(&tok, filter.clone(), None, 2)
.await
.unwrap();
let (first_b, next_b) = rt
.list_edges_after(&tok, filter.clone(), None, 2)
.await
.unwrap();
assert_eq!(
first_a.iter().map(|e| e.id.0).collect::<Vec<_>>(),
first_b.iter().map(|e| e.id.0).collect::<Vec<_>>(),
);
assert_eq!(next_a, next_b);
}
#[tokio::test]
async fn list_edges_after_single_namespace_exact_final_page_has_no_next_after() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "SingleCursorA", None, None, vec![])
.await
.unwrap();
for i in 0..4 {
let t = rt
.create_entity(
&tok,
"concept",
None,
&format!("SingleCursorT{i}"),
None,
None,
vec![],
)
.await
.unwrap();
rt.link(&tok, a.id, t.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
}
let filter = EdgeListFilter {
source_id: Some(a.id),
relations: vec![EdgeRelation::Extends],
..Default::default()
};
let (page1, next1) = rt
.list_edges_after(&tok, filter.clone(), None, 2)
.await
.unwrap();
assert_eq!(page1.len(), 2);
let cursor = next1.expect("first page must report a cursor when two rows remain");
let (page2, next2) = rt
.list_edges_after(&tok, filter, Some(cursor), 2)
.await
.unwrap();
assert_eq!(page2.len(), 2);
assert_eq!(
next2, None,
"an exact-size final single-namespace page must not report a cursor"
);
}
#[tokio::test]
async fn list_edges_after_multi_namespace_exact_final_page_has_no_next_after() {
let rt = rt();
let ns_a = Namespace::parse("cursor-ns-a").unwrap();
let ns_b = Namespace::parse("cursor-ns-b").unwrap();
let tok_a = NamespaceToken::for_namespace(ns_a.clone());
let tok_b = NamespaceToken::for_namespace(ns_b.clone());
let visible = NamespaceToken::mint_with_visibility(ns_a, vec![ns_b], ActorRef::anonymous());
for (tok, prefix) in [(&tok_a, "A"), (&tok_b, "B")] {
let source = rt
.create_entity(
tok,
"concept",
None,
&format!("MultiCursor{prefix}Source"),
None,
None,
vec![],
)
.await
.unwrap();
for i in 0..2 {
let target = rt
.create_entity(
tok,
"concept",
None,
&format!("MultiCursor{prefix}Target{i}"),
None,
None,
vec![],
)
.await
.unwrap();
rt.link(tok, source.id, target.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
}
}
let filter = EdgeListFilter {
relations: vec![EdgeRelation::Extends],
..Default::default()
};
let (page1, next1) = rt
.list_edges_after(&visible, filter.clone(), None, 2)
.await
.unwrap();
assert_eq!(page1.len(), 2);
let cursor = next1.expect("first merged page must report a cursor when rows remain");
let (page2, next2) = rt
.list_edges_after(&visible, filter, Some(cursor), 2)
.await
.unwrap();
assert_eq!(page2.len(), 2);
assert_eq!(
next2, None,
"an exact-size final multi-namespace page must not report a cursor"
);
}
#[tokio::test]
async fn count_edges_by_relation_matches_fixtures() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, a.id, c.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, b.id, c.id, EdgeRelation::Enables, 1.0, None)
.await
.unwrap();
let counts = rt.count_edges_by_relation(&tok).await.unwrap();
assert_eq!(counts.get("extends").copied(), Some(2));
assert_eq!(counts.get("enables").copied(), Some(1));
}
#[tokio::test]
async fn delete_edge_removes_from_storage() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_id: Uuid = edge.id.into();
let deleted = rt.delete_edge(&tok, edge_id, true).await.unwrap();
assert!(deleted);
let fetched = rt.get_edge(&tok, edge_id).await.unwrap();
assert!(fetched.is_none(), "edge should be gone after delete");
}
#[tokio::test]
async fn count_edges_matches_filter() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, a.id, c.id, EdgeRelation::Enables, 1.0, None)
.await
.unwrap();
let all = rt
.count_edges(&tok, EdgeListFilter::default())
.await
.unwrap();
assert_eq!(all, 2);
let just_extends = rt
.count_edges(
&tok,
EdgeListFilter {
relations: vec![EdgeRelation::Extends],
..Default::default()
},
)
.await
.unwrap();
assert_eq!(just_extends, 1);
}
#[tokio::test]
async fn edge_in_visible_namespace_reachable_as_graph_root() {
let rt = rt();
let ns_a = Namespace::parse("vis-edge-a").unwrap();
let ns_b = Namespace::parse("vis-edge-b").unwrap();
let tok_b = NamespaceToken::for_namespace(ns_b.clone());
let src = rt
.create_entity(&tok_b, "concept", None, "SrcB", None, None, vec![])
.await
.unwrap();
let tgt = rt
.create_entity(&tok_b, "concept", None, "TgtB", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok_b, src.id, tgt.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let tok_a_vis = rt
.authorize_with_visibility(ns_a.clone(), vec![ns_b.clone()])
.unwrap();
let got = rt.get_edge_visible(&tok_a_vis, edge.id.0).await.unwrap();
assert!(
got.is_some(),
"edge in visible namespace must be retrievable via get_edge_visible"
);
let neighbors = rt
.neighbors(&tok_a_vis, src.id, Direction::Out, Some(16), None)
.await
.unwrap();
assert!(
neighbors.iter().any(|h| h.node_id == tgt.id),
"neighbors of visible-ns node must include its visible-ns neighbor; got: {neighbors:?}"
);
}
#[tokio::test]
async fn get_entity_cross_namespace_no_longer_denied() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let entity = rt
.create_entity(&ns_a, "concept", None, "Alpha", None, None, vec![])
.await
.unwrap();
let found = rt.get_entity(&ns_a, entity.id).await;
assert!(found.is_ok(), "same-namespace get must succeed");
let cross = rt.get_entity(&ns_b, entity.id).await;
assert!(
cross.is_ok(),
"cross-namespace get must succeed in shared-brain OSS (ADR-007 rule 2)"
);
assert_eq!(cross.unwrap().id, entity.id);
}
#[tokio::test]
async fn delete_entity_cross_namespace_no_longer_denied() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let entity = rt
.create_entity(&ns_a, "concept", None, "Beta", None, None, vec![])
.await
.unwrap();
let cross_ns_result = rt.delete_entity(&ns_b, entity.id, true).await;
assert!(
cross_ns_result.is_ok(),
"cross-namespace delete must succeed in shared-brain OSS; got {:?}",
cross_ns_result
);
assert!(cross_ns_result.unwrap(), "delete must return true");
let gone = rt.get_entity(&ns_a, entity.id).await;
assert!(gone.is_err(), "entity must be gone after delete");
}
#[tokio::test]
async fn create_note_indexes_into_fts5() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"FlashAttention reduces memory by using tiling",
Some(0.8),
None,
vec![],
)
.await
.unwrap();
let ns = tok.namespace().as_str().to_string();
let hits = rt
.text_for_notes(&tok)
.unwrap()
.search(khive_storage::types::TextSearchRequest {
query: "FlashAttention".to_string(),
mode: khive_storage::types::TextQueryMode::Plain,
filter: Some(khive_storage::types::TextFilter {
namespaces: vec![ns],
..Default::default()
}),
top_k: 10,
snippet_chars: 100,
})
.await
.unwrap();
assert!(
hits.iter().any(|h| h.subject_id == note.id),
"note should be indexed in FTS5 after create"
);
}
#[tokio::test]
async fn search_notes_with_residual_fts5_char_fails_loud() {
let rt = rt();
let tok = NamespaceToken::local();
rt.create_note(
&tok,
"observation",
None,
"use foo@bar to chain calls",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let result = rt
.search_notes(&tok, "foo@bar", None, 10, None, false, &[], None)
.await;
assert!(
result.is_err(),
"#569 search_notes must fail loud when the FTS leg errors on a residual \
FTS5 char ('@'), not silently degrade to vector-only fusion, got: {:?}",
result.ok()
);
assert!(
matches!(result.unwrap_err(), RuntimeError::InvalidInput(_)),
"residual FTS5 parser failure must surface as RuntimeError::InvalidInput"
);
}
#[tokio::test]
async fn search_notes_propagates_non_parser_fts_error() {
let rt = rt();
let tok = NamespaceToken::local();
rt.create_note(
&tok,
"observation",
None,
"FlashAttention reduces memory by using tiling",
Some(0.8),
None,
vec![],
)
.await
.unwrap();
let ns = tok.namespace().as_str().to_string();
arm_fts_search_fail(&ns);
let result = rt
.search_notes(&tok, "FlashAttention", None, 10, None, false, &[], None)
.await;
assert!(
result.is_err(),
"search_notes must propagate a non-parser FTS StorageError (Timeout) \
as Err, not silently degrade it to an empty result, got: {:?}",
result.ok()
);
assert!(
matches!(
result.unwrap_err(),
RuntimeError::Storage(khive_storage::StorageError::Timeout { .. })
),
"propagated error must be the injected StorageError::Timeout, unwrapped \
through RuntimeError::Storage"
);
}
#[tokio::test]
async fn create_note_with_properties() {
let rt = rt();
let tok = NamespaceToken::local();
let props = serde_json::json!({"source": "arxiv:2205.14135"});
let note = rt
.create_note(
&tok,
"insight",
None,
"FlashAttention is IO-aware",
Some(0.9),
Some(props.clone()),
vec![],
)
.await
.unwrap();
assert_eq!(note.properties.as_ref().unwrap(), &props);
}
#[tokio::test]
async fn create_note_creates_annotates_edges() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "FlashAttention", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"FlashAttention uses SRAM tiling for memory efficiency",
Some(0.9),
None,
vec![entity.id],
)
.await
.unwrap();
let out_neighbors = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(out_neighbors.len(), 1);
assert_eq!(out_neighbors[0].node_id, entity.id);
assert_eq!(out_neighbors[0].relation, EdgeRelation::Annotates);
let in_neighbors = rt
.neighbors(
&tok,
entity.id,
Direction::In,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(in_neighbors.len(), 1);
assert_eq!(in_neighbors[0].node_id, note.id);
}
#[tokio::test]
async fn neighbors_without_relation_filter_returns_all() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, a.id, c.id, EdgeRelation::Enables, 1.0, None)
.await
.unwrap();
let all = rt
.neighbors(&tok, a.id, Direction::Out, None, None)
.await
.unwrap();
assert_eq!(all.len(), 2);
}
#[tokio::test]
async fn neighbors_with_relation_filter_returns_subset() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
rt.link(&tok, a.id, c.id, EdgeRelation::Enables, 1.0, None)
.await
.unwrap();
let filtered = rt
.neighbors(
&tok,
a.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Extends]),
)
.await
.unwrap();
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].node_id, b.id);
assert_eq!(filtered[0].relation, EdgeRelation::Extends);
}
#[tokio::test]
async fn neighbors_with_query_directed_preserves_self_loop_direction_parity() {
let rt = rt();
let tok = NamespaceToken::local();
let centre = rt
.create_entity(&tok, "concept", None, "Centre", None, None, vec![])
.await
.unwrap();
let now = chrono::Utc::now();
rt.graph(&tok)
.unwrap()
.upsert_edge(Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: "local".to_string(),
source_id: centre.id,
target_id: centre.id,
relation: EdgeRelation::Extends,
weight: 0.7,
created_at: now,
updated_at: now,
deleted_at: None,
metadata: None,
target_backend: None,
})
.await
.unwrap();
let directed = rt
.neighbors_with_query_directed(
&tok,
centre.id,
NeighborQuery {
direction: Direction::Both,
relations: None,
limit: None,
min_weight: None,
},
)
.await
.unwrap();
assert_eq!(
directed.len(),
2,
"a self-loop edge must produce both an Out hit and an In hit, not one collapsed hit"
);
let directions: Vec<Direction> = directed.iter().map(|(_, d)| d.clone()).collect();
assert!(
directions.contains(&Direction::Out),
"self-loop must retain its Out-tagged hit"
);
assert!(
directions.contains(&Direction::In),
"self-loop must retain its In-tagged hit"
);
}
#[tokio::test]
async fn search_notes_returns_relevant_note() {
let rt = rt();
let tok = NamespaceToken::local();
rt.create_note(
&tok,
"observation",
None,
"GQA reduces KV cache memory for large models",
Some(0.8),
None,
vec![],
)
.await
.unwrap();
let results = rt
.search_notes(&tok, "GQA KV cache", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(!results.is_empty(), "search should return the indexed note");
let hit = &results[0];
assert!(
hit.title.is_some(),
"note hit title should be populated (falls back to content)"
);
assert!(
hit.snippet.is_some(),
"note hit snippet should be populated"
);
}
#[tokio::test]
async fn search_notes_excludes_soft_deleted() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"RoPE positional encoding rotary embeddings",
Some(0.7),
None,
vec![],
)
.await
.unwrap();
rt.notes(&tok)
.unwrap()
.delete_note(note.id, DeleteMode::Soft)
.await
.unwrap();
let results = rt
.search_notes(
&tok,
"RoPE rotary positional",
None,
10,
None,
false,
&[],
None,
)
.await
.unwrap();
assert!(
results.iter().all(|h| h.note_id != note.id),
"soft-deleted note should be excluded from search"
);
}
#[tokio::test]
async fn search_notes_tag_filter_pushed_before_truncation() {
let rt = rt();
let tok = NamespaceToken::local();
rt.create_note(
&tok,
"observation",
None,
"kappa lambda mu note decoy kappa lambda mu note decoy kappa lambda mu",
Some(0.5),
Some(serde_json::json!({"tags": ["other-note-tag"]})),
vec![],
)
.await
.unwrap();
let target = rt
.create_note(
&tok,
"observation",
None,
"kappa lambda mu note target",
Some(0.5),
Some(serde_json::json!({"tags": ["note-target-tag"]})),
vec![],
)
.await
.unwrap();
let hits = rt
.search_notes(
&tok,
"kappa lambda mu note",
None,
1,
None,
false,
&["note-target-tag".to_string()],
None,
)
.await
.unwrap();
assert_eq!(
hits.len(),
1,
"exactly one hit expected (tag-matching note)"
);
assert_eq!(
hits[0].note_id, target.id,
"tag-filtered note must be returned even when ranked below limit in raw fusion"
);
}
#[tokio::test]
async fn search_notes_props_filter_pushed_before_truncation() {
let rt = rt();
let tok = NamespaceToken::local();
rt.create_note(
&tok,
"observation",
None,
"nu xi omicron note decoy nu xi omicron note decoy nu xi omicron",
Some(0.5),
Some(serde_json::json!({"source": "other"})),
vec![],
)
.await
.unwrap();
let target = rt
.create_note(
&tok,
"observation",
None,
"nu xi omicron note target",
Some(0.5),
Some(serde_json::json!({"source": "target"})),
vec![],
)
.await
.unwrap();
let filter = serde_json::json!({"source": "target"});
let hits = rt
.search_notes(
&tok,
"nu xi omicron note",
None,
1,
None,
false,
&[],
Some(&filter),
)
.await
.unwrap();
assert_eq!(
hits.len(),
1,
"exactly one hit expected (properties-matching note)"
);
assert_eq!(
hits[0].note_id, target.id,
"properties-filtered note must be returned even when ranked below limit"
);
}
#[tokio::test]
async fn resolve_returns_entity() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "LoRA", None, None, vec![])
.await
.unwrap();
let resolved = rt.resolve(&tok, entity.id).await.unwrap();
match resolved {
Some(Resolved::Entity(e)) => assert_eq!(e.id, entity.id),
other => panic!("expected Resolved::Entity, got {:?}", other),
}
}
#[tokio::test]
async fn resolve_returns_note() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"LoRA fine-tunes LLMs with low-rank adapters",
Some(0.85),
None,
vec![],
)
.await
.unwrap();
let resolved = rt.resolve(&tok, note.id).await.unwrap();
match resolved {
Some(Resolved::Note(n)) => assert_eq!(n.id, note.id),
other => panic!("expected Resolved::Note, got {:?}", other),
}
}
#[tokio::test]
async fn resolve_returns_none_for_unknown_uuid() {
let rt = rt();
let tok = NamespaceToken::local();
let unknown = Uuid::new_v4();
let resolved = rt.resolve(&tok, unknown).await.unwrap();
assert!(resolved.is_none(), "unknown UUID should resolve to None");
}
#[tokio::test]
async fn resolve_prefix_finds_entity_in_own_namespace() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "PrefixTest", None, None, vec![])
.await
.unwrap();
let prefix = &entity.id.to_string()[..8];
let resolved = rt.resolve_prefix(&tok, prefix).await.unwrap();
assert_eq!(resolved, Some(entity.id));
}
#[test]
fn hex_prefix_to_uuid_pattern_inserts_hyphens_at_canonical_boundaries() {
let full = "aabbccdd112240008000000000000ab1";
let cases: &[(usize, &str)] = &[
(1, "a"),
(7, "aabbccd"),
(8, "aabbccdd"),
(9, "aabbccdd-1"),
(12, "aabbccdd-1122"),
(13, "aabbccdd-1122-4"),
(16, "aabbccdd-1122-4000"),
(18, "aabbccdd-1122-4000-80"),
(20, "aabbccdd-1122-4000-8000"),
(23, "aabbccdd-1122-4000-8000-000"),
(24, "aabbccdd-1122-4000-8000-0000"),
(28, "aabbccdd-1122-4000-8000-00000000"),
(31, "aabbccdd-1122-4000-8000-000000000ab"),
];
for (len, expected) in cases {
let input = &full[..*len];
assert_eq!(
hex_prefix_to_uuid_pattern(input),
*expected,
"len={len} input={input:?}"
);
}
}
#[test]
fn hex_prefix_to_uuid_pattern_full_32_char_matches_canonical_uuid() {
let compact = "aabbccdd112240008000000000000ab1";
let compact32 = &compact[..32];
assert_eq!(
hex_prefix_to_uuid_pattern(compact32),
"aabbccdd-1122-4000-8000-000000000ab1"
);
}
#[test]
fn hex_prefix_to_uuid_pattern_overlong_input_is_not_truncated() {
let compact32 = "aabbccdd112240008000000000000ab1";
let overlong = format!("{compact32}extrahex");
let pattern = hex_prefix_to_uuid_pattern(&overlong);
assert_eq!(
pattern, "aabbccdd-1122-4000-8000-000000000ab1extrahex",
"overlong input must keep its extra chars, not truncate to the valid UUID"
);
assert_ne!(
pattern, "aabbccdd-1122-4000-8000-000000000ab1",
"overlong pattern must not collapse to the canonical 36-char UUID form"
);
}
#[test]
fn hex_prefix_to_uuid_pattern_passes_through_hyphenated_input() {
let hyphenated = "aabbccdd-1122-4000-8000-000000000ab1";
assert_eq!(hex_prefix_to_uuid_pattern(hyphenated), hyphenated);
let partial = "aabbccdd-11";
assert_eq!(hex_prefix_to_uuid_pattern(partial), partial);
}
#[tokio::test]
async fn resolve_prefix_compact_9_to_31_char_matches() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(
&tok,
"concept",
None,
"CompactPrefixTest",
None,
None,
vec![],
)
.await
.unwrap();
let compact = entity.id.simple().to_string();
for len in [9, 12, 16, 20, 24, 28, 31] {
let prefix = &compact[..len];
let resolved = rt.resolve_prefix(&tok, prefix).await.unwrap();
assert_eq!(
resolved,
Some(entity.id),
"compact prefix of len {len} should resolve"
);
}
}
#[tokio::test]
async fn resolve_prefix_compact_full_32_char_matches() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "Full32Test", None, None, vec![])
.await
.unwrap();
let compact = entity.id.simple().to_string();
assert_eq!(compact.len(), 32);
let resolved = rt.resolve_prefix(&tok, &compact).await.unwrap();
assert_eq!(resolved, Some(entity.id));
}
#[tokio::test]
async fn resolve_prefix_rejects_overlong_all_hex_input() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "OverlongTest", None, None, vec![])
.await
.unwrap();
let compact = entity.id.simple().to_string();
assert_eq!(compact.len(), 32);
let overlong = format!("{compact}ab");
let resolved = rt.resolve_prefix(&tok, &overlong).await.unwrap();
assert_eq!(
resolved, None,
"a 32-char id plus extra hex chars must not resolve to the valid entity"
);
}
#[tokio::test]
async fn resolve_prefix_rejects_like_wildcard_input() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "WildcardTest", None, None, vec![])
.await
.unwrap();
let compact = entity.id.simple().to_string();
let wildcard_prefix = format!("{}%", &compact[..8]);
let resolved = rt.resolve_prefix(&tok, &wildcard_prefix).await.unwrap();
assert_eq!(
resolved, None,
"prefix containing a LIKE wildcard must be rejected, not resolved"
);
}
#[tokio::test]
async fn resolve_prefix_boundary_at_hyphen_positions() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "BoundaryTest", None, None, vec![])
.await
.unwrap();
let compact = entity.id.simple().to_string();
for len in [8, 12, 16, 20, 24] {
let prefix = &compact[..len];
let resolved = rt.resolve_prefix(&tok, prefix).await.unwrap();
assert_eq!(
resolved,
Some(entity.id),
"boundary prefix of len {len} should resolve"
);
}
}
#[tokio::test]
async fn resolve_prefix_ambiguous_still_detected_after_normalization() {
use khive_storage::entity::Entity;
let rt = rt();
let tok = NamespaceToken::local();
let id_a = Uuid::parse_str("aabbccdd-1111-4000-8000-000000000001").unwrap();
let id_b = Uuid::parse_str("aabbccdd-1111-4000-8000-000000000002").unwrap();
let mut entity_a = Entity::new("local", "concept", "AmbigCompactA");
entity_a.id = id_a;
let mut entity_b = Entity::new("local", "concept", "AmbigCompactB");
entity_b.id = id_b;
let store = rt.entities(&tok).unwrap();
store.upsert_entity(entity_a).await.unwrap();
store.upsert_entity(entity_b).await.unwrap();
let shared_compact = &id_a.simple().to_string()[..20];
let err = rt.resolve_prefix(&tok, shared_compact).await.unwrap_err();
assert!(
matches!(
err,
RuntimeError::AmbiguousPrefix { ref matches, .. } if matches.len() == 2
),
"shared compact prefix must still return AmbiguousPrefix; got {err:?}"
);
}
#[tokio::test]
async fn resolve_prefix_invisible_across_namespaces() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let entity = rt
.create_entity(&ns_a, "concept", None, "Invisible", None, None, vec![])
.await
.unwrap();
let prefix = &entity.id.to_string()[..8];
let resolved = rt.resolve_prefix(&ns_b, prefix).await.unwrap();
assert_eq!(resolved, None);
}
#[tokio::test]
async fn resolve_prefix_ambiguous_same_namespace() {
use khive_storage::entity::Entity;
let rt = rt();
let tok = NamespaceToken::local();
let id_a = Uuid::parse_str("aabbccdd-1111-4000-8000-000000000001").unwrap();
let id_b = Uuid::parse_str("aabbccdd-2222-4000-8000-000000000002").unwrap();
let mut entity_a = Entity::new("local", "concept", "AmbigA");
entity_a.id = id_a;
let mut entity_b = Entity::new("local", "concept", "AmbigB");
entity_b.id = id_b;
let store = rt.entities(&tok).unwrap();
store.upsert_entity(entity_a).await.unwrap();
store.upsert_entity(entity_b).await.unwrap();
let err = rt.resolve_prefix(&tok, "aabbccdd").await.unwrap_err();
assert!(
matches!(
err,
RuntimeError::AmbiguousPrefix { ref prefix, ref matches }
if prefix == "aabbccdd" && matches.len() == 2
),
"shared 8-char prefix must return AmbiguousPrefix; got {err:?}"
);
}
#[tokio::test]
async fn resolve_prefix_cross_table_duplicate_uuid_resolves_cleanly() {
use khive_storage::entity::Entity;
let rt = rt();
let tok = NamespaceToken::local();
let shared_id = Uuid::parse_str("ccddeeff-1111-4000-8000-000000000001").unwrap();
let mut entity = Entity::new("local", "concept", "Nvk749Entity");
entity.id = shared_id;
rt.entities(&tok)
.unwrap()
.upsert_entity(entity)
.await
.unwrap();
let mut note = Note::new("local", "observation", "nvk749 note with the same id");
note.id = shared_id;
rt.notes(&tok).unwrap().upsert_note(note).await.unwrap();
let resolved = rt
.resolve_prefix(&tok, "ccddeeff")
.await
.expect("#749: a UUID present in two tables must not be reported as ambiguous");
assert_eq!(
resolved,
Some(shared_id),
"#749: cross-table duplicate must resolve to the single shared UUID"
);
}
#[tokio::test]
async fn resolve_prefix_early_exit_uses_deduped_match_count() {
use khive_storage::entity::Entity;
let rt = rt();
let tok = NamespaceToken::local();
let shared_id = Uuid::parse_str("ddeeff11-2222-4000-8000-000000000002").unwrap();
let mut entity = Entity::new("local", "concept", "Nvk749bEntity");
entity.id = shared_id;
rt.entities(&tok)
.unwrap()
.upsert_entity(entity)
.await
.unwrap();
let mut note = Note::new("local", "observation", "nvk749b note with the same id");
note.id = shared_id;
rt.notes(&tok).unwrap().upsert_note(note).await.unwrap();
let resolved = rt
.resolve_prefix(&tok, "ddeeff11")
.await
.expect("#749: deduped early-exit must not falsely report ambiguity");
assert_eq!(resolved, Some(shared_id));
}
#[tokio::test]
async fn resolve_finds_event_by_full_uuid() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let resolved = rt.resolve(&tok, event_id).await.unwrap();
assert!(
matches!(resolved, Some(Resolved::Event(_))),
"event UUID must resolve to Resolved::Event, got {resolved:?}"
);
}
#[tokio::test]
async fn resolve_prefix_finds_event() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let prefix = &event_id.to_string()[..8];
let resolved = rt.resolve_prefix(&tok, prefix).await.unwrap();
assert_eq!(
resolved,
Some(event_id),
"resolve_prefix must return event UUID for 8-char prefix"
);
}
#[tokio::test]
async fn link_phantom_source_returns_not_found() {
let rt = rt();
let tok = NamespaceToken::local();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let phantom = Uuid::new_v4();
let result = rt
.link(&tok, phantom, b.id, EdgeRelation::Extends, 1.0, None)
.await;
match result {
Err(RuntimeError::NotFound(msg)) => {
assert!(
msg.contains("source"),
"error message must name 'source': {msg}"
);
}
other => panic!("expected NotFound for phantom source, got {other:?}"),
}
}
#[tokio::test]
async fn link_phantom_target_returns_not_found() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let phantom = Uuid::new_v4();
let result = rt
.link(&tok, a.id, phantom, EdgeRelation::Extends, 1.0, None)
.await;
match result {
Err(RuntimeError::NotFound(msg)) => {
assert!(
msg.contains("target"),
"error message must name 'target': {msg}"
);
}
other => panic!("expected NotFound for phantom target, got {other:?}"),
}
}
#[tokio::test]
async fn link_real_entities_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 0.8, None)
.await
.unwrap();
assert_eq!(edge.source_id, a.id);
assert_eq!(edge.target_id, b.id);
assert_eq!(edge.relation, EdgeRelation::Extends);
}
#[tokio::test]
async fn link_write_time_guard_blocks_dangling_edge_after_target_vanishes() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
rt.validate_edge_relation_endpoints(&tok, a.id, x.id, EdgeRelation::Extends)
.await
.expect("prepare-time validation must pass while X is live");
assert!(rt.delete_entity(&tok, x.id, true).await.unwrap());
let now = chrono::Utc::now();
let edge = Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: tok.namespace().as_str().to_string(),
source_id: a.id,
target_id: x.id,
relation: EdgeRelation::Extends,
weight: 1.0,
created_at: now,
updated_at: now,
deleted_at: None,
metadata: None,
target_backend: None,
};
let outcome = rt
.graph(&tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
.unwrap();
match outcome {
khive_storage::GuardedWriteOutcome::Refused(missing) => {
assert!(missing.target, "target must be reported missing");
}
other => panic!(
"guarded write must refuse an edge whose target vanished before commit, got {other:?}"
),
}
let edges = rt
.list_edges(
&tok,
crate::curation::EdgeListFilter {
source_id: Some(a.id),
target_id: Some(x.id),
relations: vec![EdgeRelation::Extends],
..Default::default()
},
10,
0,
)
.await
.unwrap();
assert!(
edges.is_empty(),
"no dangling edge may be persisted after the guarded write refused it"
);
}
#[tokio::test]
async fn link_many_writes_nothing_when_one_target_vanishes_before_write() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
let specs = vec![
LinkSpec {
namespace: None,
source_id: a.id,
target_id: x.id,
relation: EdgeRelation::Extends,
weight: 1.0,
metadata: None,
},
LinkSpec {
namespace: None,
source_id: a.id,
target_id: b.id,
relation: EdgeRelation::Extends,
weight: 1.0,
metadata: None,
},
];
let mut edges = Vec::with_capacity(specs.len());
for spec in &specs {
edges.push(rt.build_edge(&tok, spec).await.unwrap());
}
assert!(rt.delete_entity(&tok, x.id, true).await.unwrap());
let outcome = rt
.graph(&tok)
.unwrap()
.upsert_edges_guarded(edges)
.await
.unwrap();
assert_eq!(
outcome.summary.affected, 0,
"no edge from the batch may be persisted when any endpoint vanished"
);
assert!(
outcome.refused.is_some(),
"refused batch entry must be reported"
);
let edges = rt
.list_edges(
&tok,
crate::curation::EdgeListFilter {
source_id: Some(a.id),
relations: vec![EdgeRelation::Extends],
..Default::default()
},
10,
0,
)
.await
.unwrap();
assert!(
edges.is_empty(),
"link_many's guarded batch must be all-or-nothing: the live A-B edge \
must not have been persisted alongside the doomed A-X edge"
);
}
fn raw_edge(source_id: Uuid, target_id: Uuid, ns: &str) -> Edge {
let now = chrono::Utc::now();
Edge {
id: LinkId::from(Uuid::new_v4()),
namespace: ns.to_string(),
source_id,
target_id,
relation: EdgeRelation::Extends,
weight: 1.0,
created_at: now,
updated_at: now,
deleted_at: None,
metadata: None,
target_backend: None,
}
}
async fn assert_no_edges_touch(rt: &KhiveRuntime, tok: &NamespaceToken, node_id: Uuid) {
let as_source = rt
.list_edges(
tok,
crate::curation::EdgeListFilter {
source_id: Some(node_id),
..Default::default()
},
10,
0,
)
.await
.unwrap();
let as_target = rt
.list_edges(
tok,
crate::curation::EdgeListFilter {
target_id: Some(node_id),
..Default::default()
},
10,
0,
)
.await
.unwrap();
assert!(
as_source.is_empty() && as_target.is_empty(),
"no edge may reference hard-deleted node {node_id}: source-side={as_source:?} \
target-side={as_target:?}"
);
}
#[tokio::test]
async fn hard_delete_entity_purges_edge_written_before_delete() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
let edge = raw_edge(a.id, x.id, tok.namespace().as_str());
assert_eq!(
rt.graph(&tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
.unwrap(),
khive_storage::GuardedWriteOutcome::Written
);
assert!(rt.delete_entity(&tok, x.id, true).await.unwrap());
assert_no_edges_touch(&rt, &tok, x.id).await;
}
#[tokio::test]
async fn hard_delete_entity_concurrent_with_guarded_write_never_leaves_dangling_edge() {
let rt = std::sync::Arc::new(rt());
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
let delete_rt = std::sync::Arc::clone(&rt);
let delete_tok = tok.clone();
let delete_task =
tokio::spawn(async move { delete_rt.delete_entity(&delete_tok, x.id, true).await });
let write_rt = std::sync::Arc::clone(&rt);
let write_tok = tok.clone();
let ns = tok.namespace().as_str().to_string();
let write_task = tokio::spawn(async move {
let edge = raw_edge(a.id, x.id, &ns);
write_rt
.graph(&write_tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
});
let (deleted, _written) = tokio::join!(delete_task, write_task);
deleted.unwrap().unwrap();
assert_no_edges_touch(&rt, &tok, x.id).await;
}
#[tokio::test]
async fn hard_delete_note_purges_edge_written_before_delete() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let n = rt
.create_note(
&tok,
"observation",
None,
"note content",
None,
None,
vec![],
)
.await
.unwrap();
let edge = raw_edge(a.id, n.id, tok.namespace().as_str());
assert_eq!(
rt.graph(&tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
.unwrap(),
khive_storage::GuardedWriteOutcome::Written
);
assert!(rt.delete_note(&tok, n.id, true).await.unwrap());
assert_no_edges_touch(&rt, &tok, n.id).await;
}
#[tokio::test]
async fn hard_delete_note_concurrent_with_guarded_write_never_leaves_dangling_edge() {
let rt = std::sync::Arc::new(rt());
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let n = rt
.create_note(
&tok,
"observation",
None,
"note content",
None,
None,
vec![],
)
.await
.unwrap();
let delete_rt = std::sync::Arc::clone(&rt);
let delete_tok = tok.clone();
let note_id = n.id;
let delete_task =
tokio::spawn(async move { delete_rt.delete_note(&delete_tok, note_id, true).await });
let write_rt = std::sync::Arc::clone(&rt);
let write_tok = tok.clone();
let ns = tok.namespace().as_str().to_string();
let write_task = tokio::spawn(async move {
let edge = raw_edge(a.id, note_id, &ns);
write_rt
.graph(&write_tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
});
let (deleted, _written) = tokio::join!(delete_task, write_task);
deleted.unwrap().unwrap();
assert_no_edges_touch(&rt, &tok, note_id).await;
}
#[tokio::test]
async fn hard_delete_edge_endpoint_purges_annotating_edge_written_before_delete() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let n = rt
.create_note(
&tok,
"observation",
None,
"note content",
None,
None,
vec![],
)
.await
.unwrap();
let base_edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 0.8, None)
.await
.unwrap();
let base_edge_id = Uuid::from(base_edge.id);
let annotating = raw_edge(n.id, base_edge_id, tok.namespace().as_str());
assert_eq!(
rt.graph(&tok)
.unwrap()
.upsert_edge_guarded(annotating)
.await
.unwrap(),
khive_storage::GuardedWriteOutcome::Written
);
assert!(rt.delete_edge(&tok, base_edge_id, true).await.unwrap());
assert_no_edges_touch(&rt, &tok, base_edge_id).await;
}
#[tokio::test]
async fn hard_delete_edge_endpoint_concurrent_with_guarded_write_never_leaves_dangling_edge() {
let rt = std::sync::Arc::new(rt());
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let n = rt
.create_note(
&tok,
"observation",
None,
"note content",
None,
None,
vec![],
)
.await
.unwrap();
let base_edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 0.8, None)
.await
.unwrap();
let base_edge_id = Uuid::from(base_edge.id);
let delete_rt = std::sync::Arc::clone(&rt);
let delete_tok = tok.clone();
let delete_task =
tokio::spawn(
async move { delete_rt.delete_edge(&delete_tok, base_edge_id, true).await },
);
let write_rt = std::sync::Arc::clone(&rt);
let write_tok = tok.clone();
let ns = tok.namespace().as_str().to_string();
let write_task = tokio::spawn(async move {
let edge = raw_edge(n.id, base_edge_id, &ns);
write_rt
.graph(&write_tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
});
let (deleted, _written) = tokio::join!(delete_task, write_task);
deleted.unwrap().unwrap();
assert_no_edges_touch(&rt, &tok, base_edge_id).await;
}
fn file_backed_runtime(
dir: &tempfile::TempDir,
name: &str,
write_queue_enabled: bool,
) -> KhiveRuntime {
let path = dir.path().join(name);
if write_queue_enabled {
std::env::set_var("KHIVE_WRITE_QUEUE", "1");
} else {
std::env::remove_var("KHIVE_WRITE_QUEUE");
}
let rt = KhiveRuntime::new(crate::config::RuntimeConfig {
db_path: Some(path),
packs: vec!["kg".to_string()],
brain_profile: None,
actor_id: None,
..crate::config::RuntimeConfig::no_embeddings()
})
.unwrap();
std::env::remove_var("KHIVE_WRITE_QUEUE");
rt
}
async fn assert_guarded_write_committed_before_delete_is_swept(rt: &KhiveRuntime) {
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
let edge = raw_edge(a.id, x.id, tok.namespace().as_str());
assert_eq!(
rt.graph(&tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
.unwrap(),
khive_storage::GuardedWriteOutcome::Written,
"write must succeed while both endpoints are still live"
);
assert!(rt.delete_entity(&tok, x.id, true).await.unwrap());
assert_no_edges_touch(rt, &tok, x.id).await;
}
async fn assert_guarded_write_attempted_after_delete_is_refused(rt: &KhiveRuntime) {
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let x = rt
.create_entity(&tok, "concept", None, "X", None, None, vec![])
.await
.unwrap();
assert!(rt.delete_entity(&tok, x.id, true).await.unwrap());
let edge = raw_edge(a.id, x.id, tok.namespace().as_str());
let outcome = rt
.graph(&tok)
.unwrap()
.upsert_edge_guarded(edge)
.await
.unwrap();
match outcome {
khive_storage::GuardedWriteOutcome::Refused(missing) => {
assert!(
missing.target,
"target must be reported missing once the delete has committed"
);
assert!(!missing.source, "source was never deleted");
}
other => panic!(
"guarded write attempted after the delete committed must be refused, got {other:?}"
),
}
assert_no_edges_touch(rt, &tok, x.id).await;
}
#[tokio::test]
#[serial_test::serial(khive_write_queue_env)]
async fn guarded_write_before_delete_swept_file_backed_write_queue_off() {
let dir = tempfile::tempdir().unwrap();
let rt = file_backed_runtime(&dir, "guard_before_off.db", false);
assert_guarded_write_committed_before_delete_is_swept(&rt).await;
}
#[tokio::test]
#[serial_test::serial(khive_write_queue_env)]
async fn guarded_write_after_delete_refused_file_backed_write_queue_off() {
let dir = tempfile::tempdir().unwrap();
let rt = file_backed_runtime(&dir, "guard_after_off.db", false);
assert_guarded_write_attempted_after_delete_is_refused(&rt).await;
}
#[tokio::test]
#[serial_test::serial(khive_write_queue_env)]
async fn guarded_write_before_delete_swept_file_backed_write_queue_on() {
let dir = tempfile::tempdir().unwrap();
let rt = file_backed_runtime(&dir, "guard_before_on.db", true);
assert_guarded_write_committed_before_delete_is_swept(&rt).await;
}
#[tokio::test]
#[serial_test::serial(khive_write_queue_env)]
async fn guarded_write_after_delete_refused_file_backed_write_queue_on() {
let dir = tempfile::tempdir().unwrap();
let rt = file_backed_runtime(&dir, "guard_after_on.db", true);
assert_guarded_write_attempted_after_delete_is_refused(&rt).await;
}
#[tokio::test]
async fn create_note_annotates_phantom_returns_not_found() {
let rt = rt();
let tok = NamespaceToken::local();
let phantom = Uuid::new_v4();
let result = rt
.create_note(
&tok,
"observation",
None,
"some content",
Some(0.5),
None,
vec![phantom],
)
.await;
assert!(
matches!(result, Err(RuntimeError::NotFound(_))),
"annotates with phantom uuid must return NotFound, got {result:?}"
);
}
#[tokio::test]
async fn create_note_annotates_real_entity_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "RealTarget", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"content",
Some(0.5),
None,
vec![entity.id],
)
.await
.unwrap();
let neighbors = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(neighbors.len(), 1);
assert_eq!(neighbors[0].node_id, entity.id);
}
#[tokio::test]
async fn create_note_multi_annotates_creates_all_edges() {
let rt = rt();
let tok = NamespaceToken::local();
let t1 = rt
.create_entity(&tok, "concept", None, "Target1", None, None, vec![])
.await
.unwrap();
let t2 = rt
.create_entity(&tok, "concept", None, "Target2", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"content",
Some(0.5),
None,
vec![t1.id, t2.id],
)
.await
.unwrap();
let neighbors = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
neighbors.len(),
2,
"multi-annotates note must have exactly 2 outbound annotates edges"
);
let target_ids: Vec<Uuid> = neighbors.iter().map(|n| n.node_id).collect();
assert!(target_ids.contains(&t1.id));
assert!(target_ids.contains(&t2.id));
}
#[tokio::test]
async fn link_target_in_different_namespace_succeeds() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let a = rt
.create_entity(&ns_a, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&ns_b, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&ns_a, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await;
assert!(
result.is_ok(),
"target in a different namespace than the caller must resolve (#631), got {result:?}"
);
}
#[tokio::test]
async fn link_phantom_self_loop_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let phantom = Uuid::new_v4();
let result = rt
.link(&tok, phantom, phantom, EdgeRelation::Extends, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("self-loop"),
"self-loop must be rejected with self-loop message: {msg}"
);
}
other => panic!("expected InvalidInput for self-loop, got {other:?}"),
}
}
#[tokio::test]
async fn link_note_to_edge_annotates_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let note = rt
.create_note(
&tok,
"observation",
None,
"edge note",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(&tok, note.id, edge_uuid, EdgeRelation::Annotates, 1.0, None)
.await;
assert!(
result.is_ok(),
"note→edge Annotates must succeed, got {result:?}"
);
}
#[tokio::test]
async fn create_note_annotates_real_edge_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let note = rt
.create_note(
&tok,
"observation",
None,
"annotating an edge",
Some(0.5),
None,
vec![edge_uuid],
)
.await
.unwrap();
let neighbors = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(neighbors.len(), 1);
assert_eq!(neighbors[0].node_id, edge_uuid);
}
#[tokio::test]
async fn create_note_annotates_phantom_is_atomic_no_note_persisted() {
let rt = rt();
let tok = NamespaceToken::local();
let phantom = Uuid::new_v4();
let before_count = rt.list_notes(&tok, None, 1000, 0).await.unwrap().len();
let result = rt
.create_note(
&tok,
"observation",
None,
"should not persist",
Some(0.5),
None,
vec![phantom],
)
.await;
assert!(
matches!(result, Err(RuntimeError::NotFound(_))),
"phantom annotates target must return NotFound, got {result:?}"
);
let after_count = rt.list_notes(&tok, None, 1000, 0).await.unwrap().len();
assert_eq!(
before_count, after_count,
"failed create_note must not persist any note row (atomicity)"
);
let search_hits = rt
.search_notes(&tok, "should not persist", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(
search_hits.is_empty(),
"failed create_note must not index into FTS (atomicity)"
);
}
#[tokio::test]
async fn link_entity_to_edge_uuid_non_annotates_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let result = rt
.link(&tok, a.id, edge_uuid, EdgeRelation::Extends, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("target"),
"error message must name 'target': {msg}"
);
}
other => {
panic!("expected InvalidInput for edge-uuid target with Extends, got {other:?}")
}
}
}
#[tokio::test]
async fn link_note_as_source_non_annotates_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(&tok, "observation", None, "a note", Some(0.5), None, vec![])
.await
.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "E", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, note.id, entity.id, EdgeRelation::DependsOn, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("source"),
"error message must name 'source': {msg}"
);
}
other => panic!("expected InvalidInput for note source with DependsOn, got {other:?}"),
}
}
#[tokio::test]
async fn link_entity_as_annotates_source_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, a.id, b.id, EdgeRelation::Annotates, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("source") && msg.contains("note"),
"error must say source must be a note: {msg}"
);
}
other => {
panic!("expected InvalidInput for entity source with Annotates, got {other:?}")
}
}
}
#[tokio::test]
async fn link_edge_as_annotates_source_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let result = rt
.link(&tok, edge_uuid, a.id, EdgeRelation::Annotates, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("source") && msg.contains("note"),
"edge-as-annotates-source must report wrong kind, not NotFound: {msg}"
);
}
other => panic!("expected InvalidInput for edge source with Annotates, got {other:?}"),
}
}
#[tokio::test]
async fn link_note_to_event_annotates_succeeds() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"observing an event",
Some(0.6),
None,
vec![],
)
.await
.unwrap();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"test_actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let result = rt
.link(&tok, note.id, event_id, EdgeRelation::Annotates, 1.0, None)
.await;
assert!(
result.is_ok(),
"note→event Annotates must succeed, got {result:?}"
);
}
#[tokio::test]
async fn create_note_annotates_event_succeeds() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"test_actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let result = rt
.create_note(
&tok,
"observation",
None,
"note annotating an event",
Some(0.5),
None,
vec![event_id],
)
.await;
assert!(
result.is_ok(),
"create_note with event annotates target must succeed, got {result:?}"
);
let note = result.unwrap();
let neighbors = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(neighbors.len(), 1);
assert_eq!(neighbors[0].node_id, event_id);
}
#[tokio::test]
async fn link_supersedes_note_to_note_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let old_note = rt
.create_note(
&tok,
"observation",
None,
"old observation",
Some(0.7),
None,
vec![],
)
.await
.unwrap();
let new_note = rt
.create_note(
&tok,
"observation",
None,
"revised observation superseding the old one",
Some(0.9),
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(
&tok,
new_note.id,
old_note.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
assert!(
result.is_ok(),
"note→note Supersedes must succeed (note supersession), got {result:?}"
);
}
#[tokio::test]
async fn link_supersedes_entity_to_entity_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let old_entity = rt
.create_entity(&tok, "concept", None, "OldConcept", None, None, vec![])
.await
.unwrap();
let new_entity = rt
.create_entity(&tok, "concept", None, "NewConcept", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
new_entity.id,
old_entity.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
assert!(
result.is_ok(),
"entity→entity Supersedes must succeed, got {result:?}"
);
}
#[tokio::test]
async fn link_supersedes_note_to_entity_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(&tok, "observation", None, "a note", Some(0.5), None, vec![])
.await
.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "SomeEntity", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
note.id,
entity.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("same substrate") || msg.contains("same-substrate"),
"error must name the same-substrate rule: {msg}"
);
}
other => panic!(
"expected InvalidInput for note→entity Supersedes (cross-substrate), got {other:?}"
),
}
}
#[tokio::test]
async fn link_supersedes_entity_to_note_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "SomeEntity", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(&tok, "observation", None, "a note", Some(0.5), None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
entity.id,
note.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(
msg.contains("same substrate") || msg.contains("same-substrate"),
"error must name the same-substrate rule: {msg}"
);
}
other => panic!(
"expected InvalidInput for entity→note Supersedes (cross-substrate), got {other:?}"
),
}
}
#[tokio::test]
async fn link_supersedes_event_source_returns_invalid_input() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"test_actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "SomeEntity", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
event_id,
entity.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(msg.contains("event"), "error must mention 'event': {msg}");
}
other => {
panic!("expected InvalidInput for event source with Supersedes, got {other:?}")
}
}
}
#[tokio::test]
async fn link_supersedes_event_target_returns_invalid_input() {
use khive_storage::Event;
use khive_types::{EventKind, SubstrateKind};
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().as_str();
let event = Event::new(
ns,
"test_verb",
EventKind::Audit,
SubstrateKind::Entity,
"test_actor",
);
let event_id = event.id;
rt.events(&tok).unwrap().append_event(event).await.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "SomeEntity", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
entity.id,
event_id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(msg.contains("event"), "error must mention 'event': {msg}");
}
other => {
panic!("expected InvalidInput for event target with Supersedes, got {other:?}")
}
}
}
#[tokio::test]
async fn link_supersedes_edge_source_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let result = rt
.link(&tok, edge_uuid, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(msg.contains("source"), "error must name 'source': {msg}");
}
other => {
panic!("expected InvalidInput for edge-uuid source with Supersedes, got {other:?}")
}
}
}
#[tokio::test]
async fn link_supersedes_edge_target_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let result = rt
.link(&tok, a.id, edge_uuid, EdgeRelation::Supersedes, 1.0, None)
.await;
match result {
Err(RuntimeError::InvalidInput(msg)) => {
assert!(msg.contains("target"), "error must name 'target': {msg}");
}
other => {
panic!("expected InvalidInput for edge-uuid target with Supersedes, got {other:?}")
}
}
}
#[tokio::test]
async fn link_supersedes_phantom_source_returns_not_found() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"existing note",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let phantom = Uuid::new_v4();
let result = rt
.link(&tok, phantom, note.id, EdgeRelation::Supersedes, 1.0, None)
.await;
match result {
Err(RuntimeError::NotFound(msg)) => {
assert!(msg.contains("source"), "error must name 'source': {msg}");
}
other => panic!("expected NotFound for phantom source with Supersedes, got {other:?}"),
}
}
#[tokio::test]
async fn link_supersedes_phantom_target_returns_not_found() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"existing note",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let phantom = Uuid::new_v4();
let result = rt
.link(&tok, note.id, phantom, EdgeRelation::Supersedes, 1.0, None)
.await;
match result {
Err(RuntimeError::NotFound(msg)) => {
assert!(msg.contains("target"), "error must name 'target': {msg}");
}
other => panic!("expected NotFound for phantom target with Supersedes, got {other:?}"),
}
}
#[tokio::test]
async fn link_supersedes_cross_namespace_source_succeeds() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let note_a = rt
.create_note(
&ns_a,
"observation",
None,
"note in ns-a",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let note_b = rt
.create_note(
&ns_b,
"observation",
None,
"note in ns-b",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(
&ns_a,
note_b.id,
note_a.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
assert!(
result.is_ok(),
"cross-namespace supersedes source must resolve (#631), got {result:?}"
);
}
#[tokio::test]
async fn link_extends_note_source_still_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"a note that cannot be an extends source",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let entity = rt
.create_entity(&tok, "concept", None, "E", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, note.id, entity.id, EdgeRelation::Extends, 1.0, None)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"note source with Extends must still return InvalidInput after this fix, got {result:?}"
);
}
#[tokio::test]
async fn link_annotates_note_to_edge_still_succeeds_after_fix() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let note = rt
.create_note(
&tok,
"observation",
None,
"annotating an edge",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(&tok, note.id, edge_uuid, EdgeRelation::Annotates, 1.0, None)
.await;
assert!(
result.is_ok(),
"note→edge Annotates must still succeed after supersedes fix, got {result:?}"
);
}
#[tokio::test]
async fn create_note_multi_annotates_compensation_cleanup_restores_pristine_state() {
let rt = rt();
let tok = NamespaceToken::local();
let t1 = rt
.create_entity(&tok, "concept", None, "T1", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"partial note",
Some(0.5),
None,
vec![t1.id],
)
.await
.unwrap();
let before_notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert_eq!(before_notes.len(), 1, "note must be present before cleanup");
let before_edges = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
before_edges.len(),
1,
"one annotates edge must exist before cleanup"
);
let edge_id: Uuid = before_edges[0].edge_id;
rt.delete_edge(&tok, edge_id, true).await.unwrap();
rt.delete_note(&tok, note.id, true )
.await
.unwrap();
let after_notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
after_notes.is_empty(),
"compensation must remove the note row; got {after_notes:?}"
);
let search_hits = rt
.search_notes(&tok, "partial note", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(
search_hits.is_empty(),
"compensation must clean the FTS index; got {search_hits:?}"
);
let after_edges = rt
.neighbors(&tok, note.id, Direction::Out, None, None)
.await
.unwrap();
assert!(
after_edges.is_empty(),
"compensation must remove all partial edges; got {after_edges:?}"
);
}
#[tokio::test]
async fn annotated_entity_hard_delete_cascades_annotate_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "E", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"note about entity",
Some(0.5),
None,
vec![entity.id],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
before.len(),
1,
"annotates edge must exist before entity delete"
);
let deleted = rt.delete_entity(&tok, entity.id, true).await.unwrap();
assert!(deleted, "entity hard delete must return true");
let after = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert!(
after.is_empty(),
"annotates edge must be cascaded on entity hard delete; got {after:?}"
);
}
#[tokio::test]
async fn annotated_note_hard_delete_cascades_annotate_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let note_target = rt
.create_note(
&tok,
"observation",
None,
"target note",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let note_source = rt
.create_note(
&tok,
"insight",
None,
"annotation",
Some(0.5),
None,
vec![note_target.id],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note_source.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
before.len(),
1,
"annotates edge must exist before note delete"
);
let deleted = rt.delete_note(&tok, note_target.id, true).await.unwrap();
assert!(deleted, "note hard delete must return true");
let after = rt
.neighbors(
&tok,
note_source.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert!(
after.is_empty(),
"annotates edge must be cascaded on note-target hard delete; got {after:?}"
);
}
#[tokio::test]
async fn annotated_edge_delete_cascades_annotate_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let base_edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let base_edge_uuid: Uuid = base_edge.id.into();
let note = rt
.create_note(
&tok,
"observation",
None,
"note about edge",
Some(0.5),
None,
vec![base_edge_uuid],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
before.len(),
1,
"annotates edge must exist before base edge delete"
);
let deleted = rt.delete_edge(&tok, base_edge_uuid, true).await.unwrap();
assert!(deleted, "edge delete must return true");
let after = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert!(
after.is_empty(),
"annotates edge must be cascaded on base edge delete; got {after:?}"
);
}
#[tokio::test]
async fn mixed_multi_annotates_partial_target_hard_delete_leaves_remaining_edges() {
let rt = rt();
let tok = NamespaceToken::local();
let t1 = rt
.create_entity(&tok, "concept", None, "T1", None, None, vec![])
.await
.unwrap();
let t2 = rt
.create_entity(&tok, "concept", None, "T2", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"multi-target note",
Some(0.5),
None,
vec![t1.id, t2.id],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
before.len(),
2,
"must have 2 annotates edges before any delete"
);
rt.delete_entity(&tok, t1.id, true).await.unwrap();
let after = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
after.len(),
1,
"only the edge to t1 must be cascaded; t2 edge must remain"
);
assert_eq!(
after[0].node_id, t2.id,
"remaining annotates edge must point to t2"
);
}
#[tokio::test]
async fn annotated_note_soft_delete_preserves_annotate_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let note_target = rt
.create_note(&tok, "observation", None, "target", Some(0.5), None, vec![])
.await
.unwrap();
let note_source = rt
.create_note(
&tok,
"insight",
None,
"annotation",
Some(0.5),
None,
vec![note_target.id],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note_source.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(before.len(), 1);
let edge_id = before[0].edge_id;
let deleted = rt.delete_note(&tok, note_target.id, false).await.unwrap();
assert!(deleted, "soft delete must return true");
let edge_after = rt.get_edge(&tok, edge_id).await.unwrap();
assert!(
edge_after.is_some(),
"soft delete must NOT cascade edges; get_edge returned None"
);
let after = rt
.neighbors(
&tok,
note_source.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
after.len(),
0,
"#748: neighbors() must screen out the soft-deleted note target; got {after:?}"
);
}
#[tokio::test]
async fn delete_edge_non_edge_uuid_has_no_side_effects() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "Target", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(
&tok,
"observation",
None,
"annotates the entity",
Some(0.5),
None,
vec![entity.id],
)
.await
.unwrap();
let before = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(before.len(), 1, "annotates edge must exist before test");
let annotates_edge_id: Uuid = before[0].edge_id;
let result = rt.delete_edge(&tok, entity.id, true).await;
assert!(
result.is_ok(),
"delete_edge must not error on a non-edge UUID"
);
assert!(
!result.unwrap(),
"delete_edge must return false for a non-edge UUID"
);
let after = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(
after.len(),
1,
"delete_edge with a non-edge UUID must not touch inbound annotates edges"
);
assert_eq!(
after[0].edge_id, annotates_edge_id,
"the original annotates edge must be unchanged"
);
}
#[tokio::test]
async fn create_note_multi_annotates_second_link_failure_rolls_back_partial_write() {
let rt = rt();
let tok = NamespaceToken::local();
let t1 = rt
.create_entity(&tok, "concept", None, "T1", None, None, vec![])
.await
.unwrap();
let t2 = rt
.create_entity(&tok, "concept", None, "T2", None, None, vec![])
.await
.unwrap();
LINK_FAIL_AFTER.with(|cell| cell.set(2));
let result = rt
.create_note(
&tok,
"observation",
None,
"rollback target",
Some(0.5),
None,
vec![t1.id, t2.id],
)
.await;
assert!(
result.is_err(),
"create_note must propagate the injected link failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected link failure"),
"error must carry injection message; got: {err_msg}"
);
let notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
notes.is_empty(),
"compensation must remove the note row; got {notes:?}"
);
let hits = rt
.search_notes(&tok, "rollback target", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(
hits.is_empty(),
"compensation must clean FTS index; got {hits:?}"
);
let edges_from_t1 = rt
.neighbors(
&tok,
t1.id,
Direction::In,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
let edges_from_t2 = rt
.neighbors(
&tok,
t2.id,
Direction::In,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert!(
edges_from_t1.is_empty(),
"compensation must delete the first annotates edge; got {edges_from_t1:?}"
);
assert!(
edges_from_t2.is_empty(),
"no second annotates edge must exist; got {edges_from_t2:?}"
);
}
#[tokio::test]
async fn create_note_fts_failure_rolls_back_note_row() {
let rt = rt();
let ns = Namespace::parse("fault-fts-rollback").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_fts_fail(ns.as_str());
let result = rt
.create_note(
&tok,
"observation",
None,
"fts-fail rollback target",
None,
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_note must propagate the injected FTS failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected FTS failure"),
"error must carry injection message; got: {err_msg}"
);
let notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
notes.is_empty(),
"compensation must remove the note row after FTS failure; got {notes:?}"
);
}
#[tokio::test]
async fn create_note_vector_failure_rolls_back_note_row_and_fts() {
const MODEL: &str = "test-vec-inject";
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider, _counter) = ConstVecProvider::new(MODEL, DIMS);
rt.register_embedder(provider);
let ns = Namespace::parse("fault-vec-rollback").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_vector_fail(ns.as_str());
let result = rt
.create_note(
&tok,
"observation",
None,
"vec-fail rollback target",
None,
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_note must propagate the injected vector failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected vector failure"),
"error must carry injection message; got: {err_msg}"
);
let notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
notes.is_empty(),
"compensation must remove note row after vector failure; got {notes:?}"
);
}
#[tokio::test]
async fn create_note_with_embedding_content_fts_failure_rolls_back_note_row() {
let rt = rt();
let ns = Namespace::parse("fault-fts-rollback-embedding-content").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_fts_fail(ns.as_str());
let full = "fts-fail rollback target with an embedding-content override";
let head = &full[.."fts-fail rollback target".len()];
let result = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
full,
Some(head),
None,
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_note_with_embedding_content must propagate the injected FTS failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected FTS failure"),
"error must carry injection message; got: {err_msg}"
);
let notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
notes.is_empty(),
"compensation must remove the note row after FTS failure; got {notes:?}"
);
}
#[tokio::test]
async fn create_note_with_embedding_content_vector_failure_rolls_back_note_row_and_fts() {
const MODEL: &str = "test-vec-inject-embedding-content";
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider, _counter) = ConstVecProvider::new(MODEL, DIMS);
rt.register_embedder(provider);
let ns = Namespace::parse("fault-vec-rollback-embedding-content").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_vector_fail(ns.as_str());
let full = "vec-fail rollback target with an embedding-content override";
let head = &full[.."vec-fail rollback target".len()];
let result = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
full,
Some(head),
None,
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_note_with_embedding_content must propagate the injected vector failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected vector failure"),
"error must carry injection message; got: {err_msg}"
);
let notes = rt.list_notes(&tok, None, 1000, 0).await.unwrap();
assert!(
notes.is_empty(),
"compensation must remove note row after vector failure; got {notes:?}"
);
}
#[tokio::test]
async fn soft_delete_entity_removes_indexes() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(
&tok,
"concept",
None,
"QuantumEntanglement",
Some("unique FTS term xzqjwv for soft delete test"),
None,
vec![],
)
.await
.unwrap();
let ns = tok.namespace().as_str().to_string();
let before = rt
.text(&tok)
.unwrap()
.search(TextSearchRequest {
query: "xzqjwv".to_string(),
mode: TextQueryMode::Plain,
filter: Some(TextFilter {
namespaces: vec![ns.clone()],
..Default::default()
}),
top_k: 10,
snippet_chars: 100,
})
.await
.unwrap();
assert!(
before.iter().any(|h| h.subject_id == entity.id),
"entity must be in FTS before soft-delete"
);
let deleted = rt.delete_entity(&tok, entity.id, false).await.unwrap();
assert!(deleted, "soft delete must return true");
let after = rt
.text(&tok)
.unwrap()
.search(TextSearchRequest {
query: "xzqjwv".to_string(),
mode: TextQueryMode::Plain,
filter: Some(TextFilter {
namespaces: vec![ns],
..Default::default()
}),
top_k: 10,
snippet_chars: 100,
})
.await
.unwrap();
assert!(
after.iter().all(|h| h.subject_id != entity.id),
"soft-deleted entity must be removed from FTS index"
);
}
#[tokio::test]
async fn soft_delete_note_removes_indexes() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"observation",
None,
"SpectralDecomposition unique term yvwkqz for soft delete test",
Some(0.7),
None,
vec![],
)
.await
.unwrap();
let before = rt
.search_notes(&tok, "yvwkqz", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(
before.iter().any(|h| h.note_id == note.id),
"note must be in FTS before soft-delete"
);
let deleted = rt.delete_note(&tok, note.id, false).await.unwrap();
assert!(deleted, "soft delete must return true");
let after = rt
.search_notes(&tok, "yvwkqz", None, 10, None, false, &[], None)
.await
.unwrap();
assert!(
after.iter().all(|h| h.note_id != note.id),
"soft-deleted note must be removed from FTS index"
);
}
#[tokio::test]
async fn link_extends_document_to_document_returns_invalid_input() {
let rt = rt();
let tok = NamespaceToken::local();
let d1 = rt
.create_entity(&tok, "document", None, "DocA", None, None, vec![])
.await
.unwrap();
let d2 = rt
.create_entity(&tok, "document", None, "DocB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, d1.id, d2.id, EdgeRelation::Extends, 1.0, None)
.await;
assert!(
result.is_err(),
"F010: document->document Extends must be rejected by the base allowlist; \
current generic entity fallthrough incorrectly accepts it"
);
}
#[tokio::test]
async fn link_extends_concept_to_concept_succeeds() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "CA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "CB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await;
assert!(
result.is_ok(),
"F010: concept->concept Extends must be allowed (base allowlist)"
);
}
#[tokio::test]
async fn link_symmetric_relation_canonicalizes_endpoint_order() {
use khive_storage::EdgeFilter;
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "ConceptP", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "ConceptQ", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::CompetesWith, 1.0, None)
.await
.unwrap();
rt.link(&tok, b.id, a.id, EdgeRelation::CompetesWith, 1.0, None)
.await
.unwrap();
let count = rt
.graph(&tok)
.unwrap()
.count_edges(EdgeFilter::default())
.await
.unwrap();
assert_eq!(
count,
1,
"F012: CompetesWith is symmetric; A->B and B->A must deduplicate to one canonical row; \
found {count} rows (canonicalization not yet implemented)"
);
}
#[tokio::test]
async fn f010_supersedes_document_to_document_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "document", None, "DocA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "document", None, "DocB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
result.is_ok(),
"document->document Supersedes must be allowed (allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_artifact_to_artifact_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "artifact", None, "ArtA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "artifact", None, "ArtB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
result.is_ok(),
"artifact->artifact Supersedes must be allowed (allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_service_to_service_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "service", None, "SvcA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "service", None, "SvcB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
result.is_ok(),
"service->service Supersedes must be allowed (allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_dataset_to_dataset_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "dataset", None, "DataA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "dataset", None, "DataB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
result.is_ok(),
"dataset->dataset Supersedes must be allowed (allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_project_to_project_rejected() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "project", None, "ProjA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "project", None, "ProjB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"project->project Supersedes must be rejected (not in allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_person_to_person_rejected() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "person", None, "Alice", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "person", None, "Bob", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"person->person Supersedes must be rejected (not in allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_org_to_org_rejected() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "org", None, "OrgA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "org", None, "OrgB", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"org->org Supersedes must be rejected (not in allowlist), got {result:?}"
);
}
#[tokio::test]
async fn f010_supersedes_same_kind_entity_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "OldV", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "NewV", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, b.id, a.id, EdgeRelation::Supersedes, 1.0, None)
.await;
assert!(
result.is_ok(),
"concept->concept Supersedes must be allowed by the base allowlist, got {result:?}"
);
}
#[tokio::test]
async fn f161_link_always_writes_null_target_backend() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
assert!(
edge.target_backend.is_none(),
"F161: target_backend must be None for locally-routed edges; got {:?}",
edge.target_backend
);
}
#[tokio::test]
async fn f161_link_many_always_writes_null_target_backend() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let c = rt
.create_entity(&tok, "concept", None, "C", None, None, vec![])
.await
.unwrap();
let specs = vec![
LinkSpec {
namespace: None,
source_id: a.id,
target_id: b.id,
relation: EdgeRelation::Extends,
weight: 1.0,
metadata: None,
},
LinkSpec {
namespace: None,
source_id: a.id,
target_id: c.id,
relation: EdgeRelation::Enables,
weight: 1.0,
metadata: None,
},
];
let edges = rt.link_many(&tok, specs).await.unwrap();
for edge in &edges {
assert!(
edge.target_backend.is_none(),
"F161: target_backend must be None for locally-routed edges in link_many; got {:?}",
edge.target_backend
);
}
}
#[tokio::test]
async fn f012_symmetric_neighbors_visible_from_both_endpoints() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::CompetesWith, 1.0, None)
.await
.unwrap();
let from_a = rt
.neighbors(
&tok,
a.id,
Direction::Out,
None,
Some(vec![EdgeRelation::CompetesWith]),
)
.await
.unwrap();
let from_b = rt
.neighbors(
&tok,
b.id,
Direction::Out,
None,
Some(vec![EdgeRelation::CompetesWith]),
)
.await
.unwrap();
assert_eq!(
from_a.len(),
1,
"node A must see competes_with neighbor from Direction::Out (F012); got {from_a:?}"
);
assert_eq!(
from_b.len(),
1,
"node B must see competes_with neighbor from Direction::Out (F012); got {from_b:?}"
);
}
#[tokio::test]
async fn f010_supersedes_cross_kind_entity_rejected() {
let rt = rt();
let tok = NamespaceToken::local();
let concept = rt
.create_entity(&tok, "concept", None, "MyConcept", None, None, vec![])
.await
.unwrap();
let doc = rt
.create_entity(&tok, "document", None, "MyDoc", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
concept.id,
doc.id,
EdgeRelation::Supersedes,
1.0,
None,
)
.await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"concept->document Supersedes must be rejected by the base allowlist, got {result:?}"
);
}
#[tokio::test]
async fn delete_note_cross_namespace_succeeds() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let note = rt
.create_note(
&ns_a,
"observation",
None,
"note in ns-a",
Some(0.8),
None,
vec![],
)
.await
.unwrap();
let result = rt.delete_note(&ns_b, note.id, false).await;
assert!(
result.unwrap(),
"cross-namespace delete_note (soft) must return Ok(true)"
);
let note_store = rt.notes(&ns_a).unwrap();
let gone = note_store.get_note(note.id).await.unwrap();
assert!(
gone.is_none(),
"note must be soft-deleted in its home namespace after cross-ns delete"
);
let note2 = rt
.create_note(
&ns_a,
"observation",
None,
"note2 in ns-a",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let hard_result = rt.delete_note(&ns_b, note2.id, true).await;
assert!(
hard_result.unwrap(),
"cross-namespace hard delete_note must return Ok(true)"
);
let gone2 = rt
.get_note_including_deleted(&ns_a, note2.id)
.await
.unwrap();
assert!(
gone2.is_none(),
"hard-deleted note must not appear even in including_deleted query"
);
}
#[tokio::test]
async fn link_many_overlapping_triple_returns_persisted_ids() {
let rt = rt();
let tok = NamespaceToken::local();
let a = rt
.create_entity(&tok, "concept", None, "A", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "B", None, None, vec![])
.await
.unwrap();
let spec = || LinkSpec {
namespace: None,
source_id: a.id,
target_id: b.id,
relation: EdgeRelation::Extends,
weight: 1.0,
metadata: None,
};
let first = rt.link_many(&tok, vec![spec()]).await.unwrap();
assert_eq!(first.len(), 1);
let persisted_id: Uuid = first[0].id.into();
let second = rt.link_many(&tok, vec![spec()]).await.unwrap();
assert_eq!(second.len(), 1);
let second_id: Uuid = second[0].id.into();
assert_eq!(
persisted_id, second_id,
"link_many with an existing triple must return the persisted row ID ({persisted_id}), \
not a new phantom ID ({second_id})"
);
let count = rt
.count_edges(&tok, crate::curation::EdgeListFilter::default())
.await
.unwrap();
assert_eq!(count, 1, "upsert must not duplicate the edge row");
}
#[tokio::test]
async fn create_many_persists_all_entities() {
let rt = rt();
let tok = NamespaceToken::local();
let specs: Vec<EntityCreateSpec> = (0..5)
.map(|i| EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: format!("BulkConcept-{i}"),
description: Some(format!("desc {i}")),
properties: None,
tags: vec!["bulk-test".into()],
})
.collect();
let entities = rt.create_many(&tok, specs).await.unwrap();
assert_eq!(entities.len(), 5, "all 5 entities must be returned");
for entity in &entities {
let fetched = rt.get_entity(&tok, entity.id).await.unwrap();
assert_eq!(fetched.id, entity.id);
}
}
#[tokio::test]
async fn create_many_empty_name_rejects_atomically() {
let rt = rt();
let tok = NamespaceToken::local();
let specs = vec![
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "ValidEntity".into(),
description: None,
properties: None,
tags: vec![],
},
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "".into(), description: None,
properties: None,
tags: vec![],
},
];
let result = rt.create_many(&tok, specs).await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"empty name must produce InvalidInput error"
);
let rows = rt.list_entities(&tok, None, None, 100, 0).await.unwrap();
assert_eq!(
rows.len(),
0,
"atomic rejection must leave storage unchanged"
);
}
#[tokio::test]
async fn create_many_rejects_unknown_entity_type_when_validator_installed() {
let rt = rt();
let tok = NamespaceToken::local();
rt.install_entity_type_validator(Arc::new(|kind, entity_type| {
let Some(raw) = entity_type else {
return Ok(None);
};
if kind == "concept" && raw == "algorithm" {
return Ok(Some("algorithm".to_string()));
}
Err(RuntimeError::InvalidInput(format!(
"unknown entity_type {raw:?} for {kind:?}; valid: algorithm"
)))
}));
let bad_spec = vec![EntityCreateSpec {
kind: "concept".into(),
entity_type: Some("not_a_registered_type".into()),
name: "ShouldNotLand".into(),
description: None,
properties: None,
tags: vec![],
}];
let result = rt.create_many(&tok, bad_spec).await;
assert!(
matches!(result, Err(RuntimeError::InvalidInput(_))),
"unknown entity_type must be rejected by the runtime-layer validator; got {result:?}"
);
let rows = rt.list_entities(&tok, None, None, 100, 0).await.unwrap();
assert_eq!(
rows.len(),
0,
"validator rejection must leave storage empty"
);
}
#[tokio::test]
async fn create_many_accepts_valid_entity_type_via_validator() {
let rt = rt();
let tok = NamespaceToken::local();
rt.install_entity_type_validator(Arc::new(|kind, entity_type| {
let Some(raw) = entity_type else {
return Ok(None);
};
if kind == "concept" && raw == "algorithm" {
return Ok(Some("algorithm".to_string()));
}
Err(RuntimeError::InvalidInput(format!(
"unknown entity_type {raw:?} for {kind:?}"
)))
}));
let specs = vec![EntityCreateSpec {
kind: "concept".into(),
entity_type: Some("algorithm".into()),
name: "BubbleSort".into(),
description: None,
properties: None,
tags: vec![],
}];
let entities = rt.create_many(&tok, specs).await.unwrap();
assert_eq!(entities.len(), 1, "valid entity_type must be accepted");
assert_eq!(
entities[0].entity_type.as_deref(),
Some("algorithm"),
"entity_type must be stored as returned by the validator"
);
}
#[tokio::test]
async fn create_many_fts_failure_rolls_back_both_substrates() {
let ns = format!("fts-fail-many-{}", uuid::Uuid::new_v4().as_simple());
let rt = rt();
let tok = NamespaceToken::for_namespace(Namespace::parse(&ns).unwrap());
let specs = vec![
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "FtsRollbackA".into(),
description: None,
properties: None,
tags: vec![],
},
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "FtsRollbackB".into(),
description: None,
properties: None,
tags: vec![],
},
];
arm_fts_fail_many(&ns);
let result = rt.create_many(&tok, specs).await;
assert!(
result.is_err(),
"create_many must return Err when FTS write fails"
);
let entity_rows = rt.list_entities(&tok, None, None, 100, 0).await.unwrap();
assert_eq!(
entity_rows.len(),
0,
"entity rows must be rolled back on FTS failure; found {entity_rows:?}"
);
let fts = rt.text(&tok).unwrap();
let fts_count = fts
.count(TextFilter {
ids: vec![],
kinds: vec![],
namespaces: vec![ns.clone()],
})
.await
.unwrap();
assert_eq!(
fts_count, 0,
"fts_entities must be empty after FTS-failure rollback; found {fts_count}"
);
}
#[tokio::test]
async fn create_many_fts_partial_failure_rolls_back_both_substrates() {
let ns = format!("fts-fail-partial-{}", uuid::Uuid::new_v4().as_simple());
let rt = rt();
let tok = NamespaceToken::for_namespace(Namespace::parse(&ns).unwrap());
let specs = vec![
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "PartialRollbackA".into(),
description: None,
properties: None,
tags: vec![],
},
EntityCreateSpec {
kind: "concept".into(),
entity_type: None,
name: "PartialRollbackB".into(),
description: None,
properties: None,
tags: vec![],
},
];
arm_fts_fail_many_partial(&ns);
let result = rt.create_many(&tok, specs).await;
assert!(
result.is_err(),
"create_many must return Err when FTS summary.failed > 0"
);
let entity_rows = rt.list_entities(&tok, None, None, 100, 0).await.unwrap();
assert_eq!(
entity_rows.len(),
0,
"entity rows must be rolled back when FTS summary.failed > 0; found {entity_rows:?}"
);
let fts = rt.text(&tok).unwrap();
let fts_count = fts
.count(TextFilter {
ids: vec![],
kinds: vec![],
namespaces: vec![ns.clone()],
})
.await
.unwrap();
assert_eq!(
fts_count, 0,
"fts_entities must be empty after partial-FTS-failure rollback; found {fts_count}"
);
}
#[tokio::test]
async fn get_edge_cross_namespace_succeeds() {
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let src = rt
.create_entity(&ns_a, "concept", None, "Src", None, None, vec![])
.await
.unwrap();
let tgt = rt
.create_entity(&ns_a, "concept", None, "Tgt", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&ns_a, src.id, tgt.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let own_ns = rt.get_edge(&ns_a, Uuid::from(edge.id)).await;
assert!(
own_ns.is_ok() && own_ns.unwrap().is_some(),
"edge must be visible in its own namespace"
);
let cross_ns = rt.get_edge(&ns_b, Uuid::from(edge.id)).await;
assert!(
matches!(cross_ns, Ok(Some(_))),
"cross-namespace get_edge must return Ok(Some(_)) after PR-A1, got {cross_ns:?}"
);
let absent = rt.get_edge(&ns_b, Uuid::new_v4()).await;
assert!(
matches!(absent, Ok(None)),
"absent edge must return Ok(None), got {absent:?}"
);
}
#[tokio::test]
async fn traverse_cross_namespace_root_is_accepted() {
use khive_storage::types::TraversalOptions;
let rt = rt();
let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
let a = rt
.create_entity(&ns_a, "concept", None, "A", None, None, vec![])
.await
.unwrap();
rt.create_entity(&ns_a, "concept", None, "B", None, None, vec![])
.await
.unwrap();
rt.link(&ns_a, a.id, a.id, EdgeRelation::Extends, 1.0, None)
.await
.ok();
let result = rt
.traverse(
&ns_b,
TraversalRequest {
roots: vec![a.id],
options: TraversalOptions {
max_depth: 1,
direction: Direction::Out,
..Default::default()
},
include_roots: true,
include_properties: false,
},
)
.await;
assert!(result.is_ok(), "traverse must not error; got {:?}", result);
}
async fn count_all_incident_edges(rt: &KhiveRuntime, node_id: Uuid, ns: &str) -> u64 {
let mut reader = rt.sql().reader().await.expect("sql reader must open");
let row = reader
.query_scalar(SqlStatement {
sql: "SELECT COUNT(*) FROM graph_edges \
WHERE namespace = ?1 AND (source_id = ?2 OR target_id = ?2)"
.into(),
params: vec![
SqlValue::Text(ns.to_string()),
SqlValue::Text(node_id.to_string()),
],
label: Some("count_all_incident_edges".into()),
})
.await
.expect("count query must succeed");
match row {
Some(SqlValue::Integer(n)) => n as u64,
_ => panic!("count must return an integer"),
}
}
#[tokio::test]
async fn hard_delete_entity_purges_already_soft_deleted_incident_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().to_string();
let a = rt
.create_entity(&tok, "concept", None, "SrcA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "TgtB", None, None, vec![])
.await
.unwrap();
rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_hit = rt
.neighbors(&tok, a.id, Direction::Out, None, None)
.await
.unwrap();
assert_eq!(edge_hit.len(), 1, "edge must exist before soft-delete");
let edge_uuid = edge_hit[0].edge_id;
rt.delete_edge(&tok, edge_uuid, false).await.unwrap();
let visible = rt
.neighbors(&tok, a.id, Direction::Out, None, None)
.await
.unwrap();
assert!(visible.is_empty(), "soft-deleted edge must be invisible");
let raw_before = count_all_incident_edges(&rt, a.id, &ns).await;
assert_eq!(
raw_before, 1,
"soft-deleted edge must still be a physical row"
);
rt.delete_entity(&tok, a.id, true).await.unwrap();
let raw_after = count_all_incident_edges(&rt, a.id, &ns).await;
assert_eq!(
raw_after, 0,
"purge_incident_edges must physically remove soft-deleted edge rows (ADR-002)"
);
}
#[tokio::test]
async fn hard_delete_note_purges_already_soft_deleted_incident_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().to_string();
let target = rt
.create_note(
&tok,
"observation",
None,
"purge-cascade target note",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let annotating = rt
.create_note(
&tok,
"insight",
None,
"annotator note",
Some(0.5),
None,
vec![target.id],
)
.await
.unwrap();
let edge_hit = rt
.neighbors(
&tok,
annotating.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(edge_hit.len(), 1, "annotates edge must exist");
let edge_uuid = edge_hit[0].edge_id;
rt.delete_edge(&tok, edge_uuid, false).await.unwrap();
let raw_before = count_all_incident_edges(&rt, target.id, &ns).await;
assert_eq!(
raw_before, 1,
"soft-deleted edge must still be a physical row before note purge"
);
rt.delete_note(&tok, target.id, true).await.unwrap();
let raw_after = count_all_incident_edges(&rt, target.id, &ns).await;
assert_eq!(
raw_after, 0,
"purge_incident_edges must physically remove soft-deleted edge rows on note purge (ADR-002)"
);
}
async fn count_all_incident_edges_global(rt: &KhiveRuntime, node_id: Uuid) -> u64 {
let mut reader = rt.sql().reader().await.expect("sql reader must open");
let row = reader
.query_scalar(SqlStatement {
sql: "SELECT COUNT(*) FROM graph_edges WHERE source_id = ?1 OR target_id = ?1"
.into(),
params: vec![SqlValue::Text(node_id.to_string())],
label: Some("count_all_incident_edges_global".into()),
})
.await
.expect("count query must succeed");
match row {
Some(SqlValue::Integer(n)) => n as u64,
_ => panic!("count must return an integer"),
}
}
#[tokio::test]
async fn cross_namespace_hard_delete_entity_purges_all_incident_edges() {
let rt = rt();
let ns_owner = NamespaceToken::for_namespace(Namespace::parse("ns-owner").unwrap());
let ns_caller = NamespaceToken::for_namespace(Namespace::parse("ns-caller").unwrap());
let entity = rt
.create_entity(
&ns_owner,
"concept",
None,
"ForeignEntity",
None,
None,
vec![],
)
.await
.unwrap();
let peer = rt
.create_entity(&ns_owner, "concept", None, "Peer", None, None, vec![])
.await
.unwrap();
rt.link(
&ns_owner,
entity.id,
peer.id,
EdgeRelation::Extends,
1.0,
None,
)
.await
.unwrap();
rt.link(
&ns_owner,
peer.id,
entity.id,
EdgeRelation::Extends,
1.0,
None,
)
.await
.unwrap();
let before = count_all_incident_edges_global(&rt, entity.id).await;
assert_eq!(before, 2, "two incident edges must exist before delete");
let deleted = rt.delete_entity(&ns_caller, entity.id, true).await.unwrap();
assert!(deleted, "cross-ns hard delete must return true");
let after = count_all_incident_edges_global(&rt, entity.id).await;
assert_eq!(
after, 0,
"purge_incident_edges must remove all incident edges across namespaces (ADR-002, ADR-007)"
);
}
async fn count_edge_rows_by_id(rt: &KhiveRuntime, edge_id: Uuid, ns: &str) -> u64 {
let mut reader = rt.sql().reader().await.expect("sql reader must open");
let row = reader
.query_scalar(SqlStatement {
sql: "SELECT COUNT(*) FROM graph_edges WHERE namespace = ?1 AND id = ?2".into(),
params: vec![
SqlValue::Text(ns.to_string()),
SqlValue::Text(edge_id.to_string()),
],
label: Some("count_edge_rows_by_id".into()),
})
.await
.expect("count query must succeed");
match row {
Some(SqlValue::Integer(n)) => n as u64,
_ => panic!("count must return an integer"),
}
}
#[tokio::test]
async fn hard_delete_edge_purges_already_soft_deleted_primary_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().to_string();
let a = rt
.create_entity(&tok, "concept", None, "EA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "EB", None, None, vec![])
.await
.unwrap();
let edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let edge_uuid: Uuid = edge.id.into();
let soft = rt.delete_edge(&tok, edge_uuid, false).await.unwrap();
assert!(soft, "soft delete must succeed");
assert!(
rt.get_edge(&tok, edge_uuid).await.unwrap().is_none(),
"soft-deleted edge must be invisible to get_edge"
);
assert_eq!(
count_edge_rows_by_id(&rt, edge_uuid, &ns).await,
1,
"soft-deleted edge must still be a physical row"
);
let purged = rt.delete_edge(&tok, edge_uuid, true).await.unwrap();
assert!(
purged,
"hard delete of a soft-deleted edge must return true"
);
assert_eq!(
count_edge_rows_by_id(&rt, edge_uuid, &ns).await,
0,
"hard-delete must physically remove the soft-deleted edge row (ADR-002)"
);
}
#[tokio::test]
async fn hard_delete_base_edge_purges_already_soft_deleted_annotates_edge() {
let rt = rt();
let tok = NamespaceToken::local();
let ns = tok.namespace().to_string();
let a = rt
.create_entity(&tok, "concept", None, "CA", None, None, vec![])
.await
.unwrap();
let b = rt
.create_entity(&tok, "concept", None, "CB", None, None, vec![])
.await
.unwrap();
let base_edge = rt
.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
.await
.unwrap();
let base_edge_uuid: Uuid = base_edge.id.into();
let note = rt
.create_note(
&tok,
"observation",
None,
"note about base edge",
Some(0.5),
None,
vec![base_edge_uuid],
)
.await
.unwrap();
let ann_hits = rt
.neighbors(
&tok,
note.id,
Direction::Out,
None,
Some(vec![EdgeRelation::Annotates]),
)
.await
.unwrap();
assert_eq!(ann_hits.len(), 1, "annotates edge must exist");
let ann_edge_uuid = ann_hits[0].edge_id;
rt.delete_edge(&tok, ann_edge_uuid, false).await.unwrap();
assert_eq!(
count_edge_rows_by_id(&rt, ann_edge_uuid, &ns).await,
1,
"soft-deleted annotates edge must still be a physical row"
);
let purged = rt.delete_edge(&tok, base_edge_uuid, true).await.unwrap();
assert!(purged, "hard delete of base edge must return true");
assert_eq!(
count_edge_rows_by_id(&rt, ann_edge_uuid, &ns).await,
0,
"hard-delete of base edge must purge already-soft-deleted annotates edge row (ADR-002)"
);
assert_eq!(
count_edge_rows_by_id(&rt, base_edge_uuid, &ns).await,
0,
"hard-delete must physically remove the base edge row"
);
}
#[tokio::test]
async fn create_entity_fts_failure_rolls_back_entity_row() {
let rt = KhiveRuntime::memory().unwrap();
let ns = Namespace::parse("fault-entity-fts").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_fts_fail(ns.as_str());
let result = rt
.create_entity(
&tok,
"concept",
None,
"fts-fail rollback target",
None,
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_entity must propagate the injected FTS failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected FTS failure"),
"error must carry injection message; got: {err_msg}"
);
let entities = rt.list_entities(&tok, None, None, 1000, 0).await.unwrap();
assert!(
entities.is_empty(),
"compensation must remove the entity row after FTS failure; got {entities:?}"
);
}
#[tokio::test]
async fn create_entity_vector_failure_rolls_back_entity_row_and_fts() {
const MODEL: &str = "test-entity-vec-inject";
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider, _counter) = ConstVecProvider::new(MODEL, DIMS);
rt.register_embedder(provider);
let ns = Namespace::parse("fault-entity-vec").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_vector_fail(ns.as_str());
let result = rt
.create_entity(
&tok,
"concept",
None,
"vec-fail rollback target",
Some("description so embed body is non-empty"),
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_entity must propagate the injected vector failure"
);
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("injected vector failure"),
"error must carry injection message; got: {err_msg}"
);
let entities = rt.list_entities(&tok, None, None, 1000, 0).await.unwrap();
assert!(
entities.is_empty(),
"compensation must remove entity row after vector failure; got {entities:?}"
);
use khive_storage::types::{TextFilter, TextQueryMode, TextSearchRequest};
let fts_hits = rt
.text(&tok)
.unwrap()
.search(TextSearchRequest {
query: "vec-fail rollback target".to_string(),
mode: TextQueryMode::Plain,
filter: Some(TextFilter {
namespaces: vec![ns.as_str().to_string()],
..Default::default()
}),
top_k: 10,
snippet_chars: 100,
})
.await
.unwrap();
assert!(
fts_hits.is_empty(),
"compensation must remove FTS document after vector failure; got {fts_hits:?}"
);
}
#[tokio::test]
async fn create_entity_multi_model_second_vector_failure_rolls_back_all() {
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider_a, _ca) = ConstVecProvider::new("model-a", DIMS);
let (provider_b, _cb) = ConstVecProvider::new("model-b", DIMS);
rt.register_embedder(provider_a);
rt.register_embedder(provider_b);
let ns = Namespace::parse("fault-entity-multi").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
arm_vector_fail_after(1);
let result = rt
.create_entity(
&tok,
"concept",
None,
"multi-model rollback target",
Some("description for embedding"),
None,
vec![],
)
.await;
assert!(
result.is_err(),
"create_entity must propagate the injected multi-model vector failure"
);
let entities = rt.list_entities(&tok, None, None, 1000, 0).await.unwrap();
assert!(
entities.is_empty(),
"compensation must remove entity row; got {entities:?}"
);
use khive_storage::types::VectorSearchRequest;
let query_vec = vec![1.0_f32; DIMS];
let hits_a = rt
.vectors_for_model(&tok, "model-a")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec.clone()],
top_k: 100,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Entity),
embedding_model: Some("model-a".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_a.is_empty(),
"model-a vector store must be empty after rollback; got {hits_a:?}"
);
let hits_b = rt
.vectors_for_model(&tok, "model-b")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec],
top_k: 100,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Entity),
embedding_model: Some("model-b".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_b.is_empty(),
"model-b vector store must be empty after rollback; got {hits_b:?}"
);
}
#[tokio::test]
async fn update_entity_fans_out_to_all_registered_models() {
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider_a, _ca) = ConstVecProvider::new("embed-a", DIMS);
let (provider_b, _cb) = ConstVecProvider::new("embed-b", DIMS);
rt.register_embedder(provider_a);
rt.register_embedder(provider_b);
let ns = Namespace::parse("update-entity-fanout").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
let entity = rt
.create_entity(
&tok,
"concept",
None,
"FanOutEntity",
Some("initial description"),
None,
vec![],
)
.await
.expect("create_entity must succeed");
use crate::curation::EntityPatch;
let patch = EntityPatch {
description: Some(Some("updated description after fan-out fix".to_string())),
..Default::default()
};
rt.update_entity(&tok, entity.id, patch)
.await
.expect("update_entity must succeed");
use khive_storage::types::VectorSearchRequest;
let query_vec = vec![1.0_f32; DIMS];
let hits_a = rt
.vectors_for_model(&tok, "embed-a")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec.clone()],
top_k: 10,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Entity),
embedding_model: Some("embed-a".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_a.iter().any(|h| h.subject_id == entity.id),
"embed-a must hold a vector for the entity after update; got {hits_a:?}"
);
let hits_b = rt
.vectors_for_model(&tok, "embed-b")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec],
top_k: 10,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Entity),
embedding_model: Some("embed-b".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_b.iter().any(|h| h.subject_id == entity.id),
"embed-b must hold a vector for the entity after update; got {hits_b:?}"
);
}
#[tokio::test]
async fn update_note_fans_out_to_all_registered_models() {
const DIMS: usize = 4;
let rt = KhiveRuntime::memory().unwrap();
let (provider_a, _ca) = ConstVecProvider::new("embed-a", DIMS);
let (provider_b, _cb) = ConstVecProvider::new("embed-b", DIMS);
rt.register_embedder(provider_a);
rt.register_embedder(provider_b);
let ns = Namespace::parse("update-note-fanout").unwrap();
let tok = NamespaceToken::for_namespace(ns.clone());
let note = rt
.create_note(
&tok,
"observation",
None,
"initial note content for fan-out test",
None,
None,
vec![],
)
.await
.expect("create_note must succeed");
use crate::curation::NotePatch;
let patch = NotePatch {
content: Some("updated content after fan-out fix".to_string()),
..Default::default()
};
rt.update_note(&tok, note.id, patch)
.await
.expect("update_note must succeed");
use khive_storage::types::VectorSearchRequest;
let query_vec = vec![1.0_f32; DIMS];
let hits_a = rt
.vectors_for_model(&tok, "embed-a")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec.clone()],
top_k: 10,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Note),
embedding_model: Some("embed-a".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_a.iter().any(|h| h.subject_id == note.id),
"embed-a must hold a vector for the note after update; got {hits_a:?}"
);
let hits_b = rt
.vectors_for_model(&tok, "embed-b")
.unwrap()
.search(VectorSearchRequest {
query_vectors: vec![query_vec],
top_k: 10,
namespace: Some(ns.as_str().to_string()),
kind: Some(khive_types::SubstrateKind::Note),
embedding_model: Some("embed-b".to_string()),
filter: None,
backend_hints: None,
})
.await
.unwrap();
assert!(
hits_b.iter().any(|h| h.subject_id == note.id),
"embed-b must hold a vector for the note after update; got {hits_b:?}"
);
}
#[tokio::test]
async fn get_entity_cross_namespace_succeeds() {
let rt = rt();
let leo_tok = NamespaceToken::for_namespace(Namespace::parse("lambda:leo").unwrap());
let entity = rt
.create_entity(&leo_tok, "concept", None, "Peer-Entity", None, None, vec![])
.await
.unwrap();
assert_eq!(entity.namespace, "lambda:leo");
let local_tok = NamespaceToken::local();
let fetched = rt.get_entity(&local_tok, entity.id).await;
assert!(
fetched.is_ok(),
"get_entity from local token must find lambda:leo entity; got {:?}",
fetched
);
assert_eq!(fetched.unwrap().id, entity.id);
}
#[tokio::test]
async fn update_entity_cross_namespace_succeeds() {
let rt = rt();
let leo_tok = NamespaceToken::for_namespace(Namespace::parse("lambda:leo").unwrap());
let entity = rt
.create_entity(
&leo_tok,
"concept",
None,
"Peer-Entity-Update",
None,
None,
vec![],
)
.await
.unwrap();
let local_tok = NamespaceToken::local();
let patch = crate::curation::EntityPatch {
name: Some("Peer-Entity-Updated".to_string()),
..Default::default()
};
let result = rt.update_entity(&local_tok, entity.id, patch).await;
assert!(
result.is_ok(),
"update_entity from local token must succeed on lambda:leo entity; got {:?}",
result
);
assert_eq!(result.unwrap().name, "Peer-Entity-Updated");
}
#[tokio::test]
async fn delete_entity_cross_namespace_succeeds() {
let rt = rt();
let leo_tok = NamespaceToken::for_namespace(Namespace::parse("lambda:leo").unwrap());
let entity = rt
.create_entity(
&leo_tok,
"concept",
None,
"Peer-Entity-Delete",
None,
None,
vec![],
)
.await
.unwrap();
let local_tok = NamespaceToken::local();
let deleted = rt.delete_entity(&local_tok, entity.id, false).await;
assert!(
deleted.is_ok(),
"delete_entity from local token must succeed on lambda:leo entity; got {:?}",
deleted
);
assert!(
deleted.unwrap(),
"delete must return true when entity existed"
);
}
#[tokio::test]
async fn namespace_preserved_on_entity_after_cross_namespace_get() {
let rt = rt();
let leo_tok = NamespaceToken::for_namespace(Namespace::parse("lambda:leo").unwrap());
let entity = rt
.create_entity(
&leo_tok,
"concept",
None,
"NS-Preserved",
None,
None,
vec![],
)
.await
.unwrap();
let local_tok = NamespaceToken::local();
let fetched = rt.get_entity(&local_tok, entity.id).await.unwrap();
assert_eq!(
fetched.namespace, "lambda:leo",
"namespace column must be preserved; not overwritten with caller's namespace"
);
}
use crate::pack::PackByIdResolver;
use tokio::sync::Mutex as TokioMutex;
#[derive(Debug, Default)]
struct MockResolverState {
owned: Vec<Uuid>,
deleted: Vec<Uuid>,
delete_calls: Vec<(Uuid, bool)>,
}
struct MockPackResolver(TokioMutex<MockResolverState>);
impl MockPackResolver {
fn new() -> Self {
Self(TokioMutex::new(MockResolverState::default()))
}
}
#[async_trait::async_trait]
impl crate::pack::PackByIdResolver for MockPackResolver {
async fn resolve_by_id(&self, id: Uuid) -> Result<Option<Resolved>, RuntimeError> {
let state = self.0.lock().await;
if state.owned.contains(&id) && !state.deleted.contains(&id) {
Ok(Some(Resolved::PackRecord {
pack: "mock".into(),
kind: "widget".into(),
data: serde_json::json!({ "id": id.to_string(), "name": "test-widget" }),
}))
} else {
Ok(None)
}
}
async fn resolve_by_id_including_deleted(
&self,
id: Uuid,
) -> Result<Option<Resolved>, RuntimeError> {
let state = self.0.lock().await;
if state.owned.contains(&id) {
Ok(Some(Resolved::PackRecord {
pack: "mock".into(),
kind: "widget".into(),
data: serde_json::json!({ "id": id.to_string(), "name": "test-widget" }),
}))
} else {
Ok(None)
}
}
async fn delete_by_id(
&self,
id: Uuid,
hard: bool,
) -> Result<serde_json::Value, RuntimeError> {
let mut state = self.0.lock().await;
if !state.owned.contains(&id) {
return Err(RuntimeError::NotFound(format!(
"mock widget not found: {id}"
)));
}
state.delete_calls.push((id, hard));
if hard {
state.owned.retain(|&x| x != id);
state.deleted.retain(|&x| x != id);
} else {
state.deleted.push(id);
}
Ok(
serde_json::json!({ "deleted": true, "id": id.to_string(), "kind": "widget", "hard": hard }),
)
}
}
fn registry_with_mock_resolver(
rt: KhiveRuntime,
resolver: Box<dyn crate::pack::PackByIdResolver>,
) -> crate::VerbRegistry {
use crate::pack::{PackRuntime, VerbRegistryBuilder};
use khive_types::{HandlerDef, VerbCategory, Visibility};
static MINIMAL_HANDLERS: &[HandlerDef] = &[HandlerDef {
name: "minimal.noop",
description: "noop",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &[],
}];
struct MinimalPack;
impl khive_types::Pack for MinimalPack {
const NAME: &'static str = "minimal";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = MINIMAL_HANDLERS;
}
#[async_trait::async_trait]
impl PackRuntime for MinimalPack {
fn name(&self) -> &str {
"minimal"
}
fn note_kinds(&self) -> &'static [&'static str] {
&[]
}
fn entity_kinds(&self) -> &'static [&'static str] {
&[]
}
fn handlers(&self) -> &'static [HandlerDef] {
MINIMAL_HANDLERS
}
async fn dispatch(
&self,
_verb: &str,
_params: serde_json::Value,
_registry: &crate::VerbRegistry,
_token: &NamespaceToken,
) -> Result<serde_json::Value, RuntimeError> {
Err(RuntimeError::InvalidInput("stub".into()))
}
}
let _ = rt;
let mut builder = VerbRegistryBuilder::new();
builder.register(MinimalPack);
builder.register_resolver("mock", resolver);
builder.build().expect("registry build failed")
}
#[tokio::test]
async fn pack_record_resolved_pair_returns_none() {
let pr = Resolved::PackRecord {
pack: "knowledge".into(),
kind: "atom".into(),
data: serde_json::json!({}),
};
assert!(
resolved_pair(Some(&pr)).is_none(),
"PackRecord must not be a valid edge endpoint"
);
}
#[test]
fn resolved_pair_surfaces_entity_type() {
let e = Resolved::Entity(
Entity::new("mathlib", "concept", "Nat.add_comm").with_entity_type(Some("theorem")),
);
assert_eq!(
resolved_pair(Some(&e)),
Some(("entity", "concept", Some("theorem"))),
"entity_type subtype must be surfaced alongside base kind"
);
}
#[test]
fn endpoint_of_type_matches_subtype_not_base_kind() {
let kind = "concept";
let et = Some("theorem");
assert!(endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
"entity",
kind,
et
));
assert!(!endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "definition",
},
"entity",
kind,
et
));
assert!(!endpoint_matches(
&EndpointKind::EntityOfKind("theorem"),
"entity",
kind,
et
));
assert!(endpoint_matches(
&EndpointKind::EntityOfKind("concept"),
"entity",
kind,
et
));
assert!(!endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
"note",
"task",
None
));
assert!(!endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
"entity",
kind,
None
));
}
#[test]
fn endpoint_of_type_requires_base_kind_match() {
let wrong_base_kind = "project"; let et = Some("theorem");
assert!(
!endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
"entity",
wrong_base_kind,
et
),
"EntityOfType must reject an entity whose base kind != rule.kind \
even when entity_type matches — the pre-fix bug admitted this"
);
assert!(endpoint_matches(
&EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
"entity",
"concept",
et
));
}
#[tokio::test]
async fn registry_resolvers_accessor_returns_registered() {
let resolver = Box::new(MockPackResolver::new());
let registry = registry_with_mock_resolver(rt(), resolver);
assert_eq!(registry.resolvers().len(), 1);
assert_eq!(registry.resolvers()[0].0, "mock");
}
#[tokio::test]
async fn mock_resolver_resolve_by_id_returns_pack_record() {
let id = Uuid::new_v4();
let resolver: Box<dyn PackByIdResolver> = Box::new(MockPackResolver::new());
let inner = MockPackResolver::new();
inner.0.lock().await.owned.push(id);
let result: Result<Option<Resolved>, RuntimeError> = inner.resolve_by_id(id).await;
match result.unwrap() {
Some(Resolved::PackRecord { pack, kind, data }) => {
assert_eq!(pack, "mock");
assert_eq!(kind, "widget");
assert_eq!(data["id"].as_str().unwrap(), id.to_string());
}
other => panic!("expected PackRecord, got {:?}", other),
}
let _ = resolver;
}
#[tokio::test]
async fn mock_resolver_resolve_unknown_uuid_returns_none() {
let inner = MockPackResolver::new();
let id = Uuid::new_v4();
let result: Result<Option<Resolved>, RuntimeError> = inner.resolve_by_id(id).await;
assert!(result.unwrap().is_none());
}
#[tokio::test]
async fn mock_resolver_delete_soft_records_call() {
let id = Uuid::new_v4();
let inner = MockPackResolver::new();
inner.0.lock().await.owned.push(id);
let result: Result<serde_json::Value, RuntimeError> = inner.delete_by_id(id, false).await;
let result = result.unwrap();
assert_eq!(result["deleted"], serde_json::json!(true));
assert_eq!(result["hard"], serde_json::json!(false));
let live: Result<Option<Resolved>, RuntimeError> = inner.resolve_by_id(id).await;
assert!(live.unwrap().is_none());
let incl: Result<Option<Resolved>, RuntimeError> =
inner.resolve_by_id_including_deleted(id).await;
assert!(incl.unwrap().is_some());
}
#[tokio::test]
async fn mock_resolver_delete_hard_removes_record() {
let id = Uuid::new_v4();
let inner = MockPackResolver::new();
inner.0.lock().await.owned.push(id);
let result: Result<serde_json::Value, RuntimeError> = inner.delete_by_id(id, true).await;
assert_eq!(result.unwrap()["hard"], serde_json::json!(true));
let incl: Result<Option<Resolved>, RuntimeError> =
inner.resolve_by_id_including_deleted(id).await;
assert!(incl.unwrap().is_none());
}
#[tokio::test]
async fn pack_record_not_valid_context_entity() {
let pr = Resolved::PackRecord {
pack: "knowledge".into(),
kind: "atom".into(),
data: serde_json::json!({}),
};
assert!(matches!(pr, Resolved::PackRecord { .. }));
}
fn neighbor_hit(node_id: Uuid) -> NeighborHit {
NeighborHit {
node_id,
edge_id: Uuid::new_v4(),
relation: EdgeRelation::Extends,
weight: 1.0,
name: None,
kind: None,
entity_type: None,
}
}
fn path_node(node_id: Uuid, depth: usize) -> PathNode {
PathNode {
node_id,
via_edge: None,
depth,
name: None,
kind: None,
properties: None,
}
}
#[tokio::test]
async fn enrich_neighbor_hits_batch_entity_note_and_bogus() {
let rt = rt();
let tok = NamespaceToken::local();
let entity = rt
.create_entity(&tok, "concept", None, "MyEntity", None, None, vec![])
.await
.unwrap();
let note = rt
.create_note(&tok, "observation", None, "body", Some(0.5), None, vec![])
.await
.unwrap();
let bogus_id = Uuid::new_v4();
let mut hits = vec![
neighbor_hit(entity.id),
neighbor_hit(note.id),
neighbor_hit(bogus_id),
];
rt.enrich_neighbor_hits(&tok, &mut hits).await;
assert_eq!(hits[0].name.as_deref(), Some("MyEntity"));
assert_eq!(hits[0].kind.as_deref(), Some("concept"));
assert_eq!(hits[1].name.as_deref(), Some("[observation]"));
assert_eq!(hits[1].kind.as_deref(), Some("observation"));
assert!(hits[2].name.is_none());
assert!(hits[2].kind.is_none());
}
#[tokio::test]
async fn enrich_neighbor_hits_note_with_name_uses_name() {
let rt = rt();
let tok = NamespaceToken::local();
let note = rt
.create_note(
&tok,
"insight",
Some("NoteTitle"),
"body",
Some(0.5),
None,
vec![],
)
.await
.unwrap();
let mut hits = vec![neighbor_hit(note.id)];
rt.enrich_neighbor_hits(&tok, &mut hits).await;
assert_eq!(hits[0].name.as_deref(), Some("NoteTitle"));
assert_eq!(hits[0].kind.as_deref(), Some("insight"));
}
#[tokio::test]
async fn enrich_path_nodes_batch_dedup_and_unresolved() {
let rt = rt();
let tok = NamespaceToken::local();
let ea = rt
.create_entity(&tok, "concept", None, "Alpha", None, None, vec![])
.await
.unwrap();
let eb = rt
.create_entity(&tok, "document", None, "Beta", None, None, vec![])
.await
.unwrap();
let bogus_id = Uuid::new_v4();
let mut paths = vec![
GraphPath {
root_id: ea.id,
nodes: vec![
path_node(ea.id, 0),
path_node(eb.id, 1),
path_node(bogus_id, 2),
],
total_weight: 1.0,
},
GraphPath {
root_id: eb.id,
nodes: vec![path_node(eb.id, 0), path_node(ea.id, 1)],
total_weight: 1.0,
},
];
rt.enrich_path_nodes(&tok, &mut paths, false).await;
assert_eq!(paths[0].nodes[0].name.as_deref(), Some("Alpha"));
assert_eq!(paths[0].nodes[0].kind.as_deref(), Some("concept"));
assert_eq!(paths[0].nodes[1].name.as_deref(), Some("Beta"));
assert_eq!(paths[0].nodes[1].kind.as_deref(), Some("document"));
assert!(paths[0].nodes[2].name.is_none());
assert!(paths[0].nodes[2].kind.is_none());
assert_eq!(paths[1].nodes[0].name.as_deref(), Some("Beta"));
assert_eq!(paths[1].nodes[0].kind.as_deref(), Some("document"));
assert_eq!(paths[1].nodes[1].name.as_deref(), Some("Alpha"));
assert_eq!(paths[1].nodes[1].kind.as_deref(), Some("concept"));
}
#[tokio::test]
async fn enrich_resolves_entities_in_extra_visible_namespace() {
let rt = KhiveRuntime::memory().unwrap();
let ns_a = Namespace::parse("enrich-ns-a").unwrap();
let ns_b = Namespace::parse("enrich-ns-b").unwrap();
let tok_b = rt.authorize(ns_b.clone()).unwrap();
let entity_b = rt
.create_entity(&tok_b, "concept", None, "EntityInB", None, None, vec![])
.await
.unwrap();
assert_eq!(entity_b.namespace, "enrich-ns-b");
let vis_tok = rt
.authorize_with_visibility(ns_a.clone(), vec![ns_b.clone()])
.unwrap();
let mut hits = vec![neighbor_hit(entity_b.id)];
rt.enrich_neighbor_hits(&vis_tok, &mut hits).await;
assert_eq!(
hits[0].name.as_deref(),
Some("EntityInB"),
"entity in extra-visible ns must be enriched by enrich_neighbor_hits"
);
assert_eq!(hits[0].kind.as_deref(), Some("concept"));
let mut paths = vec![GraphPath {
root_id: entity_b.id,
nodes: vec![path_node(entity_b.id, 0)],
total_weight: 1.0,
}];
rt.enrich_path_nodes(&vis_tok, &mut paths, false).await;
assert_eq!(
paths[0].nodes[0].name.as_deref(),
Some("EntityInB"),
"entity in extra-visible ns must be enriched by enrich_path_nodes"
);
assert_eq!(paths[0].nodes[0].kind.as_deref(), Some("concept"));
}
#[tokio::test]
async fn enrich_neighbor_hits_populates_entity_type() {
let rt = rt();
let tok = NamespaceToken::local();
let props = serde_json::json!({"domain": "attention"});
let entity = rt
.create_entity(
&tok,
"concept",
Some("algorithm"),
"FlashAttn",
None,
Some(props),
vec![],
)
.await
.unwrap();
let entity_no_type = rt
.create_entity(&tok, "concept", None, "PlainConcept", None, None, vec![])
.await
.unwrap();
let mut hits = vec![neighbor_hit(entity.id), neighbor_hit(entity_no_type.id)];
rt.enrich_neighbor_hits(&tok, &mut hits).await;
assert_eq!(hits[0].entity_type.as_deref(), Some("algorithm"));
assert!(
hits[1].entity_type.is_none(),
"entity without entity_type must leave the field as None"
);
}
#[tokio::test]
async fn enrich_path_nodes_populates_properties() {
let rt = rt();
let tok = NamespaceToken::local();
let props = serde_json::json!({"year": 2024, "venue": "NeurIPS"});
let entity_with_props = rt
.create_entity(
&tok,
"document",
None,
"AttentionPaper",
None,
Some(props.clone()),
vec![],
)
.await
.unwrap();
let entity_no_props = rt
.create_entity(&tok, "concept", None, "BareConceptNode", None, None, vec![])
.await
.unwrap();
let mut paths = vec![GraphPath {
root_id: entity_with_props.id,
nodes: vec![
path_node(entity_with_props.id, 0),
path_node(entity_no_props.id, 1),
],
total_weight: 1.0,
}];
rt.enrich_path_nodes(&tok, &mut paths, true).await;
assert_eq!(
paths[0].nodes[0].properties.as_ref(),
Some(&props),
"properties must be filled when entity has a non-null properties blob"
);
assert!(
paths[0].nodes[1].properties.is_none(),
"entity without properties must leave the field as None"
);
}
#[tokio::test]
async fn traverse_chunks_root_binds_over_host_param_limit() {
use khive_storage::types::TraversalOptions;
let rt = rt();
let tok = NamespaceToken::local();
let graph = rt.graph(&tok).unwrap();
const N: usize = 1_000;
let now = chrono::Utc::now();
let mut roots: Vec<uuid::Uuid> = Vec::with_capacity(N);
let mut expected_children: std::collections::HashMap<uuid::Uuid, uuid::Uuid> =
std::collections::HashMap::with_capacity(N);
for _ in 0..N {
let root = uuid::Uuid::new_v4();
let child = uuid::Uuid::new_v4();
graph
.upsert_edge(Edge {
id: LinkId::from(uuid::Uuid::new_v4()),
namespace: "local".to_string(),
source_id: root,
target_id: child,
relation: EdgeRelation::Extends,
weight: 1.0,
created_at: now,
updated_at: now,
deleted_at: None,
metadata: None,
target_backend: None,
})
.await
.unwrap();
roots.push(root);
expected_children.insert(root, child);
}
let paths = graph
.traverse(TraversalRequest {
roots: roots.clone(),
options: TraversalOptions {
max_depth: 1,
direction: Direction::Out,
relations: None,
min_weight: None,
limit: None,
},
include_roots: false,
include_properties: false,
})
.await
.unwrap();
assert_eq!(
paths.len(),
N,
"traverse over {N} roots must return one GraphPath per root"
);
for path in &paths {
let expected_child = expected_children[&path.root_id];
assert_eq!(
path.nodes.len(),
1,
"root {:?} must reach exactly 1 node",
path.root_id
);
assert_eq!(
path.nodes[0].node_id, expected_child,
"root {:?} must reach its direct child",
path.root_id
);
}
}
#[test]
fn pack_entity_of_type_rules_do_not_shadow_base_entity_of_kind_rule() {
let pack_rules: Vec<EdgeEndpointRule> = vec![
EdgeEndpointRule {
relation: EdgeRelation::VariantOf,
source: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "goal",
},
target: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
},
EdgeEndpointRule {
relation: EdgeRelation::VariantOf,
source: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "goal",
},
target: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "definition",
},
},
];
let goal_a =
Resolved::Entity(Entity::new("local", "concept", "G-a").with_entity_type(Some("goal")));
let goal_b =
Resolved::Entity(Entity::new("local", "concept", "G-b").with_entity_type(Some("goal")));
assert!(
!pack_rule_allows(
&pack_rules,
EdgeRelation::VariantOf,
Some(&goal_a),
Some(&goal_b)
),
"pack rules must not cover goal->goal variant_of (no such rule declared)"
);
assert!(
base_entity_rule_allows("concept", EdgeRelation::VariantOf, "concept"),
"base contract must allow concept->concept variant_of regardless of pack EntityOfType rules"
);
}
#[tokio::test]
async fn link_variant_of_goal_to_goal_allowed_when_pack_has_entity_of_type_rules() {
let rt = rt();
let tok = NamespaceToken::local();
rt.install_edge_rules(vec![
EdgeEndpointRule {
relation: EdgeRelation::VariantOf,
source: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "goal",
},
target: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
},
EdgeEndpointRule {
relation: EdgeRelation::VariantOf,
source: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "goal",
},
target: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "definition",
},
},
]);
let a = rt
.create_entity(
&tok,
"concept",
Some("goal"),
"Goal Alpha",
None,
None,
vec![],
)
.await
.unwrap();
let b = rt
.create_entity(
&tok,
"concept",
Some("goal"),
"Goal Beta",
None,
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(&tok, a.id, b.id, EdgeRelation::VariantOf, 1.0, None)
.await;
assert!(
result.is_ok(),
"goal->goal variant_of must be allowed via the base concept->concept rule \
even when EntityOfType rules for variant_of are installed; got {result:?}"
);
let p = rt
.create_entity(&tok, "project", None, "Proj", None, None, vec![])
.await
.unwrap();
let bad = rt
.link(&tok, a.id, p.id, EdgeRelation::VariantOf, 1.0, None)
.await;
assert!(
bad.is_err(),
"additive pack rules must not make validation fail-open; \
goal(concept)->project variant_of must be rejected (pack miss + base miss); \
got {bad:?}"
);
}
#[tokio::test]
async fn link_depends_on_theorem_to_definition_allowed_only_via_pack_rule() {
let rt = rt();
let tok = NamespaceToken::local();
assert!(
!base_entity_rule_allows("concept", EdgeRelation::DependsOn, "concept"),
"base contract must not allow concept->concept DependsOn; \
test would be vacuous if this precondition fails"
);
rt.install_edge_rules(vec![EdgeEndpointRule {
relation: EdgeRelation::DependsOn,
source: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "theorem",
},
target: EndpointKind::EntityOfType {
kind: "concept",
entity_type: "definition",
},
}]);
let thm = rt
.create_entity(&tok, "concept", Some("theorem"), "T1", None, None, vec![])
.await
.unwrap();
let def = rt
.create_entity(
&tok,
"concept",
Some("definition"),
"D1",
None,
None,
vec![],
)
.await
.unwrap();
let result = rt
.link(&tok, thm.id, def.id, EdgeRelation::DependsOn, 1.0, None)
.await;
assert!(
result.is_ok(),
"theorem->definition DependsOn must be allowed by the installed pack rule; \
the base contract has no concept->concept DependsOn row; got {result:?}"
);
}
#[tokio::test]
async fn link_document_introduced_by_person_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let doc = rt
.create_entity(&tok, "document", None, "Paper", None, None, vec![])
.await
.unwrap();
let author = rt
.create_entity(&tok, "person", None, "Author", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
doc.id,
author.id,
EdgeRelation::IntroducedBy,
1.0,
None,
)
.await;
assert!(
result.is_ok(),
"document->person introduced_by must be allowed by the ADR-002 \
endpoint amendment; got {result:?}"
);
}
#[tokio::test]
async fn link_document_introduced_by_org_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let doc = rt
.create_entity(&tok, "document", None, "Whitepaper", None, None, vec![])
.await
.unwrap();
let org = rt
.create_entity(&tok, "org", None, "Publisher", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, doc.id, org.id, EdgeRelation::IntroducedBy, 1.0, None)
.await;
assert!(
result.is_ok(),
"document->org introduced_by must be allowed by the ADR-002 \
endpoint amendment; got {result:?}"
);
}
#[tokio::test]
async fn link_concept_introduced_by_org_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let concept = rt
.create_entity(&tok, "concept", None, "Architecture", None, None, vec![])
.await
.unwrap();
let org = rt
.create_entity(&tok, "org", None, "Originator", None, None, vec![])
.await
.unwrap();
let result = rt
.link(
&tok,
concept.id,
org.id,
EdgeRelation::IntroducedBy,
1.0,
None,
)
.await;
assert!(
result.is_ok(),
"concept->org introduced_by must be allowed by the ADR-002 \
endpoint amendment; got {result:?}"
);
}
#[tokio::test]
async fn link_document_depends_on_document_allowed() {
let rt = rt();
let tok = NamespaceToken::local();
let doc_a = rt
.create_entity(&tok, "document", None, "Spec A", None, None, vec![])
.await
.unwrap();
let doc_b = rt
.create_entity(&tok, "document", None, "Spec B", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, doc_a.id, doc_b.id, EdgeRelation::DependsOn, 1.0, None)
.await;
assert!(
result.is_ok(),
"document->document depends_on must be allowed by the ADR-002 \
endpoint amendment; got {result:?}"
);
let edge = result.unwrap();
let dk = edge
.metadata
.as_ref()
.and_then(|m| m.get("dependency_kind"))
.and_then(|v| v.as_str());
assert_eq!(
dk,
Some("normative"),
"document->document depends_on must infer dependency_kind=normative"
);
}
#[tokio::test]
async fn link_org_introduced_by_document_rejected_direction_matters() {
let rt = rt();
let tok = NamespaceToken::local();
let org = rt
.create_entity(&tok, "org", None, "Publisher", None, None, vec![])
.await
.unwrap();
let doc = rt
.create_entity(&tok, "document", None, "Paper", None, None, vec![])
.await
.unwrap();
let result = rt
.link(&tok, org.id, doc.id, EdgeRelation::IntroducedBy, 1.0, None)
.await;
assert!(
result.is_err(),
"org->document introduced_by must remain rejected; only \
document->org is permitted, not the reverse; got {result:?}"
);
}
struct CapturingVecService {
dims: usize,
captured: Arc<std::sync::Mutex<Vec<String>>>,
}
#[async_trait]
impl EmbeddingService for CapturingVecService {
async fn embed(
&self,
texts: &[String],
_model: EmbeddingModel,
) -> std::result::Result<Vec<Vec<f32>>, EmbedError> {
self.captured.lock().unwrap().extend(texts.iter().cloned());
Ok(texts.iter().map(|_| vec![1.0_f32; self.dims]).collect())
}
fn supports_model(&self, _model: EmbeddingModel) -> bool {
true
}
fn name(&self) -> &'static str {
"capturing-vec"
}
}
struct CapturingVecProvider {
provider_name: String,
dims: usize,
captured: Arc<std::sync::Mutex<Vec<String>>>,
}
#[async_trait]
impl EmbedderProvider for CapturingVecProvider {
fn name(&self) -> &str {
&self.provider_name
}
fn dimensions(&self) -> usize {
self.dims
}
async fn build(&self) -> crate::error::RuntimeResult<Arc<dyn EmbeddingService>> {
Ok(Arc::new(CapturingVecService {
dims: self.dims,
captured: Arc::clone(&self.captured),
}))
}
}
#[tokio::test]
async fn create_note_with_embedding_content_none_matches_create_note() {
let rt = rt();
let tok = NamespaceToken::local();
let captured = Arc::new(std::sync::Mutex::new(Vec::new()));
rt.register_embedder(CapturingVecProvider {
provider_name: "capturing-vec".into(),
dims: 4,
captured: Arc::clone(&captured),
});
let note = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
"full content, no override",
None,
None,
None,
vec![],
)
.await
.expect("create with None override must behave like create_note");
assert_eq!(note.content, "full content, no override");
let seen = captured.lock().unwrap().clone();
assert_eq!(
seen,
vec!["full content, no override".to_string()],
"with no override the embedder must see the full content"
);
}
#[tokio::test]
async fn create_note_with_embedding_content_embeds_capped_override_and_stores_full_content() {
let rt = rt();
let tok = NamespaceToken::local();
let captured = Arc::new(std::sync::Mutex::new(Vec::new()));
rt.register_embedder(CapturingVecProvider {
provider_name: "capturing-vec".into(),
dims: 4,
captured: Arc::clone(&captured),
});
let full = "head-term and then a very long tail-term that exceeds any cap";
let head = &full[.."head-term and then a very long".len()];
let note = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
full,
Some(head),
None,
None,
vec![],
)
.await
.expect("proper-prefix override must be accepted");
assert_eq!(note.content, full, "stored content must be the full text");
let seen = captured.lock().unwrap().clone();
assert_eq!(
seen,
vec![head.to_string()],
"embedder must see only the capped override"
);
}
#[tokio::test]
async fn create_note_with_embedding_content_fans_out_identical_override_to_multiple_models() {
let rt = rt();
let tok = NamespaceToken::local();
let captured_a = Arc::new(std::sync::Mutex::new(Vec::new()));
let captured_b = Arc::new(std::sync::Mutex::new(Vec::new()));
rt.register_embedder(CapturingVecProvider {
provider_name: "capturing-vec-a".into(),
dims: 4,
captured: Arc::clone(&captured_a),
});
rt.register_embedder(CapturingVecProvider {
provider_name: "capturing-vec-b".into(),
dims: 4,
captured: Arc::clone(&captured_b),
});
let full = "head-only-embedded plus a long discarded tail";
let head = &full[.."head-only-embedded".len()];
rt.create_note_with_embedding_content(
&tok,
"observation",
None,
full,
Some(head),
None,
None,
vec![],
)
.await
.expect("create ok");
assert_eq!(
captured_a.lock().unwrap().clone(),
vec![head.to_string()],
"model A must receive the identical capped override"
);
assert_eq!(
captured_b.lock().unwrap().clone(),
vec![head.to_string()],
"model B must receive the identical capped override"
);
let vs_a = rt
.vectors_for_model(&tok, "capturing-vec-a")
.expect("vector store for model A");
assert_eq!(
vs_a.count().await.expect("vector count A"),
1,
"model A must have exactly one vector row for the note"
);
let vs_b = rt
.vectors_for_model(&tok, "capturing-vec-b")
.expect("vector store for model B");
assert_eq!(
vs_b.count().await.expect("vector count B"),
1,
"model B must have exactly one vector row for the note"
);
}
#[tokio::test]
async fn create_note_with_embedding_content_rejects_empty_override() {
let rt = rt();
let tok = NamespaceToken::local();
let err = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
"some content",
Some(""),
None,
None,
vec![],
)
.await
.expect_err("empty override must be rejected");
assert!(matches!(err, RuntimeError::InvalidInput(_)));
}
#[tokio::test]
async fn create_note_with_embedding_content_rejects_non_prefix_override() {
let rt = rt();
let tok = NamespaceToken::local();
let err = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
"the actual content",
Some("an unrelated string"),
None,
None,
vec![],
)
.await
.expect_err("non-prefix override must be rejected");
assert!(matches!(err, RuntimeError::InvalidInput(ref m) if m.contains("prefix")));
}
#[tokio::test]
async fn create_note_with_embedding_content_rejects_equal_length_override() {
let rt = rt();
let tok = NamespaceToken::local();
let err = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
"identical text",
Some("identical text"),
None,
None,
vec![],
)
.await
.expect_err("an equal-length override must be rejected as not a proper prefix");
assert!(matches!(err, RuntimeError::InvalidInput(ref m) if m.contains("prefix")));
}
#[tokio::test]
async fn create_note_with_embedding_content_rejects_secret_bearing_override() {
let rt = rt();
let tok = NamespaceToken::local();
let token_span = "ghp_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
let content = format!("{token_span} plus extra trailing content beyond the override");
let embedding_content = format!("{token_span} plus extra");
let err = rt
.create_note_with_embedding_content(
&tok,
"observation",
None,
&content,
Some(&embedding_content),
None,
None,
vec![],
)
.await
.expect_err("a credential-shaped override must fail the secret gate");
assert!(
matches!(err, RuntimeError::SecretDetected(_)),
"expected SecretDetected, got {err:?}"
);
let count = rt
.notes(&tok)
.unwrap()
.count_notes(tok.namespace().as_str(), None)
.await
.unwrap();
assert_eq!(count, 0, "a rejected create must leave no note behind");
}
}