use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::time::Instant;
use crate::runtime::NamespaceToken;
use async_trait::async_trait;
use khive_gate::{AllowAllGate, AuditEvent, GateDecision, GateRef, GateRequest};
use khive_storage::{Event, EventStore, EventView, SubstrateKind};
use khive_types::{EventKind, EventOutcome, Namespace};
use serde_json::Value;
pub use khive_types::{
EdgeEndpointRule, EndpointKind, HandlerDef, NoteKindSpec, NoteLifecycleSpec, PackSchemaPlan,
ParamDef, VerbCategory, VerbPresentationPolicy, Visibility,
};
#[allow(deprecated)]
pub use khive_types::VerbDef;
use crate::validation::ValidationRule;
#[derive(Debug, Default, Clone)]
pub struct SchemaPlan {
pub pack: &'static str,
pub statements: &'static [&'static str],
}
impl SchemaPlan {
pub const fn empty() -> Self {
Self {
pack: "",
statements: &[],
}
}
pub fn is_empty(&self) -> bool {
self.statements.is_empty()
}
}
#[async_trait]
pub trait DispatchHook: Send + Sync {
async fn on_dispatch(&self, view: &EventView);
}
use crate::error::{
CircularPackDependency, MissingPackDependencies, MissingPackDependency, RuntimeError,
};
use crate::KhiveRuntime;
#[async_trait]
pub trait PackRuntime: Send + Sync {
fn name(&self) -> &str;
fn note_kinds(&self) -> &'static [&'static str];
fn entity_kinds(&self) -> &'static [&'static str];
fn handlers(&self) -> &'static [HandlerDef];
fn edge_rules(&self) -> &'static [EdgeEndpointRule] {
&[]
}
fn requires(&self) -> &'static [&'static str] {
&[]
}
fn note_kind_specs(&self) -> &'static [NoteKindSpec] {
&[]
}
fn kind_hook(&self, _kind: &str) -> Option<Arc<dyn KindHook>> {
None
}
fn schema_plan(&self) -> SchemaPlan {
SchemaPlan::empty()
}
fn validation_rules(&self) -> &'static [ValidationRule] {
&[]
}
fn register_embedders(&self, _runtime: &KhiveRuntime) {}
fn register_entity_type_validator(&self, _runtime: &KhiveRuntime) {}
fn register_note_mutation_hook(&self, _runtime: &KhiveRuntime) {}
async fn warm(&self) {}
async fn dispatch(
&self,
verb: &str,
params: Value,
registry: &VerbRegistry,
token: &NamespaceToken,
) -> Result<Value, RuntimeError>;
}
#[async_trait]
pub trait KindHook: Send + Sync + std::fmt::Debug {
async fn prepare_create(
&self,
runtime: &KhiveRuntime,
args: &mut Value,
) -> Result<(), RuntimeError>;
async fn after_create(
&self,
runtime: &KhiveRuntime,
id: uuid::Uuid,
args: &Value,
) -> Result<(), RuntimeError>;
}
#[async_trait]
pub trait PackByIdResolver: Send + Sync {
async fn resolve_by_id(
&self,
id: uuid::Uuid,
) -> Result<Option<crate::Resolved>, crate::RuntimeError>;
async fn resolve_by_id_including_deleted(
&self,
id: uuid::Uuid,
) -> Result<Option<crate::Resolved>, crate::RuntimeError> {
self.resolve_by_id(id).await
}
async fn delete_by_id(
&self,
id: uuid::Uuid,
hard: bool,
) -> Result<serde_json::Value, crate::RuntimeError>;
}
pub struct VerbRegistryBuilder {
packs: Vec<Box<dyn PackRuntime>>,
resolvers: Vec<(String, Box<dyn PackByIdResolver>)>,
gate: GateRef,
default_namespace: String,
visible_namespaces: Vec<Namespace>,
actor_id: Option<String>,
event_store: Option<Arc<dyn EventStore>>,
dispatch_hook: Option<Arc<dyn DispatchHook>>,
}
impl VerbRegistryBuilder {
pub fn new() -> Self {
Self {
packs: Vec::new(),
resolvers: Vec::new(),
gate: std::sync::Arc::new(AllowAllGate),
default_namespace: Namespace::local().as_str().to_string(),
visible_namespaces: vec![],
actor_id: None,
event_store: None,
dispatch_hook: None,
}
}
pub fn with_visible_namespaces(&mut self, ns: Vec<Namespace>) -> &mut Self {
self.visible_namespaces = ns;
self
}
pub fn with_actor_id(&mut self, actor_id: Option<String>) -> &mut Self {
self.actor_id = actor_id;
self
}
pub fn register<P: khive_types::Pack + PackRuntime + 'static>(&mut self, pack: P) -> &mut Self {
self.packs.push(Box::new(pack));
self
}
pub(crate) fn register_boxed(&mut self, pack: Box<dyn PackRuntime>) -> &mut Self {
self.packs.push(pack);
self
}
pub fn register_resolver(
&mut self,
name: impl Into<String>,
resolver: Box<dyn PackByIdResolver>,
) -> &mut Self {
self.resolvers.push((name.into(), resolver));
self
}
pub fn with_gate(&mut self, gate: GateRef) -> &mut Self {
self.gate = gate;
self
}
pub fn with_default_namespace(&mut self, ns: impl Into<String>) -> &mut Self {
self.default_namespace = ns.into();
self
}
pub fn with_event_store(&mut self, store: Arc<dyn EventStore>) -> &mut Self {
self.event_store = Some(store);
self
}
pub fn with_dispatch_hook(&mut self, hook: Arc<dyn DispatchHook>) -> &mut Self {
self.dispatch_hook = Some(hook);
self
}
pub fn build(self) -> Result<VerbRegistry, RuntimeError> {
let packs = self.packs;
let mut name_to_idx: HashMap<&str, usize> = HashMap::with_capacity(packs.len());
for (idx, pack) in packs.iter().enumerate() {
if let Some(prev_idx) = name_to_idx.insert(pack.name(), idx) {
return Err(RuntimeError::PackRedeclared {
name: pack.name().to_string(),
first_idx: prev_idx,
second_idx: idx,
});
}
}
let mut missing: Vec<MissingPackDependency> = Vec::new();
let mut indegree = vec![0usize; packs.len()];
let mut dependents: Vec<Vec<usize>> = vec![Vec::new(); packs.len()];
for (idx, pack) in packs.iter().enumerate() {
for &requires in pack.requires() {
match name_to_idx.get(requires).copied() {
Some(dep_idx) => {
dependents[dep_idx].push(idx);
indegree[idx] += 1;
}
None => missing.push(MissingPackDependency {
from: pack.name().to_string(),
requires: requires.to_string(),
}),
}
}
}
if !missing.is_empty() {
return if missing.len() == 1 {
Err(RuntimeError::MissingPackDependency(missing.remove(0)))
} else {
Err(RuntimeError::MissingPackDependencies(
MissingPackDependencies { missing },
))
};
}
let mut ready: VecDeque<usize> = indegree
.iter()
.enumerate()
.filter_map(|(idx, degree)| (*degree == 0).then_some(idx))
.collect();
let mut ordered_indices = Vec::with_capacity(packs.len());
while let Some(idx) = ready.pop_front() {
ordered_indices.push(idx);
for &dep_idx in &dependents[idx] {
indegree[dep_idx] -= 1;
if indegree[dep_idx] == 0 {
ready.push_back(dep_idx);
}
}
}
if ordered_indices.len() != packs.len() {
let cycle_nodes: HashSet<usize> = indegree
.iter()
.enumerate()
.filter_map(|(idx, degree)| (*degree > 0).then_some(idx))
.collect();
let cycle = find_pack_dependency_cycle(&packs, &name_to_idx, &cycle_nodes);
return Err(RuntimeError::CircularPackDependency(
CircularPackDependency { cycle },
));
}
let mut slots: Vec<Option<Box<dyn PackRuntime>>> = packs.into_iter().map(Some).collect();
let ordered_packs: Vec<Box<dyn PackRuntime>> = ordered_indices
.into_iter()
.map(|idx| slots[idx].take().expect("topological index must exist"))
.collect();
validate_unique_note_kinds(&ordered_packs)?;
validate_unique_verb_names(&ordered_packs)?;
let available_verbs: Vec<&'static str> = ordered_packs
.iter()
.flat_map(|p| p.handlers().iter())
.filter(|h| matches!(h.visibility, Visibility::Verb))
.map(|h| h.name)
.collect();
Ok(VerbRegistry {
packs: Arc::new(ordered_packs),
resolvers: Arc::new(self.resolvers),
gate: self.gate,
default_namespace: self.default_namespace,
visible_namespaces: self.visible_namespaces,
actor_id: self.actor_id,
event_store: self.event_store,
dispatch_hook: self.dispatch_hook,
available_verbs: Arc::new(available_verbs),
reference_ring: Arc::new(crate::reference_ring::ReferenceRing::new()),
})
}
}
fn validate_unique_note_kinds(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
let mut seen: HashMap<&str, &str> = HashMap::new();
for pack in packs {
for &kind in pack.note_kinds() {
if let Some(first_pack) = seen.insert(kind, pack.name()) {
return Err(RuntimeError::InvalidInput(format!(
"duplicate note kind {kind:?}: claimed by both {first_pack:?} and {:?}",
pack.name()
)));
}
}
}
Ok(())
}
fn validate_unique_verb_names(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
let mut seen: HashMap<&str, &str> = HashMap::new();
for pack in packs {
for handler in pack.handlers() {
if !matches!(handler.visibility, Visibility::Verb) {
continue;
}
if let Some(first_pack) = seen.insert(handler.name, pack.name()) {
return Err(RuntimeError::VerbCollision {
verb: handler.name.to_string(),
first_pack: first_pack.to_string(),
second_pack: pack.name().to_string(),
});
}
}
}
Ok(())
}
fn find_pack_dependency_cycle(
packs: &[Box<dyn PackRuntime>],
name_to_idx: &HashMap<&str, usize>,
cycle_nodes: &HashSet<usize>,
) -> Vec<String> {
fn visit(
idx: usize,
packs: &[Box<dyn PackRuntime>],
name_to_idx: &HashMap<&str, usize>,
cycle_nodes: &HashSet<usize>,
visiting: &mut Vec<usize>,
visited: &mut HashSet<usize>,
) -> Option<Vec<String>> {
if let Some(pos) = visiting.iter().position(|&seen| seen == idx) {
let mut cycle: Vec<String> = visiting[pos..]
.iter()
.map(|&i| packs[i].name().to_string())
.collect();
cycle.push(packs[idx].name().to_string());
return Some(cycle);
}
if !visited.insert(idx) {
return None;
}
visiting.push(idx);
for &req in packs[idx].requires() {
let Some(&dep_idx) = name_to_idx.get(req) else {
continue;
};
if cycle_nodes.contains(&dep_idx) {
if let Some(cycle) =
visit(dep_idx, packs, name_to_idx, cycle_nodes, visiting, visited)
{
return Some(cycle);
}
}
}
visiting.pop();
None
}
let mut visited = HashSet::new();
for &idx in cycle_nodes {
let mut visiting = Vec::new();
if let Some(cycle) = visit(
idx,
packs,
name_to_idx,
cycle_nodes,
&mut visiting,
&mut visited,
) {
return cycle;
}
}
cycle_nodes
.iter()
.map(|&idx| packs[idx].name().to_string())
.collect()
}
impl Default for VerbRegistryBuilder {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone)]
pub struct VerbRegistry {
packs: std::sync::Arc<Vec<Box<dyn PackRuntime>>>,
resolvers: std::sync::Arc<Vec<(String, Box<dyn PackByIdResolver>)>>,
gate: GateRef,
default_namespace: String,
visible_namespaces: Vec<Namespace>,
actor_id: Option<String>,
event_store: Option<Arc<dyn EventStore>>,
dispatch_hook: Option<Arc<dyn DispatchHook>>,
available_verbs: Arc<Vec<&'static str>>,
reference_ring: Arc<crate::reference_ring::ReferenceRing>,
}
#[derive(Debug, Clone, Default)]
pub struct RequestIdentity {
pub namespace: String,
pub actor_id: Option<String>,
pub visible_namespaces: Vec<String>,
}
#[derive(Debug)]
pub struct PackSchemaCollisionError {
pub pack_a: &'static str,
pub pack_b: &'static str,
pub table: String,
}
impl std::fmt::Display for PackSchemaCollisionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.pack_a == self.pack_b {
write!(
f,
"pack schema boot failure for pack {:?}: {}",
self.pack_a, self.table
)
} else {
write!(
f,
"pack schema collision: packs {:?} and {:?} both declare table {:?} \
on the same backend — move one pack to a separate backend or rename the table",
self.pack_a, self.pack_b, self.table
)
}
}
}
impl std::error::Error for PackSchemaCollisionError {}
fn extract_table_names(stmt: &str) -> Vec<String> {
let normalized = stmt.split_whitespace().collect::<Vec<_>>().join(" ");
let upper = normalized.to_ascii_uppercase();
let table_name = if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE IF NOT EXISTS ") {
rest.split_whitespace().next()
} else if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE ") {
rest.split_whitespace().next()
} else if let Some(rest) = upper.strip_prefix("CREATE TABLE IF NOT EXISTS ") {
rest.split_whitespace().next()
} else if let Some(rest) = upper.strip_prefix("CREATE TABLE ") {
rest.split_whitespace().next()
} else {
None
};
match table_name {
Some(name) => {
let clean = name.trim_matches(|c: char| c == '(' || c == ';');
if clean.is_empty() {
vec![]
} else {
vec![clean.to_ascii_lowercase()]
}
}
None => vec![],
}
}
impl VerbRegistry {
pub fn default_namespace(&self) -> &str {
&self.default_namespace
}
pub fn actor_id(&self) -> Option<&str> {
self.actor_id.as_deref()
}
pub fn visible_namespaces(&self) -> &[Namespace] {
&self.visible_namespaces
}
pub fn event_store(&self) -> Option<Arc<dyn EventStore>> {
self.event_store.clone()
}
pub fn describe_verb(&self, verb: &str) -> Result<Value, RuntimeError> {
for pack in self.packs.iter() {
for handler in pack.handlers().iter() {
if handler.name == verb {
let category = format!("{:?}", handler.category);
let params_arr: Vec<Value> = handler
.params
.iter()
.map(|p| {
serde_json::json!({
"name": p.name,
"type": p.param_type,
"required": p.required,
"description": p.description,
})
})
.collect();
if matches!(handler.visibility, Visibility::Subhandler) {
return Ok(serde_json::json!({
"verb": verb,
"pack": pack.name(),
"description": handler.description,
"category": category,
"params": params_arr,
"visibility": "internal",
"callable_via_mcp": false,
"note": "This is an internal subhandler. Calling it via the MCP \
request surface returns permission denied. It can only be \
invoked by internal runtime callers.",
}));
}
return Ok(serde_json::json!({
"verb": verb,
"pack": pack.name(),
"description": handler.description,
"category": category,
"params": params_arr,
}));
}
}
}
Err(RuntimeError::InvalidInput(format!(
"unknown verb {verb:?}; available: {}",
self.available_verbs.join(", ")
)))
}
pub fn authorize_namespace(&self, ns: Namespace) -> Result<(), RuntimeError> {
let actor = crate::actor_identity::resolve_actor(self.actor_id.as_deref());
let req = GateRequest::new(actor, ns, "authorize", serde_json::Value::Null);
match self.gate.check(&req) {
Ok(decision) if decision.is_allow() => Ok(()),
Ok(GateDecision::Deny { reason }) => Err(RuntimeError::PermissionDenied {
verb: "authorize".to_string(),
reason,
}),
Ok(_) => Err(RuntimeError::PermissionDenied {
verb: "authorize".to_string(),
reason: "gate denied".to_string(),
}),
Err(e) => Err(RuntimeError::Internal(format!("gate error: {e}"))),
}
}
pub async fn dispatch(&self, verb: &str, params: Value) -> Result<Value, RuntimeError> {
self.dispatch_with_identity(verb, params, None).await
}
pub async fn dispatch_with_identity(
&self,
verb: &str,
params: Value,
identity: Option<RequestIdentity>,
) -> Result<Value, RuntimeError> {
if params.get("help").and_then(Value::as_bool) == Some(true) {
return self.describe_verb(verb);
}
let explicit_namespace = params.get("namespace").is_some_and(Value::is_string);
let default_namespace_str: &str = identity
.as_ref()
.map(|id| id.namespace.as_str())
.unwrap_or(self.default_namespace.as_str());
let ns = resolve_explicit_namespace(¶ms, default_namespace_str)?;
let actor_id_str: Option<&str> = match identity.as_ref() {
Some(id) => id.actor_id.as_deref(),
None => self.actor_id.as_deref(),
};
let resolved_actor = crate::actor_identity::resolve_actor(actor_id_str);
let gate_req = GateRequest::new(resolved_actor.clone(), ns.clone(), verb, params.clone());
let (gate_blocked, mut deferred_audit) = match self.gate.check(&gate_req) {
Ok(decision) => {
let is_deny = matches!(decision, GateDecision::Deny { .. });
let audit = AuditEvent::from_check(&gate_req, &decision, self.gate.impl_name());
tracing::info!(
audit_event = %serde_json::to_string(&audit)
.unwrap_or_else(|_| "{\"error\":\"serialize\"}".into()),
"gate.check"
);
if let Some(store) = &self.event_store {
if crate::config_ledger::PENDING
.swap(false, std::sync::atomic::Ordering::AcqRel)
{
for (key, value) in crate::config_ledger::drain_config_locked() {
let payload = serde_json::json!({ "key": key, "value": value });
let storage_event = Event::new(
gate_req.namespace.as_str(),
verb,
EventKind::ConfigLocked,
SubstrateKind::Event,
format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
)
.with_payload(payload);
append_audit_event_best_effort(store, storage_event, verb).await;
}
}
}
let defer_audit = !is_deny;
if !defer_audit {
if let Some(store) = &self.event_store {
let storage_event =
build_audit_storage_event(&gate_req, &audit, EventOutcome::Denied);
append_audit_event_best_effort(store, storage_event, verb).await;
}
}
let reason = if is_deny {
let reason = match decision {
GateDecision::Deny { reason } => reason,
_ => String::new(),
};
Some(reason)
} else {
None
};
let deferred = if defer_audit { Some(audit) } else { None };
(reason, deferred)
}
Err(err) => {
tracing::warn!(verb, error = %err, "gate check failed (fail-open)");
(None, None)
}
};
if let Some(reason) = gate_blocked {
return Err(RuntimeError::PermissionDenied {
verb: verb.to_string(),
reason,
});
}
let token = if explicit_namespace {
NamespaceToken::mint_with_visibility(ns.clone(), vec![], resolved_actor)
} else {
let primary = Namespace::local();
let mut extra_visible: Vec<Namespace> = match identity.as_ref() {
Some(id) => id
.visible_namespaces
.iter()
.filter_map(|s| match Namespace::parse(s) {
Ok(parsed) => Some(parsed),
Err(e) => {
tracing::warn!(
namespace = %s,
error = %e,
"dispatch_with_identity: skipping invalid visible_namespace \
entry from per-request identity"
);
None
}
})
.collect(),
None => self.visible_namespaces.clone(),
};
extra_visible.push(Namespace::local()); NamespaceToken::mint_with_visibility(primary, extra_visible, resolved_actor)
};
for pack in self.packs.iter() {
if let Some(handler_def) = pack.handlers().iter().find(|v| v.name == verb) {
let handler_accepts_namespace =
handler_def.params.iter().any(|p| p.name == "namespace");
let params = if !handler_accepts_namespace {
if let Value::Object(mut map) = params {
map.remove("namespace");
Value::Object(map)
} else {
params
}
} else {
params
};
let dispatch_start = Instant::now();
let result = pack.dispatch(verb, params, self, &token).await;
let dispatch_us = dispatch_start.elapsed().as_micros() as i64;
if let Some(audit) = deferred_audit.take() {
if let Some(store) = &self.event_store {
let is_link_singleton =
verb == "link" && gate_req.args.get("links").is_none();
match &result {
Ok(ok_val) if is_link_singleton => {
match link_audit_success_from_result(audit.clone(), ok_val) {
Some((edge_id, payload)) => {
let storage_event = Event::new(
gate_req.namespace.as_str(),
gate_req.verb.as_str(),
EventKind::Audit,
SubstrateKind::Event,
format!(
"{}:{}",
gate_req.actor.kind, gate_req.actor.id
),
)
.with_outcome(EventOutcome::Success)
.with_target(edge_id)
.with_payload(payload)
.with_payload_schema_version(2)
.with_duration_us(dispatch_us);
append_audit_event_best_effort(store, storage_event, verb)
.await;
}
None => {
tracing::warn!(
verb,
"link audit v2 enrichment parse failed; \
falling back to v1 audit shape"
);
let storage_event = build_audit_storage_event(
&gate_req,
&audit,
EventOutcome::Success,
)
.with_duration_us(dispatch_us);
append_audit_event_best_effort(store, storage_event, verb)
.await;
}
}
}
_ => {
let outcome = if result.is_ok() {
EventOutcome::Success
} else {
EventOutcome::Error
};
let storage_event =
build_audit_storage_event(&gate_req, &audit, outcome)
.with_duration_us(dispatch_us);
append_audit_event_best_effort(store, storage_event, verb).await;
}
}
}
}
if let (Ok(ref ok_val), Some(hook)) = (&result, &self.dispatch_hook) {
let mut dispatch_event = Event::new(
ns.as_str(),
verb,
EventKind::Audit,
SubstrateKind::Event,
pack.name(),
)
.with_outcome(EventOutcome::Success)
.with_duration_us(dispatch_us);
if verb == "memory.recall" {
let first_note_id = ok_val
.as_array()
.and_then(|arr| arr.first())
.and_then(|v| v.get("id"))
.and_then(|v| v.as_str())
.and_then(|s| s.parse::<uuid::Uuid>().ok());
if let Some(note_id) = first_note_id {
dispatch_event = dispatch_event.with_target(note_id);
}
}
let dispatch_view = EventView {
event: dispatch_event,
observations: Vec::new(),
};
let hook = Arc::clone(hook);
hook.on_dispatch(&dispatch_view).await;
}
if let Ok(ref ok_val) = result {
let admissions = crate::reference_ring::ring_admissions_for(verb, ok_val);
if !admissions.is_empty() {
let actor_key = format!("{}:{}", gate_req.actor.kind, gate_req.actor.id);
for (id, name) in admissions {
self.reference_ring.admit(
token.namespace().as_str(),
&actor_key,
id,
name,
);
}
}
}
return result;
}
}
if let Some(audit) = deferred_audit.take() {
if let Some(store) = &self.event_store {
let storage_event =
build_audit_storage_event(&gate_req, &audit, EventOutcome::Error);
append_audit_event_best_effort(store, storage_event, verb).await;
}
}
Err(RuntimeError::InvalidInput(format!(
"unknown verb {verb:?}; available: {}",
self.available_verbs.join(", ")
)))
}
pub fn resolvers(&self) -> &[(String, Box<dyn PackByIdResolver>)] {
&self.resolvers
}
pub fn reference_ring(&self) -> &Arc<crate::reference_ring::ReferenceRing> {
&self.reference_ring
}
pub fn find_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
for pack in self.packs.iter() {
let owns = pack.note_kinds().contains(&kind) || pack.entity_kinds().contains(&kind);
if owns {
if let Some(hook) = pack.kind_hook(kind) {
return Some(hook);
}
}
}
None
}
pub fn all_verbs(&self) -> Vec<&'static HandlerDef> {
self.packs
.iter()
.flat_map(|p| p.handlers().iter())
.filter(|h| matches!(h.visibility, Visibility::Verb))
.collect()
}
pub fn all_verbs_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
self.packs
.iter()
.flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
.filter(|(_, h)| matches!(h.visibility, Visibility::Verb))
.collect()
}
pub fn all_handlers_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
self.packs
.iter()
.flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
.collect()
}
pub fn all_note_kinds(&self) -> Vec<&'static str> {
let mut seen = std::collections::HashSet::new();
self.packs
.iter()
.flat_map(|p| p.note_kinds().iter().copied())
.filter(|k| seen.insert(*k))
.collect()
}
pub fn all_entity_kinds(&self) -> Vec<&'static str> {
let mut seen = std::collections::HashSet::new();
self.packs
.iter()
.flat_map(|p| p.entity_kinds().iter().copied())
.filter(|k| seen.insert(*k))
.collect()
}
pub fn pack_names(&self) -> Vec<&str> {
self.packs.iter().map(|p| p.name()).collect()
}
pub fn pack_requires(&self, name: &str) -> Option<&'static [&'static str]> {
self.packs
.iter()
.find(|p| p.name() == name)
.map(|p| p.requires())
}
pub fn pack_note_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
self.packs
.iter()
.find(|p| p.name() == name)
.map(|p| p.note_kinds())
}
pub fn pack_entity_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
self.packs
.iter()
.find(|p| p.name() == name)
.map(|p| p.entity_kinds())
}
pub fn pack_verbs(&self, name: &str) -> Option<&'static [HandlerDef]> {
self.packs
.iter()
.find(|p| p.name() == name)
.map(|p| p.handlers())
}
pub fn all_edge_rules(&self) -> Vec<EdgeEndpointRule> {
self.packs
.iter()
.flat_map(|p| p.edge_rules().iter().copied())
.collect()
}
pub fn all_note_kind_specs(&self) -> Vec<&'static NoteKindSpec> {
self.packs
.iter()
.flat_map(|p| p.note_kind_specs().iter())
.collect()
}
pub fn all_validation_rules(&self) -> Vec<&'static ValidationRule> {
self.packs
.iter()
.flat_map(|p| p.validation_rules().iter())
.collect()
}
pub fn all_schema_plans(&self) -> Vec<SchemaPlan> {
self.packs.iter().map(|p| p.schema_plan()).collect()
}
pub fn call_register_embedders(&self, runtime: &KhiveRuntime) {
for pack in self.packs.iter() {
pack.register_embedders(runtime);
}
}
pub fn call_register_entity_type_validators(&self, runtime: &KhiveRuntime) {
for pack in self.packs.iter() {
pack.register_entity_type_validator(runtime);
}
}
pub fn call_register_note_mutation_hooks(&self, runtime: &KhiveRuntime) {
for pack in self.packs.iter() {
pack.register_note_mutation_hook(runtime);
}
}
pub async fn call_warm_all(&self) {
for pack in self.packs.iter() {
pack.warm().await;
}
}
pub fn presentation_policy_for(&self, verb: &str) -> khive_types::VerbPresentationPolicy {
for pack in self.packs.iter() {
if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
return handler.presentation_policy();
}
}
khive_types::VerbPresentationPolicy::Standard
}
pub fn is_subhandler_verb(&self, verb: &str) -> bool {
for pack in self.packs.iter() {
if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
return matches!(handler.visibility, Visibility::Subhandler);
}
}
false
}
pub fn apply_schema_plans(&self, backend: &khive_db::StorageBackend) {
for plan in self.all_schema_plans() {
if plan.is_empty() {
continue;
}
if let Err(e) = backend.apply_pack_ddl_statements(plan.statements) {
tracing::warn!(
pack = plan.pack,
error = %e,
"failed to apply pack schema plan at startup (non-fatal)"
);
}
}
}
pub fn all_schema_plans_named(&self) -> Vec<(&'static str, SchemaPlan)> {
self.packs
.iter()
.map(|p| {
let plan = p.schema_plan();
(plan.pack, plan)
})
.collect()
}
pub fn apply_schema_plans_with_map(
&self,
backend_for_pack: &HashMap<&str, &khive_db::StorageBackend>,
default_backend: &khive_db::StorageBackend,
) -> Result<(), crate::PackSchemaCollisionError> {
let mut claimed: HashMap<(*const (), String), &'static str> = HashMap::new();
for (pack_name, plan) in self.all_schema_plans_named() {
if plan.is_empty() {
continue;
}
let backend = backend_for_pack
.get(pack_name)
.copied()
.unwrap_or(default_backend);
let backend_ptr = std::sync::Arc::as_ptr(&backend.pool_arc()) as *const ();
for stmt in plan.statements {
for table_name in extract_table_names(stmt) {
let key = (backend_ptr, table_name.clone());
match claimed.entry(key) {
std::collections::hash_map::Entry::Vacant(e) => {
e.insert(pack_name);
}
std::collections::hash_map::Entry::Occupied(e) => {
let prior_pack = *e.get();
return Err(crate::PackSchemaCollisionError {
pack_a: prior_pack,
pack_b: pack_name,
table: table_name,
});
}
}
}
}
backend
.apply_pack_ddl_statements(plan.statements)
.map_err(|e| crate::PackSchemaCollisionError {
pack_a: pack_name,
pack_b: pack_name,
table: format!("DDL error: {e}"),
})?;
}
Ok(())
}
}
pub struct PackInstall {
pub runtime: Box<dyn PackRuntime>,
pub resolver: Option<Box<dyn PackByIdResolver>>,
pub dispatch_hook: Option<Arc<dyn DispatchHook>>,
}
pub trait PackFactory: Send + Sync + 'static {
fn name(&self) -> &'static str;
fn requires(&self) -> &'static [&'static str] {
&[]
}
fn create(&self, runtime: KhiveRuntime) -> Box<dyn PackRuntime>;
fn create_install(&self, runtime: KhiveRuntime) -> PackInstall {
let resolver = self.create_resolver(runtime.clone());
PackInstall {
runtime: self.create(runtime),
resolver,
dispatch_hook: None,
}
}
fn create_resolver(&self, _runtime: KhiveRuntime) -> Option<Box<dyn PackByIdResolver>> {
None
}
}
pub struct PackRegistration(pub &'static dyn PackFactory);
inventory::collect!(PackRegistration);
#[derive(Debug)]
pub enum PackLoadError {
UnknownPack(String),
MissingDependency {
pack: String,
dep: String,
},
}
impl std::fmt::Display for PackLoadError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PackLoadError::UnknownPack(name) => write!(f, "unknown pack {name:?}"),
PackLoadError::MissingDependency { pack, dep } => write!(
f,
"pack {pack:?} requires {dep:?}, which is not in the requested pack list; \
add --pack {dep} before --pack {pack}"
),
}
}
}
impl std::error::Error for PackLoadError {}
pub struct PackRegistry;
impl PackRegistry {
pub fn discovered_names() -> Vec<&'static str> {
inventory::iter::<PackRegistration>
.into_iter()
.map(|r| r.0.name())
.collect()
}
pub fn register_packs(
names: &[String],
runtime: KhiveRuntime,
builder: &mut VerbRegistryBuilder,
) -> Result<(), PackLoadError> {
let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
.into_iter()
.map(|r| r.0)
.collect();
let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
all.iter().copied().find(|f| f.name() == name)
};
let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
for name in names {
factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
}
for name in names {
let factory = factory_for(name.as_str()).unwrap(); for &dep in factory.requires() {
if !requested.contains(dep) {
return Err(PackLoadError::MissingDependency {
pack: name.clone(),
dep: dep.to_string(),
});
}
}
}
for name in names {
let factory = factory_for(name.as_str()).unwrap(); let install = factory.create_install(runtime.clone());
builder.register_boxed(install.runtime);
if let Some(resolver) = install.resolver {
builder.register_resolver(name.clone(), resolver);
}
if let Some(hook) = install.dispatch_hook {
builder.with_dispatch_hook(hook);
}
}
Ok(())
}
pub fn register_packs_with_runtimes(
names: &[String],
runtimes: &HashMap<String, KhiveRuntime>,
default_runtime: &KhiveRuntime,
builder: &mut VerbRegistryBuilder,
) -> Result<(), PackLoadError> {
let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
.into_iter()
.map(|r| r.0)
.collect();
let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
all.iter().copied().find(|f| f.name() == name)
};
let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
for name in names {
factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
}
for name in names {
let factory = factory_for(name.as_str()).unwrap();
for &dep in factory.requires() {
if !requested.contains(dep) {
return Err(PackLoadError::MissingDependency {
pack: name.clone(),
dep: dep.to_string(),
});
}
}
}
for name in names {
let factory = factory_for(name.as_str()).unwrap();
let runtime = runtimes
.get(name.as_str())
.cloned()
.unwrap_or_else(|| default_runtime.clone());
let install = factory.create_install(runtime);
builder.register_boxed(install.runtime);
if let Some(resolver) = install.resolver {
builder.register_resolver(name.clone(), resolver);
}
if let Some(hook) = install.dispatch_hook {
builder.with_dispatch_hook(hook);
}
}
Ok(())
}
}
fn target_id_from_args(args: &serde_json::Value) -> Option<uuid::Uuid> {
args.get("target_id")
.and_then(serde_json::Value::as_str)
.and_then(|s| s.parse::<uuid::Uuid>().ok())
}
fn build_audit_storage_event(
gate_req: &GateRequest,
audit: &AuditEvent,
outcome: EventOutcome,
) -> Event {
let audit_data = serde_json::to_value(audit).unwrap_or_else(|e| {
tracing::warn!(error = %e, "failed to serialize AuditEvent for EventStore");
serde_json::Value::Null
});
let mut storage_event = Event::new(
gate_req.namespace.as_str(),
gate_req.verb.as_str(),
EventKind::Audit,
SubstrateKind::Event,
format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
)
.with_outcome(outcome)
.with_payload(audit_data);
if let Some(target_id) = target_id_from_args(&gate_req.args) {
storage_event = storage_event.with_target(target_id);
}
storage_event
}
async fn append_audit_event_best_effort(store: &Arc<dyn EventStore>, event: Event, verb: &str) {
if let Err(store_err) = store.append_event(event).await {
tracing::warn!(
verb,
error = %store_err,
"audit event store write failed (non-fatal)"
);
}
}
#[derive(Debug, Clone, serde::Serialize)]
struct LinkAuditSuccessV2 {
#[serde(flatten)]
audit: AuditEvent,
edge_id: uuid::Uuid,
source_id: uuid::Uuid,
target_id: uuid::Uuid,
relation: String,
weight: f64,
}
fn link_audit_success_from_result(
audit: AuditEvent,
result: &serde_json::Value,
) -> Option<(uuid::Uuid, serde_json::Value)> {
let edge_id = result.get("id")?.as_str()?.parse::<uuid::Uuid>().ok()?;
let source_id = result
.get("source_id")?
.as_str()?
.parse::<uuid::Uuid>()
.ok()?;
let target_id = result
.get("target_id")?
.as_str()?
.parse::<uuid::Uuid>()
.ok()?;
let relation = result.get("relation")?.as_str()?.to_string();
let weight = result.get("weight")?.as_f64()?;
let enriched = LinkAuditSuccessV2 {
audit,
edge_id,
source_id,
target_id,
relation,
weight,
};
let payload = serde_json::to_value(&enriched).ok()?;
Some((edge_id, payload))
}
pub fn resolve_explicit_namespace(
params: &Value,
default_namespace: &str,
) -> Result<Namespace, RuntimeError> {
match params.get("namespace") {
None => Namespace::parse(default_namespace)
.map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}"))),
Some(Value::String(ns_str)) => Namespace::parse(ns_str)
.map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace {ns_str:?}: {e}"))),
Some(other) => Err(RuntimeError::InvalidInput(format!(
"invalid namespace: expected string when present, got {}",
json_type_name(other),
))),
}
}
pub fn json_type_name(v: &Value) -> &'static str {
match v {
Value::Null => "null",
Value::Bool(_) => "boolean",
Value::Number(_) => "number",
Value::String(_) => "string",
Value::Array(_) => "array",
Value::Object(_) => "object",
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ActorRef;
use khive_types::Pack;
struct AlphaPack;
impl Pack for AlphaPack {
const NAME: &'static str = "alpha";
const NOTE_KINDS: &'static [&'static str] = &["memo", "log"];
const ENTITY_KINDS: &'static [&'static str] = &["widget"];
const HANDLERS: &'static [HandlerDef] = &[
HandlerDef {
name: "create",
description: "create a widget",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &[],
},
HandlerDef {
name: "list",
description: "list widgets",
visibility: Visibility::Verb,
category: VerbCategory::Assertive,
params: &[],
},
];
}
#[async_trait]
impl PackRuntime for AlphaPack {
fn name(&self) -> &str {
AlphaPack::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
AlphaPack::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
AlphaPack::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
AlphaPack::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
}
}
struct SleepingPack;
impl Pack for SleepingPack {
const NAME: &'static str = "sleeping";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "slow_op",
description: "sleeps before returning",
visibility: Visibility::Verb,
category: VerbCategory::Assertive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for SleepingPack {
fn name(&self) -> &str {
SleepingPack::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
SleepingPack::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
SleepingPack::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
SleepingPack::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
Ok(serde_json::json!({ "pack": "sleeping", "verb": verb }))
}
}
struct BetaPack;
impl Pack for BetaPack {
const NAME: &'static str = "beta";
const NOTE_KINDS: &'static [&'static str] = &["alert"];
const ENTITY_KINDS: &'static [&'static str] = &["widget", "gadget"];
const HANDLERS: &'static [HandlerDef] = &[
HandlerDef {
name: "notify",
description: "send alert",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &[],
},
HandlerDef {
name: "create",
description: "beta internal create (subhandler)",
visibility: Visibility::Subhandler,
category: VerbCategory::Commissive,
params: &[],
},
];
}
fn build_registry() -> VerbRegistry {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.register(BetaPack);
builder.build().expect("registry builds without collision")
}
struct CollidingPack;
impl Pack for CollidingPack {
const NAME: &'static str = "colliding";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "create",
description: "duplicate Verb-visibility create",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for CollidingPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({ "pack": "colliding", "verb": verb }))
}
}
#[async_trait]
impl PackRuntime for BetaPack {
fn name(&self) -> &str {
BetaPack::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
BetaPack::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
BetaPack::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
BetaPack::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({ "pack": "beta", "verb": verb }))
}
}
#[tokio::test]
async fn dispatch_routes_to_correct_pack() {
let reg = build_registry();
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(res["pack"], "alpha");
let res = reg.dispatch("notify", Value::Null).await.unwrap();
assert_eq!(res["pack"], "beta");
}
#[test]
fn verb_collision_is_boot_time_error() {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.register(CollidingPack);
let err = builder
.build()
.err()
.expect("duplicate Verb-visibility handler must be rejected at build time");
assert!(
matches!(err, RuntimeError::VerbCollision { ref verb, .. } if verb == "create"),
"expected VerbCollision for 'create', got {err:?}"
);
let msg = err.to_string();
assert!(
msg.contains("create"),
"error must name the colliding verb: {msg}"
);
assert!(
msg.contains("alpha") || msg.contains("colliding"),
"error must name one of the conflicting packs: {msg}"
);
}
#[test]
fn subhandler_same_name_across_packs_is_not_a_collision() {
struct SubhandlerPack;
impl Pack for SubhandlerPack {
const NAME: &'static str = "subhandler_pack";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "create",
description: "internal create",
visibility: Visibility::Subhandler,
category: VerbCategory::Commissive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for SubhandlerPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({"pack": "subhandler_pack", "verb": verb}))
}
}
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack); builder.register(SubhandlerPack); builder
.build()
.expect("subhandler same name must NOT be a collision");
}
#[tokio::test]
async fn dispatch_unknown_verb_returns_error() {
let reg = build_registry();
let err = reg.dispatch("explode", Value::Null).await.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("explode"));
assert!(msg.contains("create"));
}
#[test]
fn all_verbs_aggregates_across_packs_excludes_subhandlers() {
let reg = build_registry();
let verbs: Vec<&str> = reg.all_verbs().iter().map(|v| v.name).collect();
assert_eq!(verbs, vec!["create", "list", "notify"]);
}
#[test]
fn all_verbs_with_names_pairs_pack_name_excludes_subhandlers() {
let reg = build_registry();
let pairs: Vec<(&str, &str)> = reg
.all_verbs_with_names()
.iter()
.map(|(pack, v)| (*pack, v.name))
.collect();
assert_eq!(
pairs,
vec![("alpha", "create"), ("alpha", "list"), ("beta", "notify"),]
);
}
#[test]
fn all_handlers_with_names_includes_subhandlers() {
let reg = build_registry();
let pairs: Vec<(&str, &str)> = reg
.all_handlers_with_names()
.iter()
.map(|(pack, v)| (*pack, v.name))
.collect();
assert_eq!(
pairs,
vec![
("alpha", "create"),
("alpha", "list"),
("beta", "notify"),
("beta", "create"),
]
);
}
#[test]
fn note_kinds_are_ordered() {
let reg = build_registry();
let kinds = reg.all_note_kinds();
assert_eq!(kinds, vec!["memo", "log", "alert"]);
}
#[test]
fn note_kind_duplicate_rejected_at_build_time() {
struct DupPack;
impl khive_types::Pack for DupPack {
const NAME: &'static str = "dup";
const NOTE_KINDS: &'static [&'static str] = &["memo"];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
}
#[async_trait]
impl PackRuntime for DupPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
_verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(Value::Null)
}
}
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.register(DupPack);
let err = builder
.build()
.err()
.expect("duplicate note kind must be rejected");
let msg = err.to_string();
assert!(
msg.contains("memo"),
"error must name the duplicate kind: {msg}"
);
assert!(
msg.contains("alpha") || msg.contains("dup"),
"error must name one of the conflicting packs: {msg}"
);
}
#[test]
fn entity_kinds_are_deduplicated() {
let reg = build_registry();
let kinds = reg.all_entity_kinds();
assert_eq!(kinds, vec!["widget", "gadget"]);
}
use khive_gate::{Gate, GateError};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
#[derive(Default, Debug)]
struct CountingGate {
calls: AtomicUsize,
deny_verb: Option<&'static str>,
}
impl Gate for CountingGate {
fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
self.calls.fetch_add(1, Ordering::SeqCst);
if Some(req.verb.as_str()) == self.deny_verb {
Ok(GateDecision::deny(format!("test deny for {}", req.verb)))
} else {
Ok(GateDecision::allow())
}
}
}
#[tokio::test]
async fn dispatch_consults_the_gate() {
let gate = Arc::new(CountingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
reg.dispatch("create", Value::Null).await.unwrap();
assert_eq!(
gate.calls.load(Ordering::SeqCst),
2,
"gate should be consulted once per dispatch"
);
}
#[tokio::test]
async fn dispatch_returns_permission_denied_on_deny_v03() {
let gate = Arc::new(CountingGate {
calls: AtomicUsize::new(0),
deny_verb: Some("create"),
});
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
let err = reg.dispatch("create", Value::Null).await.unwrap_err();
assert!(
matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
"expected PermissionDenied, got {err:?}"
);
let msg = err.to_string();
assert!(
msg.contains("create"),
"error message must name the verb: {msg}"
);
assert!(
msg.contains("test deny for create"),
"error message must carry the deny reason: {msg}"
);
assert_eq!(gate.calls.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn dispatch_allow_verb_succeeds_even_with_deny_gate_for_other_verb() {
let gate = Arc::new(CountingGate {
calls: AtomicUsize::new(0),
deny_verb: Some("create"),
});
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(res["pack"], "alpha");
}
#[tokio::test]
async fn dispatch_uses_allow_all_gate_by_default() {
let reg = build_registry();
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(res["pack"], "alpha");
}
#[derive(Default, Debug)]
struct NamespaceCapturingGate {
seen: std::sync::Mutex<Vec<String>>,
}
impl Gate for NamespaceCapturingGate {
fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
self.seen
.lock()
.unwrap()
.push(req.namespace.as_str().to_string());
Ok(GateDecision::allow())
}
}
#[tokio::test]
async fn dispatch_propagates_params_namespace_to_gate() {
let gate = Arc::new(NamespaceCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
builder.with_default_namespace("tenant-x");
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "tenant-y"}))
.await
.unwrap();
reg.dispatch("list", Value::Null).await.unwrap();
let err = reg
.dispatch("list", serde_json::json!({"namespace": ""}))
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::InvalidInput(_)),
"empty namespace must return InvalidInput, got {err:?}"
);
let seen = gate.seen.lock().unwrap().clone();
assert_eq!(seen, vec!["tenant-y", "tenant-x"]);
}
#[tokio::test]
async fn dispatch_falls_back_to_local_when_no_default_set() {
let gate = Arc::new(NamespaceCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let seen = gate.seen.lock().unwrap().clone();
assert_eq!(seen, vec!["local"]);
}
#[tokio::test]
async fn namespace_null_rejected_not_coerced() {
let cases: Vec<(&str, Value)> = vec![
("null", Value::Null),
("number", serde_json::json!(42)),
("boolean", serde_json::json!(true)),
("array", serde_json::json!(["local"])),
("object", serde_json::json!({"ns": "local"})),
];
for (label, ns_value) in cases {
let gate = Arc::new(NamespaceCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
builder.with_default_namespace("tenant-x");
let reg = builder.build().expect("registry builds");
let err = reg
.dispatch("list", serde_json::json!({"namespace": ns_value}))
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::InvalidInput(_)),
"case {label}: expected InvalidInput, got {err:?}"
);
let seen = gate.seen.lock().unwrap().clone();
assert!(
seen.is_empty(),
"case {label}: gate must not be consulted for malformed namespace, saw {seen:?}"
);
}
}
use khive_gate::{AuditDecision, AuditEvent, Obligation};
#[derive(Default, Debug)]
struct AuditCapturingGate {
events: std::sync::Mutex<Vec<AuditEvent>>,
deny_verb: Option<&'static str>,
}
impl Gate for AuditCapturingGate {
fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
let decision = if Some(req.verb.as_str()) == self.deny_verb {
GateDecision::deny("test deny")
} else {
GateDecision::allow_with(vec![Obligation::Audit {
tag: format!("{}.check", req.verb),
}])
};
let ev = AuditEvent::from_check(req, &decision, self.impl_name());
self.events.lock().unwrap().push(ev);
Ok(decision)
}
fn impl_name(&self) -> &'static str {
"AuditCapturingGate"
}
}
#[tokio::test]
async fn dispatch_emits_one_audit_event_per_call() {
let gate = Arc::new(AuditCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
reg.dispatch("create", Value::Null).await.unwrap();
let evs = gate.events.lock().unwrap();
assert_eq!(evs.len(), 2, "exactly one audit event per dispatch call");
}
#[tokio::test]
async fn dispatch_audit_event_allow_carries_obligations() {
let gate = Arc::new(AuditCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let evs = gate.events.lock().unwrap();
let ev = &evs[0];
assert_eq!(ev.verb, "list");
assert_eq!(ev.decision, AuditDecision::Allow);
assert!(ev.deny_reason.is_none());
assert_eq!(ev.obligations.len(), 1);
assert_eq!(ev.gate_impl, "AuditCapturingGate");
}
#[tokio::test]
async fn dispatch_audit_event_deny_carries_reason() {
let gate = Arc::new(AuditCapturingGate {
events: Default::default(),
deny_verb: Some("create"),
});
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
let err = reg.dispatch("create", Value::Null).await.unwrap_err();
assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
let evs = gate.events.lock().unwrap();
let ev = &evs[0];
assert_eq!(ev.verb, "create");
assert_eq!(ev.decision, AuditDecision::Deny);
assert_eq!(ev.deny_reason.as_deref(), Some("test deny"));
assert!(ev.obligations.is_empty());
}
#[tokio::test]
async fn dispatch_audit_event_fields_match_gate_request() {
let gate = Arc::new(AuditCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
builder.with_default_namespace("tenant-z");
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
.await
.unwrap();
let evs = gate.events.lock().unwrap();
let ev = &evs[0];
assert_eq!(ev.namespace, "tenant-q");
assert_eq!(ev.verb, "list");
assert_eq!(ev.actor.kind, "anonymous");
}
#[derive(Default, Debug)]
struct ActorCapturingGate {
requests: std::sync::Mutex<Vec<GateRequest>>,
}
impl Gate for ActorCapturingGate {
fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
self.requests.lock().unwrap().push(req.clone());
Ok(GateDecision::allow())
}
}
#[tokio::test]
async fn gate_request_carries_configured_actor_when_actor_id_is_set() {
let gate = Arc::new(ActorCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
builder.with_actor_id(Some("team-abc:implementer".to_string()));
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let reqs = gate.requests.lock().unwrap();
assert_eq!(reqs.len(), 1);
let req = &reqs[0];
assert_eq!(
req.actor.kind, "actor",
"gate request must carry kind='actor' when actor_id is configured"
);
assert_eq!(
req.actor.id, "team-abc:implementer",
"gate request must carry the configured actor id"
);
}
#[tokio::test]
async fn gate_request_carries_anonymous_when_no_actor_id_configured() {
let gate = Arc::new(ActorCapturingGate::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let reqs = gate.requests.lock().unwrap();
assert_eq!(reqs.len(), 1);
let req = &reqs[0];
assert_eq!(
req.actor.kind, "anonymous",
"gate request must carry anonymous actor when no actor_id is configured"
);
assert_eq!(req.actor.id, "local");
}
struct TokenCapturingPack {
actors: Arc<std::sync::Mutex<Vec<khive_gate::ActorRef>>>,
}
impl Pack for TokenCapturingPack {
const NAME: &'static str = "alpha";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = AlphaPack::HANDLERS;
}
#[async_trait]
impl PackRuntime for TokenCapturingPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
self.actors.lock().unwrap().push(token.actor().clone());
Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
}
}
#[tokio::test]
async fn gate_actor_and_token_actor_are_identical_when_actor_id_is_set() {
let gate = Arc::new(ActorCapturingGate::default());
let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
let pack = TokenCapturingPack {
actors: actors.clone(),
};
let mut builder = VerbRegistryBuilder::new();
builder.register(pack);
builder.with_gate(gate.clone());
builder.with_actor_id(Some("actor-alpha".to_string()));
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let reqs = gate.requests.lock().unwrap();
let gate_actor = reqs[0].actor.clone();
drop(reqs);
let captured = actors.lock().unwrap();
let token_actor = captured[0].clone();
assert_eq!(
gate_actor.kind, token_actor.kind,
"gate request actor and storage token actor must carry the same kind"
);
assert_eq!(
gate_actor.id, token_actor.id,
"gate request actor and storage token actor must carry the same id"
);
assert_eq!(gate_actor.id, "actor-alpha");
}
#[tokio::test]
async fn gate_actor_and_token_actor_are_identical_when_anonymous() {
let gate = Arc::new(ActorCapturingGate::default());
let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
let pack = TokenCapturingPack {
actors: actors.clone(),
};
let mut builder = VerbRegistryBuilder::new();
builder.register(pack);
builder.with_gate(gate.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", Value::Null).await.unwrap();
let reqs = gate.requests.lock().unwrap();
let gate_actor = reqs[0].actor.clone();
drop(reqs);
let captured = actors.lock().unwrap();
let token_actor = captured[0].clone();
assert_eq!(gate_actor.kind, token_actor.kind);
assert_eq!(gate_actor.id, token_actor.id);
assert_eq!(gate_actor.id, "local");
}
#[tokio::test]
async fn rego_gate_missing_entrypoint_returns_permission_denied() {
use khive_gate_rego::RegoGate;
let policy = r#"
package khive.gate
import rego.v1
verdict := "allow"
"#;
let gate = Arc::new(RegoGate::from_policy_str(policy).expect("policy compiles"));
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(gate);
let reg = builder.build().expect("registry builds");
let err = reg.dispatch("create", Value::Null).await.unwrap_err();
assert!(
matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
"expected PermissionDenied for missing rego entrypoint, got {err:?}"
);
}
use std::sync::{Mutex as StdMutex, Once, OnceLock};
use serial_test::serial;
use tracing::field::{Field, Visit};
#[derive(Clone, Debug, Default)]
struct CapturedEvent {
message: Option<String>,
audit_event: Option<String>,
}
#[derive(Default)]
struct CapturedEventVisitor(CapturedEvent);
impl Visit for CapturedEventVisitor {
fn record_str(&mut self, field: &Field, value: &str) {
match field.name() {
"message" => self.0.message = Some(value.to_string()),
"audit_event" => self.0.audit_event = Some(value.to_string()),
_ => {}
}
}
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
let formatted = format!("{value:?}");
let cleaned = formatted
.trim_start_matches('"')
.trim_end_matches('"')
.to_string();
match field.name() {
"message" => self.0.message = Some(cleaned),
"audit_event" => self.0.audit_event = Some(cleaned),
_ => {}
}
}
}
struct CaptureSubscriber {
events: Arc<StdMutex<Vec<CapturedEvent>>>,
}
impl CaptureSubscriber {
fn new(events: Arc<StdMutex<Vec<CapturedEvent>>>) -> Self {
Self { events }
}
}
impl tracing::Subscriber for CaptureSubscriber {
fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
true
}
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
tracing::span::Id::from_u64(1)
}
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
fn event(&self, event: &tracing::Event<'_>) {
let mut visitor = CapturedEventVisitor::default();
event.record(&mut visitor);
self.events.lock().unwrap().push(visitor.0);
}
fn enter(&self, _: &tracing::span::Id) {}
fn exit(&self, _: &tracing::span::Id) {}
}
static GLOBAL_CAPTURE: OnceLock<Arc<StdMutex<Vec<CapturedEvent>>>> = OnceLock::new();
static GLOBAL_INIT: Once = Once::new();
fn global_capture() -> Arc<StdMutex<Vec<CapturedEvent>>> {
GLOBAL_INIT.call_once(|| {
let buffer = Arc::new(StdMutex::new(Vec::new()));
let subscriber = CaptureSubscriber::new(Arc::clone(&buffer));
let _ = tracing::subscriber::set_global_default(subscriber);
let _ = GLOBAL_CAPTURE.set(buffer);
});
Arc::clone(GLOBAL_CAPTURE.get().expect("global capture initialized"))
}
fn capture_dispatch_events<Fut>(future: Fut) -> Vec<CapturedEvent>
where
Fut: std::future::Future<Output = ()>,
{
let buffer = global_capture();
buffer.lock().unwrap().clear();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("build current-thread tokio runtime");
rt.block_on(future);
let result = buffer.lock().unwrap().clone();
result
}
fn gate_check_events_for(events: &[CapturedEvent], gate_impl: &str) -> Vec<CapturedEvent> {
events
.iter()
.filter(|e| e.message.as_deref() == Some("gate.check"))
.filter(|e| {
e.audit_event
.as_deref()
.and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
.and_then(|v| {
v.get("gate_impl")
.and_then(|g| g.as_str().map(|s| s.to_string()))
})
.as_deref()
== Some(gate_impl)
})
.cloned()
.collect()
}
#[test]
#[serial]
fn dispatch_tracing_emits_one_gate_check_event_on_allow() {
#[derive(Debug)]
struct TracingAllowGate;
impl Gate for TracingAllowGate {
fn check(&self, _: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::allow())
}
fn impl_name(&self) -> &'static str {
"TracingAllowGate"
}
}
let events = capture_dispatch_events(async {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(TracingAllowGate));
builder.with_default_namespace("tenant-default");
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
.await
.unwrap();
});
let gate_events = gate_check_events_for(&events, "TracingAllowGate");
assert_eq!(
gate_events.len(),
1,
"exactly one gate.check tracing event per dispatch (allow); got {gate_events:?}"
);
let payload = gate_events[0]
.audit_event
.as_ref()
.expect("gate.check event must carry an audit_event field");
let audit: khive_gate::AuditEvent =
serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
assert_eq!(audit.decision, AuditDecision::Allow);
assert_eq!(audit.verb, "list");
assert_eq!(audit.namespace, "tenant-q");
assert_eq!(audit.gate_impl, "TracingAllowGate");
assert!(
audit.deny_reason.is_none(),
"deny_reason must be None on Allow"
);
}
use crate::runtime::NamespaceToken;
use async_trait::async_trait;
use khive_storage::{
BatchWriteSummary, Event, EventFilter, EventStore, Page, PageRequest, SubstrateKind,
};
use khive_types::EventOutcome;
#[derive(Default, Debug)]
struct MemoryEventStore {
events: std::sync::Mutex<Vec<Event>>,
}
#[async_trait]
impl EventStore for MemoryEventStore {
async fn append_event(&self, event: Event) -> khive_storage::StorageResult<()> {
self.events.lock().unwrap().push(event);
Ok(())
}
async fn append_events(
&self,
events: Vec<Event>,
) -> khive_storage::StorageResult<BatchWriteSummary> {
let attempted = events.len() as u64;
let affected = attempted;
self.events.lock().unwrap().extend(events);
Ok(BatchWriteSummary {
attempted,
affected,
failed: 0,
first_error: String::new(),
})
}
async fn get_event(&self, id: uuid::Uuid) -> khive_storage::StorageResult<Option<Event>> {
Ok(self
.events
.lock()
.unwrap()
.iter()
.find(|e| e.id == id)
.cloned())
}
async fn query_events(
&self,
_filter: EventFilter,
_page: PageRequest,
) -> khive_storage::StorageResult<Page<Event>> {
let items = self.events.lock().unwrap().clone();
let total = items.len() as u64;
Ok(Page {
items,
total: Some(total),
})
}
async fn count_events(&self, _filter: EventFilter) -> khive_storage::StorageResult<u64> {
Ok(self.events.lock().unwrap().len() as u64)
}
}
#[tokio::test]
async fn allow_all_gate_default_remains_backward_compatible() {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
let reg = builder.build().expect("registry builds");
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(
res["pack"], "alpha",
"AllowAllGate must allow every verb — backward compat guarantee"
);
let res = reg.dispatch("create", Value::Null).await.unwrap();
assert_eq!(res["pack"], "alpha");
}
#[tokio::test]
async fn deny_gate_returns_permission_denied_pack_never_invoked() {
#[derive(Debug)]
struct AlwaysDenyGate;
impl Gate for AlwaysDenyGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("test: always deny"))
}
}
#[derive(Debug)]
struct TrackedPack {
invoked: Arc<AtomicUsize>,
}
impl khive_types::Pack for TrackedPack {
const NAME: &'static str = "tracked";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "guarded",
description: "a guarded verb",
visibility: Visibility::Verb,
category: VerbCategory::Assertive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for TrackedPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
_verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
self.invoked.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({"invoked": true}))
}
}
let invoked = Arc::new(AtomicUsize::new(0));
let mut builder = VerbRegistryBuilder::new();
builder.register(TrackedPack {
invoked: invoked.clone(),
});
builder.with_gate(Arc::new(AlwaysDenyGate));
let reg = builder.build().expect("registry builds");
let err = reg.dispatch("guarded", Value::Null).await.unwrap_err();
assert!(
matches!(err, RuntimeError::PermissionDenied { ref verb, ref reason } if verb == "guarded" && reason.contains("always deny")),
"expected PermissionDenied with verb=guarded and reason, got: {err:?}"
);
assert_eq!(
invoked.load(Ordering::SeqCst),
0,
"pack dispatch MUST NOT be invoked when gate denies"
);
}
#[tokio::test]
async fn audit_event_persists_to_event_store_on_allow() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap();
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(count, 1, "one audit event persisted to EventStore on allow");
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
let ev = &page.items[0];
assert_eq!(ev.verb, "list");
assert_eq!(ev.namespace, "test-ns");
assert_eq!(ev.substrate, SubstrateKind::Event);
assert_eq!(ev.outcome, EventOutcome::Success);
}
#[tokio::test]
async fn audit_event_duration_us_reflects_measured_dispatch_time() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SleepingPack);
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("slow_op", serde_json::json!({}))
.await
.unwrap();
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(page.items.len(), 1);
let ev = &page.items[0];
assert!(
ev.duration_us >= 10_000,
"duration_us must reflect the ~20ms measured dispatch time, got {}",
ev.duration_us
);
}
#[tokio::test]
async fn dispatch_unknown_verb_allowed_by_gate_still_persists_audit_row() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
let result = reg.dispatch("no_such_verb", serde_json::json!({})).await;
assert!(result.is_err(), "unknown verb must still return an error");
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(
count, 1,
"an allowed-but-unknown verb must still persist one audit row"
);
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(page.items[0].duration_us, 0);
assert_eq!(page.items[0].outcome, EventOutcome::Error);
}
#[tokio::test]
async fn audit_event_persists_to_event_store_on_deny() {
#[derive(Debug)]
struct AlwaysDenyGate;
impl Gate for AlwaysDenyGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("denied by test"))
}
}
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(AlwaysDenyGate));
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
let err = reg
.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap_err();
assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(count, 1, "one audit event persisted to EventStore on deny");
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
let ev = &page.items[0];
assert_eq!(ev.verb, "list");
assert_eq!(ev.outcome, EventOutcome::Denied);
}
#[tokio::test]
async fn gate_error_does_not_persist_to_event_store() {
#[derive(Debug)]
struct FailingGate;
impl Gate for FailingGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, khive_gate::GateError> {
Err(khive_gate::GateError::Internal("gate broken".into()))
}
}
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(FailingGate));
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(
res["pack"], "alpha",
"gate error must fail-open, not block dispatch"
);
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(
count, 0,
"gate infrastructure error must NOT produce an audit event in EventStore"
);
}
#[tokio::test]
async fn no_event_store_configured_tracing_only() {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
let reg = builder.build().expect("registry builds");
let res = reg.dispatch("list", Value::Null).await.unwrap();
assert_eq!(res["pack"], "alpha");
}
#[test]
#[serial]
fn dispatch_tracing_emits_gate_check_event_with_deny_payload() {
#[derive(Debug)]
struct TracingDenyGate;
impl Gate for TracingDenyGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("denied by test gate"))
}
fn impl_name(&self) -> &'static str {
"TracingDenyGate"
}
}
let events = capture_dispatch_events(async {
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(TracingDenyGate));
let reg = builder.build().expect("registry builds");
let _ = reg.dispatch("create", serde_json::Value::Null).await;
});
let gate_events = gate_check_events_for(&events, "TracingDenyGate");
assert_eq!(
gate_events.len(),
1,
"exactly one gate.check tracing event per dispatch (deny); got {gate_events:?}"
);
let payload = gate_events[0]
.audit_event
.as_ref()
.expect("gate.check event must carry an audit_event field on Deny");
let audit: khive_gate::AuditEvent =
serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
assert_eq!(audit.decision, AuditDecision::Deny);
assert_eq!(audit.deny_reason.as_deref(), Some("denied by test gate"));
assert_eq!(audit.gate_impl, "TracingDenyGate");
let payload_json: serde_json::Value =
serde_json::from_str(payload).expect("payload must be valid JSON");
assert_eq!(
payload_json["obligations"],
serde_json::Value::Array(Vec::new()),
"obligations must be `[]` on Deny on the tracing payload, not omitted"
);
}
#[tokio::test]
async fn audit_envelope_round_trips_deny_reason_and_gate_impl_through_event_store() {
#[derive(Debug)]
struct DenyGateWithName;
impl Gate for DenyGateWithName {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("policy: write forbidden for anon"))
}
fn impl_name(&self) -> &'static str {
"DenyGateWithName"
}
}
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(DenyGateWithName));
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
let err = reg
.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::PermissionDenied { .. }),
"expected PermissionDenied, got {err:?}"
);
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(
page.items.len(),
1,
"one audit event must be persisted on deny"
);
let ev = &page.items[0];
assert_eq!(ev.outcome, EventOutcome::Denied);
let data = &ev.payload;
let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
.expect("Event.payload must deserialize to AuditEvent");
assert_eq!(
audit.deny_reason.as_deref(),
Some("policy: write forbidden for anon"),
"deny_reason must be preserved through EventStore"
);
assert_eq!(
audit.gate_impl, "DenyGateWithName",
"gate_impl must be preserved through EventStore"
);
assert_eq!(
audit.decision,
khive_gate::AuditDecision::Deny,
"decision field must be preserved through EventStore"
);
}
#[tokio::test]
async fn audit_envelope_round_trips_obligations_through_event_store() {
use khive_gate::Obligation;
#[derive(Debug)]
struct ObligationGate;
impl Gate for ObligationGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::allow_with(vec![Obligation::Audit {
tag: "billing.meter".into(),
}]))
}
fn impl_name(&self) -> &'static str {
"ObligationGate"
}
}
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(ObligationGate));
builder.with_event_store(store.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap();
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(page.items.len(), 1);
let ev = &page.items[0];
assert_eq!(ev.outcome, EventOutcome::Success);
let data = &ev.payload;
let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
.expect("Event.payload must deserialize to AuditEvent");
assert_eq!(audit.gate_impl, "ObligationGate");
assert_eq!(
audit.obligations.len(),
1,
"obligations must be preserved through EventStore"
);
match &audit.obligations[0] {
Obligation::Audit { tag } => assert_eq!(tag, "billing.meter"),
other => panic!("expected Audit obligation, got {other:?}"),
}
}
#[tokio::test]
async fn sql_backed_audit_envelope_round_trips_deny_reason_gate_impl_and_obligations() {
#[derive(Debug)]
struct SqlTestDenyGate;
impl Gate for SqlTestDenyGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("sql-path: write denied"))
}
fn impl_name(&self) -> &'static str {
"SqlTestDenyGate"
}
}
let rt = KhiveRuntime::memory().expect("in-memory runtime");
let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
let sql_store = rt
.events(&test_tok)
.expect("events_for_namespace must succeed");
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(SqlTestDenyGate));
builder.with_event_store(sql_store.clone());
let reg = builder.build().expect("registry builds");
let err = reg
.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::PermissionDenied { .. }),
"expected PermissionDenied, got {err:?}"
);
let page = sql_store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(
page.items.len(),
1,
"one audit event must be persisted on deny through SqlEventStore"
);
let ev = &page.items[0];
assert_eq!(ev.outcome, EventOutcome::Denied);
let data = &ev.payload;
let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
.expect("Event.payload must deserialize to AuditEvent after SQL round-trip");
assert_eq!(
audit.deny_reason.as_deref(),
Some("sql-path: write denied"),
"deny_reason must survive the SQL text round-trip"
);
assert_eq!(
audit.gate_impl, "SqlTestDenyGate",
"gate_impl must survive the SQL text round-trip"
);
assert_eq!(
audit.decision,
khive_gate::AuditDecision::Deny,
"decision field must survive the SQL text round-trip"
);
assert!(
audit.obligations.is_empty(),
"obligations must be preserved as empty [] through SQL round-trip"
);
}
#[tokio::test]
async fn sql_backed_audit_envelope_round_trips_non_empty_obligations() {
use khive_gate::Obligation;
#[derive(Debug)]
struct SqlTestAllowWithObligationGate;
impl Gate for SqlTestAllowWithObligationGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::allow_with(vec![Obligation::Audit {
tag: "sql-path-billing.meter".into(),
}]))
}
fn impl_name(&self) -> &'static str {
"SqlTestAllowWithObligationGate"
}
}
let rt = KhiveRuntime::memory().expect("in-memory runtime");
let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
let sql_store = rt
.events(&test_tok)
.expect("events_for_namespace must succeed");
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_gate(Arc::new(SqlTestAllowWithObligationGate));
builder.with_event_store(sql_store.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
.await
.expect("dispatch must succeed when gate allows");
let page = sql_store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(
page.items.len(),
1,
"one audit event must be persisted on allow through SqlEventStore"
);
let ev = &page.items[0];
assert_eq!(ev.outcome, EventOutcome::Success);
let data = &ev.payload;
let obligations_raw = data
.get("obligations")
.expect("Event.data JSON must contain 'obligations' key");
let obligations_arr = obligations_raw
.as_array()
.expect("'obligations' must be a JSON array");
assert!(
!obligations_arr.is_empty(),
"raw Event.data['obligations'] must be non-empty after SQL round-trip"
);
let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
.expect("Event.data must deserialize to AuditEvent after SQL round-trip");
assert_eq!(
audit.gate_impl, "SqlTestAllowWithObligationGate",
"gate_impl must survive the SQL text round-trip"
);
assert_eq!(
audit.decision,
khive_gate::AuditDecision::Allow,
"decision field must survive the SQL text round-trip"
);
assert_eq!(
audit.obligations.len(),
1,
"obligations must be non-empty after SQL round-trip (not silently defaulted to [])"
);
match &audit.obligations[0] {
Obligation::Audit { tag } => assert_eq!(
tag, "sql-path-billing.meter",
"Audit obligation tag must survive the SQL text round-trip"
),
other => panic!("expected Audit obligation, got {other:?}"),
}
}
#[tokio::test]
async fn audit_event_payload_shape_for_create_verb() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
reg.dispatch("create", serde_json::json!({"namespace": "test-ns"}))
.await
.unwrap();
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(count, 1, "exactly one audit event for one dispatch");
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
let ev = &page.items[0];
assert_eq!(ev.verb, "create", "ev.verb must be the dispatched verb");
assert_eq!(
ev.outcome,
EventOutcome::Success,
"ev.outcome must be Success on allow"
);
assert_eq!(
ev.namespace, "test-ns",
"ev.namespace must match the dispatch namespace"
);
let data = &ev.payload;
let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
.expect("ev.payload must deserialize to AuditEvent");
assert_eq!(
audit.decision,
khive_gate::AuditDecision::Allow,
"AuditEvent.decision must be Allow"
);
assert_eq!(audit.verb, "create", "AuditEvent.verb must be 'create'");
assert_eq!(
audit.namespace, "test-ns",
"AuditEvent.namespace must be preserved"
);
assert_eq!(
audit.gate_impl, "AllowAllGate",
"AuditEvent.gate_impl must name the gate implementation"
);
assert!(
audit.deny_reason.is_none(),
"AuditEvent.deny_reason must be None on Allow"
);
let payload_json: serde_json::Value =
serde_json::from_value(data.clone()).expect("data must be valid JSON");
assert_eq!(
payload_json["obligations"],
serde_json::Value::Array(Vec::new()),
"obligations must be [] on AllowAllGate"
);
}
#[tokio::test]
async fn audit_event_threads_target_id_from_dispatch_args() {
let store = Arc::new(MemoryEventStore::default());
let target = uuid::Uuid::new_v4();
let mut builder = VerbRegistryBuilder::new();
builder.register(AlphaPack);
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
reg.dispatch(
"create",
serde_json::json!({"namespace": "test-ns", "target_id": target}),
)
.await
.unwrap();
let page = store
.query_events(
EventFilter::default(),
PageRequest {
offset: 0,
limit: 10,
},
)
.await
.unwrap();
assert_eq!(
page.items[0].target_id,
Some(target),
"#282: audit event must carry target_id from dispatch params"
);
}
struct LinkResultPack {
result: std::sync::Mutex<Option<Result<Value, RuntimeError>>>,
}
impl LinkResultPack {
fn ok(value: Value) -> Self {
Self {
result: std::sync::Mutex::new(Some(Ok(value))),
}
}
fn err(message: &str) -> Self {
Self {
result: std::sync::Mutex::new(Some(Err(RuntimeError::InvalidInput(
message.to_string(),
)))),
}
}
}
impl khive_types::Pack for LinkResultPack {
const NAME: &'static str = "kg";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "link",
description: "test link handler",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for LinkResultPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
_verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
self.result
.lock()
.unwrap()
.take()
.expect("LinkResultPack dispatch called more than once in a test")
}
}
#[tokio::test]
async fn link_audit_enriches_successful_singleton_with_edge_v2() {
let store = Arc::new(MemoryEventStore::default());
let edge_id = uuid::Uuid::new_v4();
let source_id = uuid::Uuid::new_v4();
let target_id = uuid::Uuid::new_v4();
let edge_json = serde_json::json!({
"id": edge_id,
"namespace": "local",
"source_id": source_id,
"target_id": target_id,
"relation": "depends_on",
"weight": 1.0,
});
let mut builder = VerbRegistryBuilder::new();
builder.register(LinkResultPack::ok(edge_json));
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
reg.dispatch(
"link",
serde_json::json!({
"source_id": source_id,
"target_id": target_id,
"relation": "depends_on",
}),
)
.await
.unwrap();
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(
count, 1,
"exactly one deferred audit row must be persisted for a successful singleton link"
);
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
let ev = &page.items[0];
assert_eq!(ev.verb, "link");
assert_eq!(ev.outcome, EventOutcome::Success);
assert_eq!(
ev.payload_schema_version, 2,
"successful singleton link uses audit schema v2"
);
assert_eq!(
ev.target_id,
Some(edge_id),
"target_id must be the created/resolved edge id, not a raw caller arg"
);
assert_eq!(ev.payload["edge_id"], serde_json::json!(edge_id));
assert_eq!(ev.payload["source_id"], serde_json::json!(source_id));
assert_eq!(ev.payload["target_id"], serde_json::json!(target_id));
assert_eq!(ev.payload["relation"], "depends_on");
assert_eq!(ev.payload["weight"], 1.0);
assert_eq!(ev.payload["verb"], "link");
assert_eq!(ev.payload["decision"], "allow");
assert!(ev.payload.get("gate_impl").is_some());
}
#[tokio::test]
async fn link_audit_falls_back_to_v1_when_dispatch_fails() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(LinkResultPack::err("target endpoint not found"));
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
let err = reg
.dispatch(
"link",
serde_json::json!({
"source_id": "note:alpha",
"target_id": "note:missing",
"relation": "depends_on",
}),
)
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("not found")),
"the original dispatch error must be returned unchanged"
);
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(
page.items.len(),
1,
"a v1 fallback audit row must still be persisted on dispatch failure"
);
let ev = &page.items[0];
assert_eq!(
ev.payload_schema_version, 1,
"failed link keeps the v1 audit shape"
);
assert_eq!(
ev.outcome,
EventOutcome::Error,
"outcome reflects the dispatch result (Err), not the gate decision (Allow)"
);
assert!(
ev.duration_us >= 0,
"duration_us must still be populated (measured, not the Event::new \
default sentinel) on a failed dispatch"
);
assert!(
ev.target_id.is_none(),
"non-UUID caller-supplied ids do not spuriously populate target_id"
);
assert!(
ev.payload.get("edge_id").is_none(),
"v1 fallback must not carry edge enrichment fields"
);
let _: khive_gate::AuditEvent = serde_json::from_value(ev.payload.clone())
.expect("v1 fallback payload must deserialize as AuditEvent");
}
#[tokio::test]
async fn link_audit_falls_back_to_v1_when_result_missing_edge_fields() {
let store = Arc::new(MemoryEventStore::default());
let target_arg = uuid::Uuid::new_v4();
let mut builder = VerbRegistryBuilder::new();
builder.register(LinkResultPack::ok(serde_json::json!({"ok": true})));
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
reg.dispatch(
"link",
serde_json::json!({
"source_id": uuid::Uuid::new_v4(),
"target_id": target_arg,
"relation": "depends_on",
}),
)
.await
.unwrap();
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
assert_eq!(page.items.len(), 1);
let ev = &page.items[0];
assert_eq!(
ev.payload_schema_version, 1,
"an unparsable success result falls back to v1 rather than dropping the audit row"
);
assert_eq!(ev.outcome, EventOutcome::Success);
assert_eq!(
ev.target_id,
Some(target_arg),
"v1 fallback still extracts target_id from the raw dispatch args"
);
assert!(ev.payload.get("edge_id").is_none());
}
#[tokio::test]
async fn link_audit_bulk_links_get_no_enrichment() {
let store = Arc::new(MemoryEventStore::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(LinkResultPack::ok(serde_json::json!({
"attempted": 2, "created": 2, "skipped": 0, "failed": 0
})));
builder.with_event_store(store.clone());
builder.with_default_namespace("test-ns");
let reg = builder.build().expect("registry builds");
reg.dispatch(
"link",
serde_json::json!({
"links": [
{"source_id": "a", "target_id": "b", "relation": "depends_on"},
{"source_id": "c", "target_id": "d", "relation": "depends_on"},
],
}),
)
.await
.unwrap();
let count = store.count_events(EventFilter::default()).await.unwrap();
assert_eq!(
count, 1,
"bulk `links` gets exactly one v1 audit row (deferred until dispatch \
resolves like every other Allow-outcome row since ADR-103 Stage 1, \
but never v2-enriched — enrichment is singleton-`link`-only)"
);
let page = store
.query_events(
EventFilter::default(),
PageRequest {
limit: 10,
offset: 0,
},
)
.await
.unwrap();
let ev = &page.items[0];
assert_eq!(
ev.payload_schema_version, 1,
"bulk link mode is out of scope for #676's events.target_id enrichment"
);
assert!(ev.target_id.is_none());
}
#[test]
fn link_audit_success_from_result_extracts_edge_fields() {
let gate_req = GateRequest::new(
ActorRef::anonymous(),
Namespace::local(),
"link",
serde_json::json!({}),
);
let decision = GateDecision::Allow {
obligations: vec![],
};
let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
let edge_id = uuid::Uuid::new_v4();
let source_id = uuid::Uuid::new_v4();
let target_id = uuid::Uuid::new_v4();
let result = serde_json::json!({
"id": edge_id,
"source_id": source_id,
"target_id": target_id,
"relation": "depends_on",
"weight": 0.5,
});
let (returned_id, payload) = link_audit_success_from_result(audit, &result)
.expect("well-formed edge JSON must produce an enriched payload");
assert_eq!(returned_id, edge_id);
assert_eq!(payload["edge_id"], serde_json::json!(edge_id));
assert_eq!(payload["relation"], "depends_on");
assert_eq!(payload["weight"], 0.5);
assert_eq!(
payload["verb"], "link",
"v1 AuditEvent fields must flatten into the v2 payload"
);
}
#[test]
fn link_audit_success_from_result_rejects_incomplete_or_malformed_result() {
let gate_req = GateRequest::new(
ActorRef::anonymous(),
Namespace::local(),
"link",
serde_json::json!({}),
);
let decision = GateDecision::Allow {
obligations: vec![],
};
let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
assert!(
link_audit_success_from_result(
audit.clone(),
&serde_json::json!({"id": uuid::Uuid::new_v4()}),
)
.is_none(),
"missing source_id/target_id/relation/weight must not enrich"
);
assert!(
link_audit_success_from_result(audit, &serde_json::json!({"id": "not-a-uuid"}))
.is_none(),
"a non-UUID id must not enrich"
);
}
}
#[cfg(test)]
mod dep_tests {
use super::*;
use async_trait::async_trait;
use khive_types::Pack;
use serde_json::Value;
struct KgDepPack;
struct MemoryDepPack;
struct ADepPack;
struct BDepPack;
impl Pack for KgDepPack {
const NAME: &'static str = "kg_dep";
const NOTE_KINDS: &'static [&'static str] = &["observation"];
const ENTITY_KINDS: &'static [&'static str] = &["concept"];
const HANDLERS: &'static [HandlerDef] = &[];
}
impl Pack for MemoryDepPack {
const NAME: &'static str = "memory_dep";
const NOTE_KINDS: &'static [&'static str] = &["memory"];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
const REQUIRES: &'static [&'static str] = &["kg_dep"];
}
impl Pack for ADepPack {
const NAME: &'static str = "pack_a";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
const REQUIRES: &'static [&'static str] = &["pack_b"];
}
impl Pack for BDepPack {
const NAME: &'static str = "pack_b";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
const REQUIRES: &'static [&'static str] = &["pack_a"];
}
#[async_trait]
impl PackRuntime for KgDepPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!(
"KgDepPack has no verbs: {verb}"
)))
}
}
#[async_trait]
impl PackRuntime for MemoryDepPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
fn requires(&self) -> &'static [&'static str] {
Self::REQUIRES
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!(
"MemoryDepPack has no verbs: {verb}"
)))
}
}
#[async_trait]
impl PackRuntime for ADepPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
fn requires(&self) -> &'static [&'static str] {
Self::REQUIRES
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!(
"ADepPack has no verbs: {verb}"
)))
}
}
#[async_trait]
impl PackRuntime for BDepPack {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
fn requires(&self) -> &'static [&'static str] {
Self::REQUIRES
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!(
"BDepPack has no verbs: {verb}"
)))
}
}
#[test]
fn test_pack_deps_happy_path() {
let mut builder = VerbRegistryBuilder::new();
builder.register(MemoryDepPack);
builder.register(KgDepPack);
let reg = builder
.build()
.expect("kg_dep satisfies memory_dep dependency");
assert_eq!(reg.pack_requires("memory_dep").unwrap(), &["kg_dep"]);
let names = reg.pack_names();
let kg_pos = names.iter().position(|&n| n == "kg_dep").unwrap();
let mem_pos = names.iter().position(|&n| n == "memory_dep").unwrap();
assert!(
kg_pos < mem_pos,
"kg_dep must be loaded before memory_dep; order: {names:?}"
);
}
#[test]
fn test_pack_deps_missing() {
let mut builder = VerbRegistryBuilder::new();
builder.register(MemoryDepPack);
let err = match builder.build() {
Ok(_) => panic!("expected Err, got Ok"),
Err(e) => e,
};
assert!(
matches!(err, RuntimeError::MissingPackDependency(_)),
"expected MissingPackDependency, got {err:?}"
);
let msg = err.to_string();
assert!(
msg.contains("memory_dep"),
"error must name the dependent pack: {msg}"
);
assert!(
msg.contains("kg_dep"),
"error must name the missing dep: {msg}"
);
}
#[test]
fn test_pack_deps_circular() {
let mut builder = VerbRegistryBuilder::new();
builder.register(ADepPack);
builder.register(BDepPack);
let err = match builder.build() {
Ok(_) => panic!("expected Err, got Ok"),
Err(e) => e,
};
assert!(
matches!(err, RuntimeError::CircularPackDependency(_)),
"expected CircularPackDependency, got {err:?}"
);
let msg = err.to_string();
assert!(msg.contains("pack_a"), "error must name pack_a: {msg}");
assert!(msg.contains("pack_b"), "error must name pack_b: {msg}");
}
#[test]
fn test_pack_deps_no_deps() {
struct NoDepsA;
struct NoDepsB;
impl Pack for NoDepsA {
const NAME: &'static str = "no_deps_a";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
}
impl Pack for NoDepsB {
const NAME: &'static str = "no_deps_b";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
}
#[async_trait]
impl PackRuntime for NoDepsA {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!("NoDepsA: {verb}")))
}
}
#[async_trait]
impl PackRuntime for NoDepsB {
fn name(&self) -> &str {
Self::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
Self::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
Self::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
Self::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_: Value,
_: &VerbRegistry,
_: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Err(RuntimeError::InvalidInput(format!("NoDepsB: {verb}")))
}
}
let mut builder = VerbRegistryBuilder::new();
builder.register(NoDepsA);
builder.register(NoDepsB);
let reg = builder.build().expect("packs with REQUIRES=&[] build");
assert_eq!(reg.pack_requires("no_deps_a").unwrap(), &[] as &[&str]);
assert_eq!(reg.pack_requires("no_deps_b").unwrap(), &[] as &[&str]);
}
}
#[cfg(test)]
mod hook_tests {
use super::*;
use async_trait::async_trait;
use khive_types::Pack;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex as StdMutex;
struct SimplePack;
impl Pack for SimplePack {
const NAME: &'static str = "simple";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
name: "ping",
description: "ping",
visibility: Visibility::Verb,
category: VerbCategory::Assertive,
params: &[],
}];
}
#[async_trait]
impl PackRuntime for SimplePack {
fn name(&self) -> &str {
SimplePack::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
SimplePack::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
SimplePack::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
SimplePack::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({ "verb": verb }))
}
}
#[derive(Default)]
struct CountingHook {
calls: AtomicUsize,
last_verb: StdMutex<String>,
}
#[async_trait]
impl DispatchHook for CountingHook {
async fn on_dispatch(&self, view: &EventView) {
self.calls.fetch_add(1, Ordering::SeqCst);
*self.last_verb.lock().unwrap() = view.event.verb.clone();
}
}
#[tokio::test]
async fn dispatch_hook_fires_on_successful_dispatch() {
let hook = Arc::new(CountingHook::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
builder.with_dispatch_hook(hook.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("ping", Value::Null).await.unwrap();
assert_eq!(
hook.calls.load(Ordering::SeqCst),
1,
"hook must fire once per successful dispatch"
);
assert_eq!(
hook.last_verb.lock().unwrap().as_str(),
"ping",
"hook event must carry the dispatched verb"
);
}
#[tokio::test]
async fn dispatch_hook_fires_multiple_times() {
let hook = Arc::new(CountingHook::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
builder.with_dispatch_hook(hook.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("ping", Value::Null).await.unwrap();
reg.dispatch("ping", Value::Null).await.unwrap();
reg.dispatch("ping", Value::Null).await.unwrap();
assert_eq!(
hook.calls.load(Ordering::SeqCst),
3,
"hook must fire once per successful dispatch"
);
}
#[tokio::test]
async fn dispatch_hook_does_not_fire_on_unknown_verb() {
let hook = Arc::new(CountingHook::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
builder.with_dispatch_hook(hook.clone());
let reg = builder.build().expect("registry builds");
let _ = reg.dispatch("nonexistent", Value::Null).await;
assert_eq!(
hook.calls.load(Ordering::SeqCst),
0,
"hook must NOT fire for unknown verb (dispatch returns error)"
);
}
#[tokio::test]
async fn dispatch_hook_does_not_fire_on_gate_deny() {
use khive_gate::{Gate, GateDecision, GateError};
#[derive(Debug)]
struct AlwaysDenyGate;
impl Gate for AlwaysDenyGate {
fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
Ok(GateDecision::deny("test deny"))
}
}
let hook = Arc::new(CountingHook::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
builder.with_gate(Arc::new(AlwaysDenyGate));
builder.with_dispatch_hook(hook.clone());
let reg = builder.build().expect("registry builds");
let err = reg.dispatch("ping", Value::Null).await.unwrap_err();
assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
assert_eq!(
hook.calls.load(Ordering::SeqCst),
0,
"hook must NOT fire when gate denies dispatch"
);
}
#[tokio::test]
async fn dispatch_hook_event_carries_namespace_from_params() {
let hook = Arc::new(CountingHook::default());
#[derive(Default)]
struct NsCapturingHook {
ns: StdMutex<String>,
}
#[async_trait]
impl DispatchHook for NsCapturingHook {
async fn on_dispatch(&self, view: &EventView) {
*self.ns.lock().unwrap() = view.event.namespace.clone();
}
}
let ns_hook = Arc::new(NsCapturingHook::default());
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
builder.with_dispatch_hook(ns_hook.clone());
let reg = builder.build().expect("registry builds");
reg.dispatch("ping", serde_json::json!({"namespace": "tenant-abc"}))
.await
.unwrap();
assert_eq!(
ns_hook.ns.lock().unwrap().as_str(),
"tenant-abc",
"dispatch hook event must carry the resolved namespace"
);
drop(hook);
}
#[tokio::test]
async fn no_dispatch_hook_configured_dispatch_succeeds() {
let mut builder = VerbRegistryBuilder::new();
builder.register(SimplePack);
let reg = builder.build().expect("registry builds");
let res = reg.dispatch("ping", Value::Null).await.unwrap();
assert_eq!(res["verb"], "ping");
}
}
#[cfg(test)]
mod help_tests {
use super::*;
use async_trait::async_trait;
use khive_types::Pack;
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
static CREATE_PARAMS: [ParamDef; 2] = [
ParamDef {
name: "kind",
param_type: "string",
required: true,
description: "Granular kind (concept | document | ...).",
},
ParamDef {
name: "name",
param_type: "string",
required: false,
description: "Human-readable name.",
},
];
static RECALL_PARAMS: [ParamDef; 2] = [
ParamDef {
name: "query",
param_type: "string",
required: true,
description: "Semantic recall query.",
},
ParamDef {
name: "limit",
param_type: "integer",
required: false,
description: "Maximum memories to return.",
},
];
static EMBED_PARAMS: [ParamDef; 0] = [];
struct HelpPack {
invocations: Arc<AtomicUsize>,
}
impl Pack for HelpPack {
const NAME: &'static str = "helptest";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[
HandlerDef {
name: "create",
description: "Create an entity or note",
visibility: Visibility::Verb,
category: VerbCategory::Commissive,
params: &CREATE_PARAMS,
},
HandlerDef {
name: "recall",
description: "Recall memory notes with decay-aware hybrid ranking",
visibility: Visibility::Verb,
category: VerbCategory::Assertive,
params: &RECALL_PARAMS,
},
HandlerDef {
name: "recall.embed",
description: "Return the embedding vector used by memory recall",
visibility: Visibility::Subhandler,
category: VerbCategory::Assertive,
params: &EMBED_PARAMS,
},
];
}
#[async_trait]
impl PackRuntime for HelpPack {
fn name(&self) -> &str {
HelpPack::NAME
}
fn note_kinds(&self) -> &'static [&'static str] {
HelpPack::NOTE_KINDS
}
fn entity_kinds(&self) -> &'static [&'static str] {
HelpPack::ENTITY_KINDS
}
fn handlers(&self) -> &'static [HandlerDef] {
HelpPack::HANDLERS
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
self.invocations.fetch_add(1, Ordering::SeqCst);
Ok(serde_json::json!({ "pack": "helptest", "verb": verb }))
}
}
fn build_help_registry(invocations: Arc<AtomicUsize>) -> VerbRegistry {
let mut builder = VerbRegistryBuilder::new();
builder.register(HelpPack { invocations });
builder.build().expect("help registry builds")
}
#[tokio::test]
async fn test_help_true_returns_schema_for_kg_create() {
let invocations = Arc::new(AtomicUsize::new(0));
let reg = build_help_registry(invocations.clone());
let result = reg
.dispatch("create", serde_json::json!({ "help": true }))
.await
.expect("help=true must succeed for a known verb");
assert_eq!(result["verb"], "create", "envelope must name the verb");
assert_eq!(
result["pack"], "helptest",
"envelope must name the owning pack"
);
assert!(
result["description"].as_str().is_some(),
"description must be a string"
);
let params = result["params"]
.as_array()
.expect("params must be a JSON array");
assert!(!params.is_empty(), "params array must not be empty");
let kind_param = params.iter().find(|p| p["name"] == "kind");
assert!(
kind_param.is_some(),
"params array must include the 'kind' parameter"
);
let kind_param = kind_param.unwrap();
assert_eq!(
kind_param["required"],
serde_json::json!(true),
"'kind' must be required"
);
assert_eq!(kind_param["type"], "string", "'kind' type must be 'string'");
}
#[tokio::test]
async fn test_help_true_returns_schema_for_recall() {
let invocations = Arc::new(AtomicUsize::new(0));
let reg = build_help_registry(invocations.clone());
let result = reg
.dispatch("recall", serde_json::json!({ "help": true }))
.await
.expect("help=true must succeed for recall");
assert_eq!(result["verb"], "recall");
assert_eq!(result["pack"], "helptest");
let params = result["params"]
.as_array()
.expect("params must be a JSON array");
let query_param = params.iter().find(|p| p["name"] == "query");
assert!(query_param.is_some(), "params must include 'query'");
let query_param = query_param.unwrap();
assert_eq!(
query_param["required"],
serde_json::json!(true),
"'query' must be required"
);
let limit_param = params.iter().find(|p| p["name"] == "limit");
assert!(limit_param.is_some(), "params must include 'limit'");
let limit_param = limit_param.unwrap();
assert_eq!(
limit_param["required"],
serde_json::json!(false),
"'limit' must be optional"
);
}
#[tokio::test]
async fn test_help_true_does_not_execute_the_verb() {
let invocations = Arc::new(AtomicUsize::new(0));
let reg = build_help_registry(invocations.clone());
reg.dispatch("create", serde_json::json!({ "help": true }))
.await
.expect("help=true must succeed");
reg.dispatch("recall", serde_json::json!({ "help": true }))
.await
.expect("help=true must succeed");
assert_eq!(
invocations.load(Ordering::SeqCst),
0,
"pack dispatch MUST NOT be invoked when help=true; \
got {} invocation(s)",
invocations.load(Ordering::SeqCst)
);
reg.dispatch("create", serde_json::json!({}))
.await
.expect("normal dispatch must succeed");
assert_eq!(
invocations.load(Ordering::SeqCst),
1,
"pack dispatch must fire exactly once for a normal call"
);
}
#[tokio::test]
async fn help_true_on_subhandler_returns_callable_via_mcp_false() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let result = reg
.dispatch("recall.embed", serde_json::json!({ "help": true }))
.await
.expect("help=true on subhandler must succeed (no permission check on help path)");
assert_eq!(
result["callable_via_mcp"],
serde_json::json!(false),
"subhandler help must carry callable_via_mcp: false"
);
assert_eq!(
result["visibility"], "internal",
"subhandler help must carry visibility: internal"
);
assert_eq!(result["verb"], "recall.embed");
assert_eq!(result["pack"], "helptest");
}
#[tokio::test]
async fn help_true_on_public_verb_does_not_have_callable_via_mcp_false() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let result = reg
.dispatch("create", serde_json::json!({ "help": true }))
.await
.expect("help=true on public verb must succeed");
assert_ne!(
result.get("callable_via_mcp"),
Some(&serde_json::json!(false)),
"public verb help must NOT carry callable_via_mcp: false"
);
assert_ne!(
result.get("visibility"),
Some(&serde_json::json!("internal")),
"public verb help must NOT carry visibility: internal"
);
}
#[tokio::test]
async fn help_true_on_unknown_verb_returns_error() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let err = reg
.dispatch("nonexistent_verb", serde_json::json!({ "help": true }))
.await
.unwrap_err();
assert!(
matches!(err, RuntimeError::InvalidInput(_)),
"help=true on unknown verb must return InvalidInput, got {err:?}"
);
let msg = err.to_string();
assert!(
msg.contains("nonexistent_verb"),
"error must name the unknown verb: {msg}"
);
}
#[tokio::test]
async fn help_true_on_subhandler_includes_params_field() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let result = reg
.dispatch("recall.embed", serde_json::json!({ "help": true }))
.await
.expect("help=true on subhandler must succeed");
let params = result
.get("params")
.expect("subhandler help must include 'params' field");
assert!(
params.is_array(),
"subhandler help params must be a JSON array"
);
}
#[tokio::test]
async fn help_true_unknown_verb_available_list_excludes_subhandlers() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let err = reg
.dispatch("not_a_verb", serde_json::json!({ "help": true }))
.await
.unwrap_err();
let msg = err.to_string();
assert!(
!msg.contains("recall.embed"),
"unknown-verb help error must not advertise subhandler recall.embed: {msg}"
);
assert!(
msg.contains("create"),
"unknown-verb help error must still list public verb 'create': {msg}"
);
assert!(
msg.contains("recall"),
"unknown-verb help error must still list public verb 'recall': {msg}"
);
}
#[tokio::test]
async fn dispatch_unknown_verb_available_list_excludes_subhandlers() {
let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
let err = reg
.dispatch("not_a_verb", serde_json::json!({}))
.await
.unwrap_err();
let msg = err.to_string();
assert!(
!msg.contains("recall.embed"),
"dispatch unknown-verb error must not advertise subhandler recall.embed: {msg}"
);
assert!(
msg.contains("create"),
"dispatch unknown-verb error must still list public verb 'create': {msg}"
);
assert!(
msg.contains("recall"),
"dispatch unknown-verb error must still list public verb 'recall': {msg}"
);
}
struct SchemaPack {
pack_name: &'static str,
statements: &'static [&'static str],
}
impl Pack for SchemaPack {
const NAME: &'static str = "schema-pack";
const NOTE_KINDS: &'static [&'static str] = &[];
const ENTITY_KINDS: &'static [&'static str] = &[];
const HANDLERS: &'static [HandlerDef] = &[];
}
#[async_trait]
impl PackRuntime for SchemaPack {
fn name(&self) -> &str {
self.pack_name
}
fn note_kinds(&self) -> &'static [&'static str] {
&[]
}
fn entity_kinds(&self) -> &'static [&'static str] {
&[]
}
fn handlers(&self) -> &'static [HandlerDef] {
&[]
}
fn schema_plan(&self) -> SchemaPlan {
SchemaPlan {
pack: self.pack_name,
statements: self.statements,
}
}
async fn dispatch(
&self,
verb: &str,
_params: Value,
_registry: &VerbRegistry,
_token: &NamespaceToken,
) -> Result<Value, RuntimeError> {
Ok(serde_json::json!({ "pack": self.pack_name, "verb": verb }))
}
}
#[test]
fn all_schema_plans_named_returns_correct_pairs() {
let mut builder = VerbRegistryBuilder::new();
builder.register_boxed(Box::new(SchemaPack {
pack_name: "alpha",
statements: &["CREATE TABLE IF NOT EXISTS t_alpha (id INTEGER PRIMARY KEY)"],
}));
builder.register_boxed(Box::new(SchemaPack {
pack_name: "beta",
statements: &[],
}));
let reg = builder.build().expect("registry builds");
let named = reg.all_schema_plans_named();
assert_eq!(named.len(), 2);
let alpha_entry = named.iter().find(|(n, _)| *n == "alpha");
let beta_entry = named.iter().find(|(n, _)| *n == "beta");
assert!(alpha_entry.is_some(), "alpha must appear in named plans");
assert!(beta_entry.is_some(), "beta must appear in named plans");
let (_, alpha_plan) = alpha_entry.unwrap();
assert_eq!(alpha_plan.statements.len(), 1);
assert!(!alpha_plan.is_empty());
let (_, beta_plan) = beta_entry.unwrap();
assert!(beta_plan.is_empty());
}
#[tokio::test]
async fn apply_schema_plans_with_map_routes_to_correct_backend() {
use khive_storage::types::{SqlStatement, SqlValue};
let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
let pack_backend =
khive_db::StorageBackend::memory().expect("pack-specific memory backend");
let mut builder = VerbRegistryBuilder::new();
builder.register_boxed(Box::new(SchemaPack {
pack_name: "routed",
statements: &["CREATE TABLE IF NOT EXISTS t_routed (id INTEGER PRIMARY KEY)"],
}));
let reg = builder.build().expect("registry builds");
let mut backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
backend_map.insert("routed", &pack_backend);
reg.apply_schema_plans_with_map(&backend_map, &default_backend)
.expect("schema application must not collide");
let mut writer = pack_backend.sql().writer().await.expect("writer");
let result = writer
.execute(SqlStatement {
sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
params: vec![SqlValue::Integer(1)],
label: None,
})
.await;
assert!(
result.is_ok(),
"t_routed must exist on pack_backend after routing: {result:?}"
);
let mut default_writer = default_backend.sql().writer().await.expect("writer");
let default_result = default_writer
.execute(SqlStatement {
sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
params: vec![SqlValue::Integer(2)],
label: None,
})
.await;
assert!(
default_result.is_err(),
"t_routed must NOT exist on default_backend (table should not be there)"
);
}
#[tokio::test]
async fn apply_schema_plans_with_map_falls_back_to_default_for_unmapped_packs() {
use khive_storage::types::{SqlStatement, SqlValue};
let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
let mut builder = VerbRegistryBuilder::new();
builder.register_boxed(Box::new(SchemaPack {
pack_name: "unmapped",
statements: &["CREATE TABLE IF NOT EXISTS t_unmapped (id INTEGER PRIMARY KEY)"],
}));
let reg = builder.build().expect("registry builds");
let backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
reg.apply_schema_plans_with_map(&backend_map, &default_backend)
.expect("schema application must not collide");
let mut writer = default_backend.sql().writer().await.expect("writer");
let result = writer
.execute(SqlStatement {
sql: "INSERT INTO t_unmapped (id) VALUES (?1)".into(),
params: vec![SqlValue::Integer(1)],
label: None,
})
.await;
assert!(
result.is_ok(),
"t_unmapped must exist on default_backend for unmapped pack: {result:?}"
);
}
#[test]
fn apply_schema_plans_with_map_collision_is_an_error() {
let backend = khive_db::StorageBackend::memory().expect("memory backend");
let empty_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
let mut builder = VerbRegistryBuilder::new();
builder.register_boxed(Box::new(SchemaPack {
pack_name: "pack_alpha",
statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
}));
builder.register_boxed(Box::new(SchemaPack {
pack_name: "pack_beta",
statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
}));
let registry = builder.build().expect("registry builds");
let result = registry.apply_schema_plans_with_map(&empty_map, &backend);
assert!(
result.is_err(),
"two packs declaring the same table on the same backend must produce a collision error"
);
let err = result.unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("pack_alpha"),
"collision error must name first pack; got: {msg}"
);
assert!(
msg.contains("pack_beta"),
"collision error must name second pack; got: {msg}"
);
assert!(
msg.contains("collision_table"),
"collision error must name the table; got: {msg}"
);
}
}