1use std::collections::{HashMap, HashSet, VecDeque};
14use std::sync::Arc;
15use std::time::Instant;
16
17use crate::runtime::NamespaceToken;
18use async_trait::async_trait;
19use khive_gate::{AllowAllGate, AuditEvent, GateDecision, GateRef, GateRequest};
20use khive_storage::{Event, EventStore, EventView, SubstrateKind};
21use khive_types::{EventKind, EventOutcome, Namespace};
22use serde_json::Value;
23
24pub use khive_types::{
25 EdgeEndpointRule, EndpointKind, EntityTypeDef, HandlerDef, NoteKindSpec, NoteLifecycleSpec,
26 PackSchemaPlan, ParamDef, VerbCategory, VerbPresentationPolicy, Visibility,
27};
28#[allow(deprecated)]
30pub use khive_types::VerbDef;
31
32use crate::validation::ValidationRule;
33
34#[derive(Debug, Default, Clone)]
44pub struct SchemaPlan {
45 pub pack: &'static str,
47 pub statements: &'static [&'static str],
51}
52
53impl SchemaPlan {
54 pub const fn empty() -> Self {
59 Self {
60 pack: "",
61 statements: &[],
62 }
63 }
64
65 pub fn is_empty(&self) -> bool {
67 self.statements.is_empty()
68 }
69}
70
71#[async_trait]
76pub trait DispatchHook: Send + Sync {
77 async fn on_dispatch(&self, view: &EventView);
82}
83
84use crate::error::{
85 CircularPackDependency, MissingPackDependencies, MissingPackDependency, RuntimeError,
86};
87use crate::KhiveRuntime;
88
89#[async_trait]
98pub trait PackRuntime: Send + Sync {
99 fn name(&self) -> &str;
101
102 fn note_kinds(&self) -> &'static [&'static str];
104
105 fn entity_kinds(&self) -> &'static [&'static str];
107
108 fn handlers(&self) -> &'static [HandlerDef];
110
111 fn edge_rules(&self) -> &'static [EdgeEndpointRule] {
115 &[]
116 }
117
118 fn entity_types(&self) -> &'static [EntityTypeDef] {
122 &[]
123 }
124
125 fn requires(&self) -> &'static [&'static str] {
128 &[]
129 }
130
131 fn note_kind_specs(&self) -> &'static [NoteKindSpec] {
138 &[]
139 }
140
141 fn kind_hook(&self, _kind: &str) -> Option<Arc<dyn KindHook>> {
149 None
150 }
151
152 fn schema_plan(&self) -> SchemaPlan {
168 SchemaPlan::empty()
169 }
170
171 fn validation_rules(&self) -> &'static [ValidationRule] {
179 &[]
180 }
181
182 fn register_embedders(&self, _runtime: &KhiveRuntime) {}
188
189 fn register_entity_type_validator(&self, _runtime: &KhiveRuntime) {}
195
196 fn register_entity_type_validator_with_types(
203 &self,
204 runtime: &KhiveRuntime,
205 _pack_entity_types: &[EntityTypeDef],
206 ) {
207 self.register_entity_type_validator(runtime);
208 }
209
210 fn register_note_mutation_hook(&self, _runtime: &KhiveRuntime) {}
217
218 async fn warm(&self) {}
222
223 fn registered_embedding_model_names(&self) -> Vec<String> {
228 Vec::new()
229 }
230
231 async fn dispatch(
238 &self,
239 verb: &str,
240 params: Value,
241 registry: &VerbRegistry,
242 token: &NamespaceToken,
243 ) -> Result<Value, RuntimeError>;
244}
245
246#[async_trait]
260pub trait KindHook: Send + Sync + std::fmt::Debug {
261 async fn prepare_create(
267 &self,
268 runtime: &KhiveRuntime,
269 args: &mut Value,
270 ) -> Result<(), RuntimeError>;
271
272 async fn after_create(
281 &self,
282 runtime: &KhiveRuntime,
283 id: uuid::Uuid,
284 args: &Value,
285 ) -> Result<(), RuntimeError>;
286}
287
288#[async_trait]
296pub trait PackByIdResolver: Send + Sync {
297 async fn resolve_by_id(
307 &self,
308 id: uuid::Uuid,
309 ) -> Result<Option<crate::Resolved>, crate::RuntimeError>;
310
311 async fn resolve_by_id_including_deleted(
316 &self,
317 id: uuid::Uuid,
318 ) -> Result<Option<crate::Resolved>, crate::RuntimeError> {
319 self.resolve_by_id(id).await
320 }
321
322 async fn delete_by_id(
331 &self,
332 id: uuid::Uuid,
333 hard: bool,
334 ) -> Result<serde_json::Value, crate::RuntimeError>;
335}
336
337pub struct VerbRegistryBuilder {
342 packs: Vec<Box<dyn PackRuntime>>,
343 resolvers: Vec<(String, Box<dyn PackByIdResolver>)>,
344 gate: GateRef,
345 default_namespace: String,
346 visible_namespaces: Vec<Namespace>,
354 actor_id: Option<String>,
357 event_store: Option<Arc<dyn EventStore>>,
364 dispatch_hook: Option<Arc<dyn DispatchHook>>,
370}
371
372impl VerbRegistryBuilder {
373 pub fn new() -> Self {
375 Self {
376 packs: Vec::new(),
377 resolvers: Vec::new(),
378 gate: std::sync::Arc::new(AllowAllGate),
379 default_namespace: Namespace::local().as_str().to_string(),
380 visible_namespaces: vec![],
381 actor_id: None,
382 event_store: None,
383 dispatch_hook: None,
384 }
385 }
386
387 pub fn with_visible_namespaces(&mut self, ns: Vec<Namespace>) -> &mut Self {
395 self.visible_namespaces = ns;
396 self
397 }
398
399 pub fn with_actor_id(&mut self, actor_id: Option<String>) -> &mut Self {
406 self.actor_id = actor_id;
407 self
408 }
409
410 pub fn register<P: khive_types::Pack + PackRuntime + 'static>(&mut self, pack: P) -> &mut Self {
413 self.packs.push(Box::new(pack));
414 self
415 }
416
417 pub(crate) fn register_boxed(&mut self, pack: Box<dyn PackRuntime>) -> &mut Self {
424 self.packs.push(pack);
425 self
426 }
427
428 pub fn register_resolver(
433 &mut self,
434 name: impl Into<String>,
435 resolver: Box<dyn PackByIdResolver>,
436 ) -> &mut Self {
437 self.resolvers.push((name.into(), resolver));
438 self
439 }
440
441 pub fn with_gate(&mut self, gate: GateRef) -> &mut Self {
448 self.gate = gate;
449 self
450 }
451
452 pub fn with_default_namespace(&mut self, ns: impl Into<String>) -> &mut Self {
457 self.default_namespace = ns.into();
458 self
459 }
460
461 pub fn with_event_store(&mut self, store: Arc<dyn EventStore>) -> &mut Self {
470 self.event_store = Some(store);
471 self
472 }
473
474 pub fn with_dispatch_hook(&mut self, hook: Arc<dyn DispatchHook>) -> &mut Self {
485 self.dispatch_hook = Some(hook);
486 self
487 }
488
489 pub fn build(self) -> Result<VerbRegistry, RuntimeError> {
495 let packs = self.packs;
496 let mut name_to_idx: HashMap<&str, usize> = HashMap::with_capacity(packs.len());
497 for (idx, pack) in packs.iter().enumerate() {
498 if let Some(prev_idx) = name_to_idx.insert(pack.name(), idx) {
499 return Err(RuntimeError::PackRedeclared {
500 name: pack.name().to_string(),
501 first_idx: prev_idx,
502 second_idx: idx,
503 });
504 }
505 }
506
507 let mut missing: Vec<MissingPackDependency> = Vec::new();
508 let mut indegree = vec![0usize; packs.len()];
509 let mut dependents: Vec<Vec<usize>> = vec![Vec::new(); packs.len()];
510
511 for (idx, pack) in packs.iter().enumerate() {
512 for &requires in pack.requires() {
513 match name_to_idx.get(requires).copied() {
514 Some(dep_idx) => {
515 dependents[dep_idx].push(idx);
516 indegree[idx] += 1;
517 }
518 None => missing.push(MissingPackDependency {
519 from: pack.name().to_string(),
520 requires: requires.to_string(),
521 }),
522 }
523 }
524 }
525
526 if !missing.is_empty() {
527 return if missing.len() == 1 {
528 Err(RuntimeError::MissingPackDependency(missing.remove(0)))
529 } else {
530 Err(RuntimeError::MissingPackDependencies(
531 MissingPackDependencies { missing },
532 ))
533 };
534 }
535
536 let mut ready: VecDeque<usize> = indegree
537 .iter()
538 .enumerate()
539 .filter_map(|(idx, degree)| (*degree == 0).then_some(idx))
540 .collect();
541 let mut ordered_indices = Vec::with_capacity(packs.len());
542
543 while let Some(idx) = ready.pop_front() {
544 ordered_indices.push(idx);
545 for &dep_idx in &dependents[idx] {
546 indegree[dep_idx] -= 1;
547 if indegree[dep_idx] == 0 {
548 ready.push_back(dep_idx);
549 }
550 }
551 }
552
553 if ordered_indices.len() != packs.len() {
554 let cycle_nodes: HashSet<usize> = indegree
555 .iter()
556 .enumerate()
557 .filter_map(|(idx, degree)| (*degree > 0).then_some(idx))
558 .collect();
559 let cycle = find_pack_dependency_cycle(&packs, &name_to_idx, &cycle_nodes);
560 return Err(RuntimeError::CircularPackDependency(
561 CircularPackDependency { cycle },
562 ));
563 }
564
565 let mut slots: Vec<Option<Box<dyn PackRuntime>>> = packs.into_iter().map(Some).collect();
566 let ordered_packs: Vec<Box<dyn PackRuntime>> = ordered_indices
567 .into_iter()
568 .map(|idx| slots[idx].take().expect("topological index must exist"))
569 .collect();
570
571 validate_unique_note_kinds(&ordered_packs)?;
572 validate_unique_verb_names(&ordered_packs)?;
573 validate_unique_entity_types(&ordered_packs)?;
574
575 let available_verbs: Vec<&'static str> = ordered_packs
576 .iter()
577 .flat_map(|p| p.handlers().iter())
578 .filter(|h| matches!(h.visibility, Visibility::Verb))
579 .map(|h| h.name)
580 .collect();
581
582 Ok(VerbRegistry {
583 packs: Arc::new(ordered_packs),
584 resolvers: Arc::new(self.resolvers),
585 gate: self.gate,
586 default_namespace: self.default_namespace,
587 visible_namespaces: self.visible_namespaces,
588 actor_id: self.actor_id,
589 event_store: self.event_store,
590 dispatch_hook: self.dispatch_hook,
591 available_verbs: Arc::new(available_verbs),
592 reference_ring: Arc::new(crate::reference_ring::ReferenceRing::new()),
593 })
594 }
595}
596
597fn validate_unique_note_kinds(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
603 let mut seen: HashMap<&str, &str> = HashMap::new();
604 for pack in packs {
605 for &kind in pack.note_kinds() {
606 if let Some(first_pack) = seen.insert(kind, pack.name()) {
607 return Err(RuntimeError::InvalidInput(format!(
608 "duplicate note kind {kind:?}: claimed by both {first_pack:?} and {:?}",
609 pack.name()
610 )));
611 }
612 }
613 }
614 Ok(())
615}
616
617fn validate_unique_verb_names(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
624 let mut seen: HashMap<&str, &str> = HashMap::new();
625 for pack in packs {
626 for handler in pack.handlers() {
627 if !matches!(handler.visibility, Visibility::Verb) {
628 continue;
629 }
630 if let Some(first_pack) = seen.insert(handler.name, pack.name()) {
631 return Err(RuntimeError::VerbCollision {
632 verb: handler.name.to_string(),
633 first_pack: first_pack.to_string(),
634 second_pack: pack.name().to_string(),
635 });
636 }
637 }
638 }
639 Ok(())
640}
641
642fn validate_unique_entity_types(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
652 let owned_defs = packs
653 .iter()
654 .flat_map(|p| p.entity_types().iter().map(move |def| (p.name(), def)));
655 khive_types::EntityTypeRegistry::check_extra_collisions(owned_defs)
656 .map_err(RuntimeError::InvalidInput)
657}
658
659fn find_pack_dependency_cycle(
660 packs: &[Box<dyn PackRuntime>],
661 name_to_idx: &HashMap<&str, usize>,
662 cycle_nodes: &HashSet<usize>,
663) -> Vec<String> {
664 fn visit(
665 idx: usize,
666 packs: &[Box<dyn PackRuntime>],
667 name_to_idx: &HashMap<&str, usize>,
668 cycle_nodes: &HashSet<usize>,
669 visiting: &mut Vec<usize>,
670 visited: &mut HashSet<usize>,
671 ) -> Option<Vec<String>> {
672 if let Some(pos) = visiting.iter().position(|&seen| seen == idx) {
673 let mut cycle: Vec<String> = visiting[pos..]
674 .iter()
675 .map(|&i| packs[i].name().to_string())
676 .collect();
677 cycle.push(packs[idx].name().to_string());
678 return Some(cycle);
679 }
680 if !visited.insert(idx) {
681 return None;
682 }
683 visiting.push(idx);
684 for &req in packs[idx].requires() {
685 let Some(&dep_idx) = name_to_idx.get(req) else {
686 continue;
687 };
688 if cycle_nodes.contains(&dep_idx) {
689 if let Some(cycle) =
690 visit(dep_idx, packs, name_to_idx, cycle_nodes, visiting, visited)
691 {
692 return Some(cycle);
693 }
694 }
695 }
696 visiting.pop();
697 None
698 }
699
700 let mut visited = HashSet::new();
701 for &idx in cycle_nodes {
702 let mut visiting = Vec::new();
703 if let Some(cycle) = visit(
704 idx,
705 packs,
706 name_to_idx,
707 cycle_nodes,
708 &mut visiting,
709 &mut visited,
710 ) {
711 return cycle;
712 }
713 }
714 cycle_nodes
715 .iter()
716 .map(|&idx| packs[idx].name().to_string())
717 .collect()
718}
719
720impl Default for VerbRegistryBuilder {
721 fn default() -> Self {
722 Self::new()
723 }
724}
725
726#[derive(Clone)]
730pub struct VerbRegistry {
731 packs: std::sync::Arc<Vec<Box<dyn PackRuntime>>>,
732 resolvers: std::sync::Arc<Vec<(String, Box<dyn PackByIdResolver>)>>,
734 gate: GateRef,
735 default_namespace: String,
736 visible_namespaces: Vec<Namespace>,
743 actor_id: Option<String>,
747 event_store: Option<Arc<dyn EventStore>>,
749 dispatch_hook: Option<Arc<dyn DispatchHook>>,
751 available_verbs: Arc<Vec<&'static str>>,
756 reference_ring: Arc<crate::reference_ring::ReferenceRing>,
761}
762
763#[derive(Debug, Clone, Default)]
778pub struct RequestIdentity {
779 pub namespace: String,
783 pub actor_id: Option<String>,
787 pub visible_namespaces: Vec<String>,
793 pub request_id: Option<u64>,
801}
802
803#[derive(Debug, Clone, PartialEq, Eq)]
812pub struct VerifiedActor(String);
813
814impl VerifiedActor {
815 pub fn new(id: impl Into<String>) -> Result<Self, RuntimeError> {
820 let id = id.into();
821 if id.trim().is_empty() {
822 return Err(RuntimeError::InvalidInput(
823 "VerifiedActor: identifier must not be empty or whitespace-only".to_string(),
824 ));
825 }
826 Ok(Self(id))
827 }
828
829 pub fn as_str(&self) -> &str {
831 &self.0
832 }
833
834 fn into_inner(self) -> String {
835 self.0
836 }
837}
838
839#[derive(Debug)]
842pub struct PackSchemaCollisionError {
843 pub pack_a: &'static str,
845 pub pack_b: &'static str,
847 pub table: String,
849}
850
851impl std::fmt::Display for PackSchemaCollisionError {
852 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
853 if self.pack_a == self.pack_b {
854 write!(
855 f,
856 "pack schema boot failure for pack {:?}: {}",
857 self.pack_a, self.table
858 )
859 } else {
860 write!(
861 f,
862 "pack schema collision: packs {:?} and {:?} both declare table {:?} \
863 on the same backend — move one pack to a separate backend or rename the table",
864 self.pack_a, self.pack_b, self.table
865 )
866 }
867 }
868}
869
870impl std::error::Error for PackSchemaCollisionError {}
871
872fn extract_table_names(stmt: &str) -> Vec<String> {
878 let normalized = stmt.split_whitespace().collect::<Vec<_>>().join(" ");
879 let upper = normalized.to_ascii_uppercase();
880 let table_name = if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE IF NOT EXISTS ") {
881 rest.split_whitespace().next()
882 } else if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE ") {
883 rest.split_whitespace().next()
884 } else if let Some(rest) = upper.strip_prefix("CREATE TABLE IF NOT EXISTS ") {
885 rest.split_whitespace().next()
886 } else if let Some(rest) = upper.strip_prefix("CREATE TABLE ") {
887 rest.split_whitespace().next()
888 } else {
889 None
890 };
891 match table_name {
892 Some(name) => {
893 let clean = name.trim_matches(|c: char| c == '(' || c == ';');
894 if clean.is_empty() {
895 vec![]
896 } else {
897 vec![clean.to_ascii_lowercase()]
898 }
899 }
900 None => vec![],
901 }
902}
903
904impl VerbRegistry {
905 pub fn default_namespace(&self) -> &str {
911 &self.default_namespace
912 }
913
914 pub fn actor_id(&self) -> Option<&str> {
918 self.actor_id.as_deref()
919 }
920
921 pub fn visible_namespaces(&self) -> &[Namespace] {
925 &self.visible_namespaces
926 }
927
928 pub fn event_store(&self) -> Option<Arc<dyn EventStore>> {
937 self.event_store.clone()
938 }
939
940 pub fn describe_verb(&self, verb: &str) -> Result<Value, RuntimeError> {
947 for pack in self.packs.iter() {
948 for handler in pack.handlers().iter() {
949 if handler.name == verb {
950 let category = format!("{:?}", handler.category);
951 let params_arr: Vec<Value> = handler
952 .params
953 .iter()
954 .map(|p| {
955 serde_json::json!({
956 "name": p.name,
957 "type": p.param_type,
958 "required": p.required,
959 "description": p.description,
960 })
961 })
962 .collect();
963 if matches!(handler.visibility, Visibility::Subhandler) {
968 return Ok(serde_json::json!({
969 "verb": verb,
970 "pack": pack.name(),
971 "description": handler.description,
972 "category": category,
973 "params": params_arr,
974 "visibility": "internal",
975 "callable_via_mcp": false,
976 "note": "This is an internal subhandler. Calling it via the MCP \
977 request surface returns permission denied. It can only be \
978 invoked by internal runtime callers.",
979 }));
980 }
981 return Ok(serde_json::json!({
982 "verb": verb,
983 "pack": pack.name(),
984 "description": handler.description,
985 "category": category,
986 "params": params_arr,
987 }));
988 }
989 }
990 }
991 Err(RuntimeError::InvalidInput(format!(
995 "unknown verb {verb:?}; available: {}",
996 self.available_verbs.join(", ")
997 )))
998 }
999
1000 pub fn authorize_namespace(&self, ns: Namespace) -> Result<(), RuntimeError> {
1008 let actor = crate::actor_identity::resolve_actor(self.actor_id.as_deref());
1009 let req = GateRequest::new(actor, ns, "authorize", serde_json::Value::Null);
1010 match self.gate.check(&req) {
1011 Ok(decision) if decision.is_allow() => Ok(()),
1012 Ok(GateDecision::Deny { reason }) => Err(RuntimeError::PermissionDenied {
1013 verb: "authorize".to_string(),
1014 reason,
1015 }),
1016 Ok(_) => Err(RuntimeError::PermissionDenied {
1017 verb: "authorize".to_string(),
1018 reason: "gate denied".to_string(),
1019 }),
1020 Err(e) => Err(RuntimeError::Internal(format!("gate error: {e}"))),
1021 }
1022 }
1023
1024 pub async fn dispatch(&self, verb: &str, params: Value) -> Result<Value, RuntimeError> {
1034 self.dispatch_with_identity(verb, params, None).await
1035 }
1036
1037 pub async fn dispatch_with_identity(
1049 &self,
1050 verb: &str,
1051 params: Value,
1052 identity: Option<RequestIdentity>,
1053 ) -> Result<Value, RuntimeError> {
1054 if params.get("help").and_then(Value::as_bool) == Some(true) {
1056 return self.describe_verb(verb);
1057 }
1058 let explicit_namespace = params.get("namespace").is_some_and(Value::is_string);
1069 let request_id: Option<u64> = identity.as_ref().and_then(|id| id.request_id);
1073 let default_namespace_str: &str = identity
1076 .as_ref()
1077 .map(|id| id.namespace.as_str())
1078 .unwrap_or(self.default_namespace.as_str());
1079 let ns = resolve_explicit_namespace(¶ms, default_namespace_str)?;
1080 let actor_id_str: Option<&str> = match identity.as_ref() {
1081 Some(id) => id.actor_id.as_deref(),
1082 None => self.actor_id.as_deref(),
1083 };
1084 let resolved_actor = crate::actor_identity::resolve_actor(actor_id_str);
1090 let gate_req = GateRequest::new(resolved_actor.clone(), ns.clone(), verb, params.clone());
1091
1092 let (gate_blocked, mut deferred_audit) = match self.gate.check(&gate_req) {
1098 Ok(decision) => {
1099 let is_deny = matches!(decision, GateDecision::Deny { .. });
1100
1101 let audit = AuditEvent::from_check(&gate_req, &decision, self.gate.impl_name());
1103 tracing::info!(
1104 audit_event = %serde_json::to_string(&audit)
1105 .unwrap_or_else(|_| "{\"error\":\"serialize\"}".into()),
1106 "gate.check"
1107 );
1108
1109 if let Some(store) = &self.event_store {
1118 if crate::config_ledger::PENDING
1119 .swap(false, std::sync::atomic::Ordering::AcqRel)
1120 {
1121 for (key, value) in crate::config_ledger::drain_config_locked() {
1122 let payload = serde_json::json!({ "key": key, "value": value });
1123 let storage_event = Event::new(
1124 gate_req.namespace.as_str(),
1125 verb,
1126 EventKind::ConfigLocked,
1127 SubstrateKind::Event,
1128 format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
1129 )
1130 .with_payload(payload);
1131 append_audit_event_best_effort(store, storage_event, verb).await;
1132 }
1133 }
1134 }
1135
1136 let defer_audit = !is_deny;
1151
1152 if !defer_audit {
1154 if let Some(store) = &self.event_store {
1155 let storage_event = build_audit_storage_event(
1161 &gate_req,
1162 &audit,
1163 EventOutcome::Denied,
1164 Some(crate::cost_unit::base_resource_payload(request_id)),
1165 );
1166 append_audit_event_best_effort(store, storage_event, verb).await;
1167 }
1168 }
1169
1170 let reason = if is_deny {
1171 let reason = match decision {
1172 GateDecision::Deny { reason } => reason,
1173 _ => String::new(),
1174 };
1175 Some(reason)
1176 } else {
1177 None
1178 };
1179 let deferred = if defer_audit { Some(audit) } else { None };
1180 (reason, deferred)
1181 }
1182 Err(err) => {
1183 tracing::warn!(verb, error = %err, "gate check failed (fail-open)");
1186 (None, None)
1187 }
1188 };
1189
1190 if let Some(reason) = gate_blocked {
1192 return Err(RuntimeError::PermissionDenied {
1193 verb: verb.to_string(),
1194 reason,
1195 });
1196 }
1197
1198 let token = if explicit_namespace {
1224 NamespaceToken::mint_with_visibility(ns.clone(), vec![], resolved_actor)
1226 } else {
1227 let primary = Namespace::local();
1229 let mut extra_visible: Vec<Namespace> = match identity.as_ref() {
1230 Some(id) => id
1231 .visible_namespaces
1232 .iter()
1233 .filter_map(|s| match Namespace::parse(s) {
1234 Ok(parsed) => Some(parsed),
1235 Err(e) => {
1236 tracing::warn!(
1237 namespace = %s,
1238 error = %e,
1239 "dispatch_with_identity: skipping invalid visible_namespace \
1240 entry from per-request identity"
1241 );
1242 None
1243 }
1244 })
1245 .collect(),
1246 None => self.visible_namespaces.clone(),
1247 };
1248 extra_visible.push(Namespace::local()); NamespaceToken::mint_with_visibility(primary, extra_visible, resolved_actor)
1250 };
1251
1252 for pack in self.packs.iter() {
1253 if let Some(handler_def) = pack.handlers().iter().find(|v| v.name == verb) {
1254 let handler_accepts_namespace =
1264 handler_def.params.iter().any(|p| p.name == "namespace");
1265 let params = if !handler_accepts_namespace {
1266 if let Value::Object(mut map) = params {
1267 map.remove("namespace");
1268 Value::Object(map)
1269 } else {
1270 params
1271 }
1272 } else {
1273 params
1274 };
1275 let dispatch_start = Instant::now();
1276 let result = pack.dispatch(verb, params, self, &token).await;
1277 let dispatch_us = dispatch_start.elapsed().as_micros() as i64;
1278
1279 if let Some(audit) = deferred_audit.take() {
1288 if let Some(store) = &self.event_store {
1289 let is_link_singleton =
1290 verb == "link" && gate_req.args.get("links").is_none();
1291 match &result {
1292 Ok(ok_val) if is_link_singleton => {
1293 let resource = crate::cost_unit::resource_payload(
1301 verb,
1302 &gate_req.args,
1303 ok_val,
1304 || pack.registered_embedding_model_names().len() as i64,
1305 request_id,
1306 );
1307 match link_audit_success_from_result(audit.clone(), ok_val) {
1308 Some((edge_id, mut payload)) => {
1309 if let Value::Object(ref mut map) = payload {
1310 map.insert("resource".to_string(), resource);
1311 }
1312 let storage_event = Event::new(
1313 gate_req.namespace.as_str(),
1314 gate_req.verb.as_str(),
1315 EventKind::Audit,
1316 SubstrateKind::Event,
1317 format!(
1318 "{}:{}",
1319 gate_req.actor.kind, gate_req.actor.id
1320 ),
1321 )
1322 .with_outcome(EventOutcome::Success)
1323 .with_target(edge_id)
1324 .with_payload(payload)
1325 .with_payload_schema_version(2)
1326 .with_duration_us(dispatch_us);
1327 append_audit_event_best_effort(store, storage_event, verb)
1328 .await;
1329 }
1330 None => {
1331 tracing::warn!(
1332 verb,
1333 "link audit v2 enrichment parse failed; \
1334 falling back to v1 audit shape"
1335 );
1336 let storage_event = build_audit_storage_event(
1337 &gate_req,
1338 &audit,
1339 EventOutcome::Success,
1340 Some(resource),
1341 )
1342 .with_duration_us(dispatch_us);
1343 append_audit_event_best_effort(store, storage_event, verb)
1344 .await;
1345 }
1346 }
1347 }
1348 _ => {
1349 let (outcome, resource) = match &result {
1369 Ok(ok_val) => (
1370 EventOutcome::Success,
1371 Some(crate::cost_unit::resource_payload(
1372 verb,
1373 &gate_req.args,
1374 ok_val,
1375 || pack.registered_embedding_model_names().len() as i64,
1376 request_id,
1377 )),
1378 ),
1379 Err(_) => (
1380 EventOutcome::Error,
1381 Some(crate::cost_unit::base_resource_payload(request_id)),
1382 ),
1383 };
1384 let storage_event =
1385 build_audit_storage_event(&gate_req, &audit, outcome, resource)
1386 .with_duration_us(dispatch_us);
1387 append_audit_event_best_effort(store, storage_event, verb).await;
1388 }
1389 }
1390 }
1391 }
1392
1393 if let (Ok(ref ok_val), Some(hook)) = (&result, &self.dispatch_hook) {
1395 let mut dispatch_event = Event::new(
1396 ns.as_str(),
1397 verb,
1398 EventKind::Audit,
1399 SubstrateKind::Event,
1400 pack.name(),
1401 )
1402 .with_outcome(EventOutcome::Success)
1403 .with_duration_us(dispatch_us);
1404
1405 if verb == "memory.recall" {
1409 let first_note_id = ok_val
1410 .as_array()
1411 .and_then(|arr| arr.first())
1412 .and_then(|v| v.get("id"))
1413 .and_then(|v| v.as_str())
1414 .and_then(|s| s.parse::<uuid::Uuid>().ok());
1415 if let Some(note_id) = first_note_id {
1416 dispatch_event = dispatch_event.with_target(note_id);
1417 }
1418 }
1421
1422 let dispatch_view = EventView {
1423 event: dispatch_event,
1424 observations: Vec::new(),
1425 };
1426 let hook = Arc::clone(hook);
1427 hook.on_dispatch(&dispatch_view).await;
1428 }
1429
1430 if let Ok(ref ok_val) = result {
1446 let admissions = crate::reference_ring::ring_admissions_for(verb, ok_val);
1447 if !admissions.is_empty() {
1448 let actor_key = format!("{}:{}", gate_req.actor.kind, gate_req.actor.id);
1449 for (id, name) in admissions {
1450 self.reference_ring.admit(
1451 token.namespace().as_str(),
1452 &actor_key,
1453 id,
1454 name,
1455 );
1456 }
1457 }
1458 }
1459
1460 return result;
1461 }
1462 }
1463
1464 if let Some(audit) = deferred_audit.take() {
1470 if let Some(store) = &self.event_store {
1471 let storage_event = build_audit_storage_event(
1477 &gate_req,
1478 &audit,
1479 EventOutcome::Error,
1480 Some(crate::cost_unit::base_resource_payload(request_id)),
1481 );
1482 append_audit_event_best_effort(store, storage_event, verb).await;
1483 }
1484 }
1485
1486 Err(RuntimeError::InvalidInput(format!(
1490 "unknown verb {verb:?}; available: {}",
1491 self.available_verbs.join(", ")
1492 )))
1493 }
1494
1495 pub async fn dispatch_as(
1513 &self,
1514 verb: &str,
1515 params: Value,
1516 verified_actor: VerifiedActor,
1517 ) -> Result<Value, RuntimeError> {
1518 let identity = RequestIdentity {
1519 namespace: self.default_namespace.clone(),
1520 actor_id: Some(verified_actor.into_inner()),
1521 visible_namespaces: self
1522 .visible_namespaces
1523 .iter()
1524 .map(|ns| ns.as_str().to_string())
1525 .collect(),
1526 request_id: None,
1527 };
1528 self.dispatch_with_identity(verb, params, Some(identity))
1529 .await
1530 }
1531
1532 pub fn resolvers(&self) -> &[(String, Box<dyn PackByIdResolver>)] {
1538 &self.resolvers
1539 }
1540
1541 pub fn reference_ring(&self) -> &Arc<crate::reference_ring::ReferenceRing> {
1546 &self.reference_ring
1547 }
1548
1549 pub fn find_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
1556 for pack in self.packs.iter() {
1557 let owns = pack.note_kinds().contains(&kind) || pack.entity_kinds().contains(&kind);
1558 if owns {
1559 if let Some(hook) = pack.kind_hook(kind) {
1560 return Some(hook);
1561 }
1562 }
1563 }
1564 None
1565 }
1566
1567 pub fn all_verbs(&self) -> Vec<&'static HandlerDef> {
1573 self.packs
1574 .iter()
1575 .flat_map(|p| p.handlers().iter())
1576 .filter(|h| matches!(h.visibility, Visibility::Verb))
1577 .collect()
1578 }
1579
1580 pub fn all_verbs_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
1587 self.packs
1588 .iter()
1589 .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
1590 .filter(|(_, h)| matches!(h.visibility, Visibility::Verb))
1591 .collect()
1592 }
1593
1594 pub fn all_handlers_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
1600 self.packs
1601 .iter()
1602 .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
1603 .collect()
1604 }
1605
1606 pub fn all_note_kinds(&self) -> Vec<&'static str> {
1609 let mut seen = std::collections::HashSet::new();
1610 self.packs
1611 .iter()
1612 .flat_map(|p| p.note_kinds().iter().copied())
1613 .filter(|k| seen.insert(*k))
1614 .collect()
1615 }
1616
1617 pub fn all_entity_kinds(&self) -> Vec<&'static str> {
1620 let mut seen = std::collections::HashSet::new();
1621 self.packs
1622 .iter()
1623 .flat_map(|p| p.entity_kinds().iter().copied())
1624 .filter(|k| seen.insert(*k))
1625 .collect()
1626 }
1627
1628 pub fn pack_names(&self) -> Vec<&str> {
1630 self.packs.iter().map(|p| p.name()).collect()
1631 }
1632
1633 pub fn pack_requires(&self, name: &str) -> Option<&'static [&'static str]> {
1635 self.packs
1636 .iter()
1637 .find(|p| p.name() == name)
1638 .map(|p| p.requires())
1639 }
1640
1641 pub fn pack_note_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
1646 self.packs
1647 .iter()
1648 .find(|p| p.name() == name)
1649 .map(|p| p.note_kinds())
1650 }
1651
1652 pub fn pack_entity_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
1657 self.packs
1658 .iter()
1659 .find(|p| p.name() == name)
1660 .map(|p| p.entity_kinds())
1661 }
1662
1663 pub fn pack_verbs(&self, name: &str) -> Option<&'static [HandlerDef]> {
1668 self.packs
1669 .iter()
1670 .find(|p| p.name() == name)
1671 .map(|p| p.handlers())
1672 }
1673
1674 pub fn all_edge_rules(&self) -> Vec<EdgeEndpointRule> {
1680 self.packs
1681 .iter()
1682 .flat_map(|p| p.edge_rules().iter().copied())
1683 .collect()
1684 }
1685
1686 pub fn all_entity_types(&self) -> Vec<EntityTypeDef> {
1693 self.packs
1694 .iter()
1695 .flat_map(|p| p.entity_types().iter().cloned())
1696 .collect()
1697 }
1698
1699 pub fn all_note_kind_specs(&self) -> Vec<&'static NoteKindSpec> {
1703 self.packs
1704 .iter()
1705 .flat_map(|p| p.note_kind_specs().iter())
1706 .collect()
1707 }
1708
1709 pub fn all_validation_rules(&self) -> Vec<&'static ValidationRule> {
1715 self.packs
1716 .iter()
1717 .flat_map(|p| p.validation_rules().iter())
1718 .collect()
1719 }
1720
1721 pub fn all_schema_plans(&self) -> Vec<SchemaPlan> {
1728 self.packs.iter().map(|p| p.schema_plan()).collect()
1729 }
1730
1731 pub fn call_register_embedders(&self, runtime: &KhiveRuntime) {
1741 for pack in self.packs.iter() {
1742 pack.register_embedders(runtime);
1743 }
1744 }
1745
1746 pub fn call_register_entity_type_validators(&self, runtime: &KhiveRuntime) {
1760 let entity_types = self.all_entity_types();
1761 for pack in self.packs.iter() {
1762 pack.register_entity_type_validator_with_types(runtime, &entity_types);
1763 }
1764 }
1765
1766 pub fn call_register_note_mutation_hooks(&self, runtime: &KhiveRuntime) {
1777 for pack in self.packs.iter() {
1778 pack.register_note_mutation_hook(runtime);
1779 }
1780 }
1781
1782 pub async fn call_warm_all(&self) {
1786 for pack in self.packs.iter() {
1787 pack.warm().await;
1788 }
1789 }
1790
1791 pub fn presentation_policy_for(&self, verb: &str) -> khive_types::VerbPresentationPolicy {
1798 for pack in self.packs.iter() {
1799 if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
1800 return handler.presentation_policy();
1801 }
1802 }
1803 khive_types::VerbPresentationPolicy::Standard
1804 }
1805
1806 pub fn is_subhandler_verb(&self, verb: &str) -> bool {
1813 for pack in self.packs.iter() {
1814 if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
1815 return matches!(handler.visibility, Visibility::Subhandler);
1816 }
1817 }
1818 false
1819 }
1820
1821 pub fn apply_schema_plans(&self, backend: &khive_db::StorageBackend) {
1833 for plan in self.all_schema_plans() {
1834 if plan.is_empty() {
1835 continue;
1836 }
1837 if let Err(e) = backend.apply_pack_ddl_statements(plan.statements) {
1838 tracing::warn!(
1839 pack = plan.pack,
1840 error = %e,
1841 "failed to apply pack schema plan at startup (non-fatal)"
1842 );
1843 }
1844 }
1845 }
1846
1847 pub fn all_schema_plans_named(&self) -> Vec<(&'static str, SchemaPlan)> {
1853 self.packs
1854 .iter()
1855 .map(|p| {
1856 let plan = p.schema_plan();
1857 (plan.pack, plan)
1858 })
1859 .collect()
1860 }
1861
1862 pub fn apply_schema_plans_with_map(
1875 &self,
1876 backend_for_pack: &HashMap<&str, &khive_db::StorageBackend>,
1877 default_backend: &khive_db::StorageBackend,
1878 ) -> Result<(), crate::PackSchemaCollisionError> {
1879 let mut claimed: HashMap<(*const (), String), &'static str> = HashMap::new();
1882
1883 for (pack_name, plan) in self.all_schema_plans_named() {
1884 if plan.is_empty() {
1885 continue;
1886 }
1887 let backend = backend_for_pack
1888 .get(pack_name)
1889 .copied()
1890 .unwrap_or(default_backend);
1891 let backend_ptr = std::sync::Arc::as_ptr(&backend.pool_arc()) as *const ();
1892
1893 for stmt in plan.statements {
1895 for table_name in extract_table_names(stmt) {
1896 let key = (backend_ptr, table_name.clone());
1897 match claimed.entry(key) {
1898 std::collections::hash_map::Entry::Vacant(e) => {
1899 e.insert(pack_name);
1900 }
1901 std::collections::hash_map::Entry::Occupied(e) => {
1902 let prior_pack = *e.get();
1903 return Err(crate::PackSchemaCollisionError {
1904 pack_a: prior_pack,
1905 pack_b: pack_name,
1906 table: table_name,
1907 });
1908 }
1909 }
1910 }
1911 }
1912
1913 backend
1914 .apply_pack_ddl_statements(plan.statements)
1915 .map_err(|e| crate::PackSchemaCollisionError {
1916 pack_a: pack_name,
1917 pack_b: pack_name,
1918 table: format!("DDL error: {e}"),
1919 })?;
1920 }
1921 Ok(())
1922 }
1923}
1924
1925pub struct PackInstall {
1933 pub runtime: Box<dyn PackRuntime>,
1935 pub resolver: Option<Box<dyn PackByIdResolver>>,
1937 pub dispatch_hook: Option<Arc<dyn DispatchHook>>,
1939}
1940
1941pub trait PackFactory: Send + Sync + 'static {
1949 fn name(&self) -> &'static str;
1951
1952 fn requires(&self) -> &'static [&'static str] {
1959 &[]
1960 }
1961
1962 fn create(&self, runtime: KhiveRuntime) -> Box<dyn PackRuntime>;
1964
1965 fn create_install(&self, runtime: KhiveRuntime) -> PackInstall {
1974 let resolver = self.create_resolver(runtime.clone());
1975 PackInstall {
1976 runtime: self.create(runtime),
1977 resolver,
1978 dispatch_hook: None,
1979 }
1980 }
1981
1982 fn create_resolver(&self, _runtime: KhiveRuntime) -> Option<Box<dyn PackByIdResolver>> {
1988 None
1989 }
1990}
1991
1992pub struct PackRegistration(pub &'static dyn PackFactory);
1996
1997inventory::collect!(PackRegistration);
1998
1999#[derive(Debug)]
2001pub enum PackLoadError {
2002 UnknownPack(String),
2004 MissingDependency {
2006 pack: String,
2008 dep: String,
2010 },
2011}
2012
2013impl std::fmt::Display for PackLoadError {
2014 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2015 match self {
2016 PackLoadError::UnknownPack(name) => write!(f, "unknown pack {name:?}"),
2017 PackLoadError::MissingDependency { pack, dep } => write!(
2018 f,
2019 "pack {pack:?} requires {dep:?}, which is not in the requested pack list; \
2020 add --pack {dep} before --pack {pack}"
2021 ),
2022 }
2023 }
2024}
2025
2026impl std::error::Error for PackLoadError {}
2027
2028pub struct PackRegistry;
2033
2034impl PackRegistry {
2035 pub fn discovered_names() -> Vec<&'static str> {
2037 inventory::iter::<PackRegistration>
2038 .into_iter()
2039 .map(|r| r.0.name())
2040 .collect()
2041 }
2042
2043 pub fn register_packs(
2056 names: &[String],
2057 runtime: KhiveRuntime,
2058 builder: &mut VerbRegistryBuilder,
2059 ) -> Result<(), PackLoadError> {
2060 let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
2062 .into_iter()
2063 .map(|r| r.0)
2064 .collect();
2065 let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
2066 all.iter().copied().find(|f| f.name() == name)
2067 };
2068
2069 let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
2071 for name in names {
2072 factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
2073 }
2074
2075 for name in names {
2078 let factory = factory_for(name.as_str()).unwrap(); for &dep in factory.requires() {
2080 if !requested.contains(dep) {
2081 return Err(PackLoadError::MissingDependency {
2082 pack: name.clone(),
2083 dep: dep.to_string(),
2084 });
2085 }
2086 }
2087 }
2088
2089 for name in names {
2092 let factory = factory_for(name.as_str()).unwrap(); let install = factory.create_install(runtime.clone());
2094 builder.register_boxed(install.runtime);
2095 if let Some(resolver) = install.resolver {
2096 builder.register_resolver(name.clone(), resolver);
2097 }
2098 if let Some(hook) = install.dispatch_hook {
2099 builder.with_dispatch_hook(hook);
2100 }
2101 }
2102
2103 Ok(())
2104 }
2105
2106 pub fn register_packs_with_runtimes(
2116 names: &[String],
2117 runtimes: &HashMap<String, KhiveRuntime>,
2118 default_runtime: &KhiveRuntime,
2119 builder: &mut VerbRegistryBuilder,
2120 ) -> Result<(), PackLoadError> {
2121 let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
2122 .into_iter()
2123 .map(|r| r.0)
2124 .collect();
2125 let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
2126 all.iter().copied().find(|f| f.name() == name)
2127 };
2128
2129 let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
2130 for name in names {
2131 factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
2132 }
2133
2134 for name in names {
2135 let factory = factory_for(name.as_str()).unwrap();
2136 for &dep in factory.requires() {
2137 if !requested.contains(dep) {
2138 return Err(PackLoadError::MissingDependency {
2139 pack: name.clone(),
2140 dep: dep.to_string(),
2141 });
2142 }
2143 }
2144 }
2145
2146 for name in names {
2147 let factory = factory_for(name.as_str()).unwrap();
2148 let runtime = runtimes
2149 .get(name.as_str())
2150 .cloned()
2151 .unwrap_or_else(|| default_runtime.clone());
2152 let install = factory.create_install(runtime);
2153 builder.register_boxed(install.runtime);
2154 if let Some(resolver) = install.resolver {
2155 builder.register_resolver(name.clone(), resolver);
2156 }
2157 if let Some(hook) = install.dispatch_hook {
2158 builder.with_dispatch_hook(hook);
2159 }
2160 }
2161
2162 Ok(())
2163 }
2164}
2165
2166fn target_id_from_args(args: &serde_json::Value) -> Option<uuid::Uuid> {
2167 args.get("target_id")
2168 .and_then(serde_json::Value::as_str)
2169 .and_then(|s| s.parse::<uuid::Uuid>().ok())
2170}
2171
2172fn build_audit_storage_event(
2175 gate_req: &GateRequest,
2176 audit: &AuditEvent,
2177 outcome: EventOutcome,
2178 resource: Option<Value>,
2179) -> Event {
2180 let mut audit_data = serde_json::to_value(audit).unwrap_or_else(|e| {
2181 tracing::warn!(error = %e, "failed to serialize AuditEvent for EventStore");
2182 serde_json::Value::Null
2183 });
2184 if let Some(resource) = resource {
2185 if let Value::Object(ref mut map) = audit_data {
2186 map.insert("resource".to_string(), resource);
2187 }
2188 }
2189 let mut storage_event = Event::new(
2190 gate_req.namespace.as_str(),
2191 gate_req.verb.as_str(),
2192 EventKind::Audit,
2193 SubstrateKind::Event,
2194 format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
2195 )
2196 .with_outcome(outcome)
2197 .with_payload(audit_data);
2198 if let Some(target_id) = target_id_from_args(&gate_req.args) {
2199 storage_event = storage_event.with_target(target_id);
2200 }
2201 storage_event
2202}
2203
2204async fn append_audit_event_best_effort(store: &Arc<dyn EventStore>, event: Event, verb: &str) {
2209 if let Err(store_err) = store.append_event(event).await {
2210 tracing::warn!(
2211 verb,
2212 error = %store_err,
2213 "audit event store write failed (non-fatal)"
2214 );
2215 }
2216}
2217
2218#[derive(Debug, Clone, serde::Serialize)]
2221struct LinkAuditSuccessV2 {
2222 #[serde(flatten)]
2223 audit: AuditEvent,
2224 edge_id: uuid::Uuid,
2225 source_id: uuid::Uuid,
2226 target_id: uuid::Uuid,
2227 relation: String,
2228 weight: f64,
2229}
2230
2231fn link_audit_success_from_result(
2235 audit: AuditEvent,
2236 result: &serde_json::Value,
2237) -> Option<(uuid::Uuid, serde_json::Value)> {
2238 let edge_id = result.get("id")?.as_str()?.parse::<uuid::Uuid>().ok()?;
2239 let source_id = result
2240 .get("source_id")?
2241 .as_str()?
2242 .parse::<uuid::Uuid>()
2243 .ok()?;
2244 let target_id = result
2245 .get("target_id")?
2246 .as_str()?
2247 .parse::<uuid::Uuid>()
2248 .ok()?;
2249 let relation = result.get("relation")?.as_str()?.to_string();
2250 let weight = result.get("weight")?.as_f64()?;
2251 let enriched = LinkAuditSuccessV2 {
2252 audit,
2253 edge_id,
2254 source_id,
2255 target_id,
2256 relation,
2257 weight,
2258 };
2259 let payload = serde_json::to_value(&enriched).ok()?;
2260 Some((edge_id, payload))
2261}
2262
2263pub fn resolve_explicit_namespace(
2276 params: &Value,
2277 default_namespace: &str,
2278) -> Result<Namespace, RuntimeError> {
2279 match params.get("namespace") {
2280 None => Namespace::parse(default_namespace)
2281 .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}"))),
2282 Some(Value::String(ns_str)) => Namespace::parse(ns_str)
2283 .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace {ns_str:?}: {e}"))),
2284 Some(other) => Err(RuntimeError::InvalidInput(format!(
2285 "invalid namespace: expected string when present, got {}",
2286 json_type_name(other),
2287 ))),
2288 }
2289}
2290
2291pub fn json_type_name(v: &Value) -> &'static str {
2294 match v {
2295 Value::Null => "null",
2296 Value::Bool(_) => "boolean",
2297 Value::Number(_) => "number",
2298 Value::String(_) => "string",
2299 Value::Array(_) => "array",
2300 Value::Object(_) => "object",
2301 }
2302}
2303
2304#[cfg(test)]
2310mod tests {
2311 use super::*;
2312 use crate::ActorRef;
2313 use khive_types::Pack;
2314
2315 struct AlphaPack;
2316
2317 impl Pack for AlphaPack {
2318 const NAME: &'static str = "alpha";
2319 const NOTE_KINDS: &'static [&'static str] = &["memo", "log"];
2320 const ENTITY_KINDS: &'static [&'static str] = &["widget"];
2321 const HANDLERS: &'static [HandlerDef] = &[
2322 HandlerDef {
2323 name: "create",
2324 description: "create a widget",
2325 visibility: Visibility::Verb,
2326 category: VerbCategory::Commissive,
2327 params: &[],
2328 },
2329 HandlerDef {
2330 name: "list",
2331 description: "list widgets",
2332 visibility: Visibility::Verb,
2333 category: VerbCategory::Assertive,
2334 params: &[],
2335 },
2336 ];
2337 }
2338
2339 #[async_trait]
2340 impl PackRuntime for AlphaPack {
2341 fn name(&self) -> &str {
2342 AlphaPack::NAME
2343 }
2344 fn note_kinds(&self) -> &'static [&'static str] {
2345 AlphaPack::NOTE_KINDS
2346 }
2347 fn entity_kinds(&self) -> &'static [&'static str] {
2348 AlphaPack::ENTITY_KINDS
2349 }
2350 fn handlers(&self) -> &'static [HandlerDef] {
2351 AlphaPack::HANDLERS
2352 }
2353 async fn dispatch(
2354 &self,
2355 verb: &str,
2356 _params: Value,
2357 _registry: &VerbRegistry,
2358 _token: &NamespaceToken,
2359 ) -> Result<Value, RuntimeError> {
2360 Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
2361 }
2362 }
2363
2364 struct SleepingPack;
2368
2369 impl Pack for SleepingPack {
2370 const NAME: &'static str = "sleeping";
2371 const NOTE_KINDS: &'static [&'static str] = &[];
2372 const ENTITY_KINDS: &'static [&'static str] = &[];
2373 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2374 name: "slow_op",
2375 description: "sleeps before returning",
2376 visibility: Visibility::Verb,
2377 category: VerbCategory::Assertive,
2378 params: &[],
2379 }];
2380 }
2381
2382 #[async_trait]
2383 impl PackRuntime for SleepingPack {
2384 fn name(&self) -> &str {
2385 SleepingPack::NAME
2386 }
2387 fn note_kinds(&self) -> &'static [&'static str] {
2388 SleepingPack::NOTE_KINDS
2389 }
2390 fn entity_kinds(&self) -> &'static [&'static str] {
2391 SleepingPack::ENTITY_KINDS
2392 }
2393 fn handlers(&self) -> &'static [HandlerDef] {
2394 SleepingPack::HANDLERS
2395 }
2396 async fn dispatch(
2397 &self,
2398 verb: &str,
2399 _params: Value,
2400 _registry: &VerbRegistry,
2401 _token: &NamespaceToken,
2402 ) -> Result<Value, RuntimeError> {
2403 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2404 Ok(serde_json::json!({ "pack": "sleeping", "verb": verb }))
2405 }
2406 }
2407
2408 struct BetaPack;
2409
2410 impl Pack for BetaPack {
2411 const NAME: &'static str = "beta";
2412 const NOTE_KINDS: &'static [&'static str] = &["alert"];
2413 const ENTITY_KINDS: &'static [&'static str] = &["widget", "gadget"];
2414 const HANDLERS: &'static [HandlerDef] = &[
2415 HandlerDef {
2416 name: "notify",
2417 description: "send alert",
2418 visibility: Visibility::Verb,
2419 category: VerbCategory::Commissive,
2420 params: &[],
2421 },
2422 HandlerDef {
2426 name: "create",
2427 description: "beta internal create (subhandler)",
2428 visibility: Visibility::Subhandler,
2429 category: VerbCategory::Commissive,
2430 params: &[],
2431 },
2432 ];
2433 }
2434
2435 fn build_registry() -> VerbRegistry {
2441 let mut builder = VerbRegistryBuilder::new();
2442 builder.register(AlphaPack);
2443 builder.register(BetaPack);
2444 builder.build().expect("registry builds without collision")
2445 }
2446
2447 struct CollidingPack;
2450
2451 impl Pack for CollidingPack {
2452 const NAME: &'static str = "colliding";
2453 const NOTE_KINDS: &'static [&'static str] = &[];
2454 const ENTITY_KINDS: &'static [&'static str] = &[];
2455 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2456 name: "create",
2457 description: "duplicate Verb-visibility create",
2458 visibility: Visibility::Verb,
2459 category: VerbCategory::Commissive,
2460 params: &[],
2461 }];
2462 }
2463
2464 #[async_trait]
2465 impl PackRuntime for CollidingPack {
2466 fn name(&self) -> &str {
2467 Self::NAME
2468 }
2469 fn note_kinds(&self) -> &'static [&'static str] {
2470 Self::NOTE_KINDS
2471 }
2472 fn entity_kinds(&self) -> &'static [&'static str] {
2473 Self::ENTITY_KINDS
2474 }
2475 fn handlers(&self) -> &'static [HandlerDef] {
2476 Self::HANDLERS
2477 }
2478 async fn dispatch(
2479 &self,
2480 verb: &str,
2481 _params: Value,
2482 _registry: &VerbRegistry,
2483 _token: &NamespaceToken,
2484 ) -> Result<Value, RuntimeError> {
2485 Ok(serde_json::json!({ "pack": "colliding", "verb": verb }))
2486 }
2487 }
2488
2489 #[async_trait]
2490 impl PackRuntime for BetaPack {
2491 fn name(&self) -> &str {
2492 BetaPack::NAME
2493 }
2494 fn note_kinds(&self) -> &'static [&'static str] {
2495 BetaPack::NOTE_KINDS
2496 }
2497 fn entity_kinds(&self) -> &'static [&'static str] {
2498 BetaPack::ENTITY_KINDS
2499 }
2500 fn handlers(&self) -> &'static [HandlerDef] {
2501 BetaPack::HANDLERS
2502 }
2503 async fn dispatch(
2504 &self,
2505 verb: &str,
2506 _params: Value,
2507 _registry: &VerbRegistry,
2508 _token: &NamespaceToken,
2509 ) -> Result<Value, RuntimeError> {
2510 Ok(serde_json::json!({ "pack": "beta", "verb": verb }))
2511 }
2512 }
2513
2514 #[tokio::test]
2515 async fn dispatch_routes_to_correct_pack() {
2516 let reg = build_registry();
2517
2518 let res = reg.dispatch("list", Value::Null).await.unwrap();
2519 assert_eq!(res["pack"], "alpha");
2520
2521 let res = reg.dispatch("notify", Value::Null).await.unwrap();
2522 assert_eq!(res["pack"], "beta");
2523 }
2524
2525 #[test]
2529 fn verb_collision_is_boot_time_error() {
2530 let mut builder = VerbRegistryBuilder::new();
2531 builder.register(AlphaPack);
2532 builder.register(CollidingPack);
2533 let err = builder
2534 .build()
2535 .err()
2536 .expect("duplicate Verb-visibility handler must be rejected at build time");
2537 assert!(
2538 matches!(err, RuntimeError::VerbCollision { ref verb, .. } if verb == "create"),
2539 "expected VerbCollision for 'create', got {err:?}"
2540 );
2541 let msg = err.to_string();
2542 assert!(
2543 msg.contains("create"),
2544 "error must name the colliding verb: {msg}"
2545 );
2546 assert!(
2547 msg.contains("alpha") || msg.contains("colliding"),
2548 "error must name one of the conflicting packs: {msg}"
2549 );
2550 }
2551
2552 #[test]
2556 fn subhandler_same_name_across_packs_is_not_a_collision() {
2557 struct SubhandlerPack;
2558 impl Pack for SubhandlerPack {
2559 const NAME: &'static str = "subhandler_pack";
2560 const NOTE_KINDS: &'static [&'static str] = &[];
2561 const ENTITY_KINDS: &'static [&'static str] = &[];
2562 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2563 name: "create",
2564 description: "internal create",
2565 visibility: Visibility::Subhandler,
2566 category: VerbCategory::Commissive,
2567 params: &[],
2568 }];
2569 }
2570 #[async_trait]
2571 impl PackRuntime for SubhandlerPack {
2572 fn name(&self) -> &str {
2573 Self::NAME
2574 }
2575 fn note_kinds(&self) -> &'static [&'static str] {
2576 Self::NOTE_KINDS
2577 }
2578 fn entity_kinds(&self) -> &'static [&'static str] {
2579 Self::ENTITY_KINDS
2580 }
2581 fn handlers(&self) -> &'static [HandlerDef] {
2582 Self::HANDLERS
2583 }
2584 async fn dispatch(
2585 &self,
2586 verb: &str,
2587 _: Value,
2588 _: &VerbRegistry,
2589 _: &NamespaceToken,
2590 ) -> Result<Value, RuntimeError> {
2591 Ok(serde_json::json!({"pack": "subhandler_pack", "verb": verb}))
2592 }
2593 }
2594 let mut builder = VerbRegistryBuilder::new();
2595 builder.register(AlphaPack); builder.register(SubhandlerPack); builder
2598 .build()
2599 .expect("subhandler same name must NOT be a collision");
2600 }
2601
2602 #[tokio::test]
2603 async fn dispatch_unknown_verb_returns_error() {
2604 let reg = build_registry();
2605
2606 let err = reg.dispatch("explode", Value::Null).await.unwrap_err();
2607 let msg = err.to_string();
2608 assert!(msg.contains("explode"));
2609 assert!(msg.contains("create"));
2610 }
2611
2612 #[test]
2617 fn all_verbs_aggregates_across_packs_excludes_subhandlers() {
2618 let reg = build_registry();
2619 let verbs: Vec<&str> = reg.all_verbs().iter().map(|v| v.name).collect();
2620 assert_eq!(verbs, vec!["create", "list", "notify"]);
2622 }
2623
2624 #[test]
2625 fn all_verbs_with_names_pairs_pack_name_excludes_subhandlers() {
2626 let reg = build_registry();
2627 let pairs: Vec<(&str, &str)> = reg
2628 .all_verbs_with_names()
2629 .iter()
2630 .map(|(pack, v)| (*pack, v.name))
2631 .collect();
2632 assert_eq!(
2634 pairs,
2635 vec![("alpha", "create"), ("alpha", "list"), ("beta", "notify"),]
2636 );
2637 }
2638
2639 #[test]
2640 fn all_handlers_with_names_includes_subhandlers() {
2641 let reg = build_registry();
2642 let pairs: Vec<(&str, &str)> = reg
2643 .all_handlers_with_names()
2644 .iter()
2645 .map(|(pack, v)| (*pack, v.name))
2646 .collect();
2647 assert_eq!(
2649 pairs,
2650 vec![
2651 ("alpha", "create"),
2652 ("alpha", "list"),
2653 ("beta", "notify"),
2654 ("beta", "create"),
2655 ]
2656 );
2657 }
2658
2659 #[test]
2660 fn note_kinds_are_ordered() {
2661 let reg = build_registry();
2662 let kinds = reg.all_note_kinds();
2663 assert_eq!(kinds, vec!["memo", "log", "alert"]);
2664 }
2665
2666 #[test]
2667 fn note_kind_duplicate_rejected_at_build_time() {
2668 struct DupPack;
2669
2670 impl khive_types::Pack for DupPack {
2671 const NAME: &'static str = "dup";
2672 const NOTE_KINDS: &'static [&'static str] = &["memo"];
2674 const ENTITY_KINDS: &'static [&'static str] = &[];
2675 const HANDLERS: &'static [HandlerDef] = &[];
2676 }
2677
2678 #[async_trait]
2679 impl PackRuntime for DupPack {
2680 fn name(&self) -> &str {
2681 Self::NAME
2682 }
2683 fn note_kinds(&self) -> &'static [&'static str] {
2684 Self::NOTE_KINDS
2685 }
2686 fn entity_kinds(&self) -> &'static [&'static str] {
2687 Self::ENTITY_KINDS
2688 }
2689 fn handlers(&self) -> &'static [HandlerDef] {
2690 Self::HANDLERS
2691 }
2692 async fn dispatch(
2693 &self,
2694 _verb: &str,
2695 _params: Value,
2696 _registry: &VerbRegistry,
2697 _token: &NamespaceToken,
2698 ) -> Result<Value, RuntimeError> {
2699 Ok(Value::Null)
2700 }
2701 }
2702
2703 let mut builder = VerbRegistryBuilder::new();
2704 builder.register(AlphaPack);
2705 builder.register(DupPack);
2706 let err = builder
2707 .build()
2708 .err()
2709 .expect("duplicate note kind must be rejected");
2710 let msg = err.to_string();
2711 assert!(
2712 msg.contains("memo"),
2713 "error must name the duplicate kind: {msg}"
2714 );
2715 assert!(
2716 msg.contains("alpha") || msg.contains("dup"),
2717 "error must name one of the conflicting packs: {msg}"
2718 );
2719 }
2720
2721 #[test]
2722 fn entity_kinds_are_deduplicated() {
2723 let reg = build_registry();
2724 let kinds = reg.all_entity_kinds();
2725 assert_eq!(kinds, vec!["widget", "gadget"]);
2726 }
2727
2728 struct GammaPack;
2731
2732 impl Pack for GammaPack {
2733 const NAME: &'static str = "gamma";
2734 const NOTE_KINDS: &'static [&'static str] = &[];
2735 const ENTITY_KINDS: &'static [&'static str] = &[];
2736 const HANDLERS: &'static [HandlerDef] = &[];
2737 const ENTITY_TYPES: &'static [EntityTypeDef] = &[EntityTypeDef {
2738 kind: khive_types::EntityKind::Document,
2739 type_name: "gamma_report",
2740 aliases: &["gamma_rep"],
2741 }];
2742 }
2743
2744 #[async_trait]
2745 impl PackRuntime for GammaPack {
2746 fn name(&self) -> &str {
2747 Self::NAME
2748 }
2749 fn note_kinds(&self) -> &'static [&'static str] {
2750 Self::NOTE_KINDS
2751 }
2752 fn entity_kinds(&self) -> &'static [&'static str] {
2753 Self::ENTITY_KINDS
2754 }
2755 fn handlers(&self) -> &'static [HandlerDef] {
2756 Self::HANDLERS
2757 }
2758 fn entity_types(&self) -> &'static [EntityTypeDef] {
2759 Self::ENTITY_TYPES
2760 }
2761 async fn dispatch(
2762 &self,
2763 verb: &str,
2764 _params: Value,
2765 _registry: &VerbRegistry,
2766 _token: &NamespaceToken,
2767 ) -> Result<Value, RuntimeError> {
2768 Ok(serde_json::json!({ "pack": "gamma", "verb": verb }))
2769 }
2770 }
2771
2772 #[test]
2776 fn all_entity_types_empty_when_no_pack_declares_extras() {
2777 let reg = build_registry(); assert!(reg.all_entity_types().is_empty());
2779 let composed = khive_types::EntityTypeRegistry::with_extra(reg.all_entity_types());
2780 let resolved = composed
2781 .resolve(khive_types::EntityKind::Document, Some("paper"))
2782 .expect("builtin paper subtype must still resolve");
2783 assert_eq!(resolved.entity_type.as_deref(), Some("paper"));
2784 }
2785
2786 #[test]
2789 fn pack_declared_entity_type_validates_through_composed_registry() {
2790 let mut builder = VerbRegistryBuilder::new();
2791 builder.register(AlphaPack);
2792 builder.register(GammaPack);
2793 let reg = builder.build().expect("registry builds");
2794
2795 let extras = reg.all_entity_types();
2796 assert_eq!(extras.len(), 1);
2797
2798 let composed = khive_types::EntityTypeRegistry::with_extra(extras);
2799 let resolved = composed
2800 .resolve(khive_types::EntityKind::Document, Some("gamma_rep"))
2801 .expect("pack-declared alias must resolve through the composed registry");
2802 assert_eq!(resolved.entity_type.as_deref(), Some("gamma_report"));
2803
2804 let builtin_resolved = composed
2805 .resolve(khive_types::EntityKind::Document, Some("paper"))
2806 .expect("builtin subtype must remain resolvable when a pack adds extras");
2807 assert_eq!(builtin_resolved.entity_type.as_deref(), Some("paper"));
2808
2809 composed
2810 .resolve(khive_types::EntityKind::Document, Some("nonexistent_type"))
2811 .expect_err("undeclared entity_type must still be rejected");
2812 }
2813
2814 #[test]
2820 fn overlapping_pack_declared_entity_types_reject_at_boot() {
2821 struct DeltaPack;
2822 impl Pack for DeltaPack {
2823 const NAME: &'static str = "delta";
2824 const NOTE_KINDS: &'static [&'static str] = &[];
2825 const ENTITY_KINDS: &'static [&'static str] = &[];
2826 const HANDLERS: &'static [HandlerDef] = &[];
2827 const ENTITY_TYPES: &'static [EntityTypeDef] = &[EntityTypeDef {
2828 kind: khive_types::EntityKind::Document,
2829 type_name: "gamma_report",
2830 aliases: &["gamma_rep"],
2831 }];
2832 }
2833 #[async_trait]
2834 impl PackRuntime for DeltaPack {
2835 fn name(&self) -> &str {
2836 Self::NAME
2837 }
2838 fn note_kinds(&self) -> &'static [&'static str] {
2839 Self::NOTE_KINDS
2840 }
2841 fn entity_kinds(&self) -> &'static [&'static str] {
2842 Self::ENTITY_KINDS
2843 }
2844 fn handlers(&self) -> &'static [HandlerDef] {
2845 Self::HANDLERS
2846 }
2847 fn entity_types(&self) -> &'static [EntityTypeDef] {
2848 Self::ENTITY_TYPES
2849 }
2850 async fn dispatch(
2851 &self,
2852 verb: &str,
2853 _params: Value,
2854 _registry: &VerbRegistry,
2855 _token: &NamespaceToken,
2856 ) -> Result<Value, RuntimeError> {
2857 Ok(serde_json::json!({ "pack": "delta", "verb": verb }))
2858 }
2859 }
2860
2861 let mut builder = VerbRegistryBuilder::new();
2862 builder.register(GammaPack);
2863 builder.register(DeltaPack);
2864 let err = builder.build().err().expect(
2865 "overlapping ENTITY_TYPES declarations must fail at build, not silently compose",
2866 );
2867
2868 let msg = err.to_string();
2869 assert!(
2870 msg.contains("gamma") && msg.contains("delta"),
2871 "collision error must name both contributing packs: {msg}"
2872 );
2873 assert!(
2874 msg.contains("gamma_report"),
2875 "collision error must name the colliding entity_type key: {msg}"
2876 );
2877 }
2878
2879 use khive_gate::{Gate, GateError};
2882 use std::sync::atomic::{AtomicUsize, Ordering};
2883 use std::sync::Arc;
2884
2885 #[derive(Default, Debug)]
2886 struct CountingGate {
2887 calls: AtomicUsize,
2888 deny_verb: Option<&'static str>,
2889 }
2890
2891 impl Gate for CountingGate {
2892 fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
2893 self.calls.fetch_add(1, Ordering::SeqCst);
2894 if Some(req.verb.as_str()) == self.deny_verb {
2895 Ok(GateDecision::deny(format!("test deny for {}", req.verb)))
2896 } else {
2897 Ok(GateDecision::allow())
2898 }
2899 }
2900 }
2901
2902 #[tokio::test]
2903 async fn dispatch_consults_the_gate() {
2904 let gate = Arc::new(CountingGate::default());
2905 let mut builder = VerbRegistryBuilder::new();
2906 builder.register(AlphaPack);
2907 builder.with_gate(gate.clone());
2908 let reg = builder.build().expect("registry builds");
2909
2910 reg.dispatch("list", Value::Null).await.unwrap();
2911 reg.dispatch("create", Value::Null).await.unwrap();
2912 assert_eq!(
2913 gate.calls.load(Ordering::SeqCst),
2914 2,
2915 "gate should be consulted once per dispatch"
2916 );
2917 }
2918
2919 #[tokio::test]
2920 async fn dispatch_returns_permission_denied_on_deny_v03() {
2921 let gate = Arc::new(CountingGate {
2922 calls: AtomicUsize::new(0),
2923 deny_verb: Some("create"),
2924 });
2925 let mut builder = VerbRegistryBuilder::new();
2926 builder.register(AlphaPack);
2927 builder.with_gate(gate.clone());
2928 let reg = builder.build().expect("registry builds");
2929
2930 let err = reg.dispatch("create", Value::Null).await.unwrap_err();
2932 assert!(
2933 matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
2934 "expected PermissionDenied, got {err:?}"
2935 );
2936 let msg = err.to_string();
2937 assert!(
2938 msg.contains("create"),
2939 "error message must name the verb: {msg}"
2940 );
2941 assert!(
2942 msg.contains("test deny for create"),
2943 "error message must carry the deny reason: {msg}"
2944 );
2945 assert_eq!(gate.calls.load(Ordering::SeqCst), 1);
2946 }
2947
2948 #[tokio::test]
2949 async fn dispatch_allow_verb_succeeds_even_with_deny_gate_for_other_verb() {
2950 let gate = Arc::new(CountingGate {
2952 calls: AtomicUsize::new(0),
2953 deny_verb: Some("create"),
2954 });
2955 let mut builder = VerbRegistryBuilder::new();
2956 builder.register(AlphaPack);
2957 builder.with_gate(gate.clone());
2958 let reg = builder.build().expect("registry builds");
2959
2960 let res = reg.dispatch("list", Value::Null).await.unwrap();
2961 assert_eq!(res["pack"], "alpha");
2962 }
2963
2964 #[tokio::test]
2965 async fn dispatch_uses_allow_all_gate_by_default() {
2966 let reg = build_registry();
2968 let res = reg.dispatch("list", Value::Null).await.unwrap();
2969 assert_eq!(res["pack"], "alpha");
2970 }
2971
2972 #[derive(Default, Debug)]
2975 struct NamespaceCapturingGate {
2976 seen: std::sync::Mutex<Vec<String>>,
2977 }
2978
2979 impl Gate for NamespaceCapturingGate {
2980 fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
2981 self.seen
2982 .lock()
2983 .unwrap()
2984 .push(req.namespace.as_str().to_string());
2985 Ok(GateDecision::allow())
2986 }
2987 }
2988
2989 #[tokio::test]
2990 async fn dispatch_propagates_params_namespace_to_gate() {
2991 let gate = Arc::new(NamespaceCapturingGate::default());
2992 let mut builder = VerbRegistryBuilder::new();
2993 builder.register(AlphaPack);
2994 builder.with_gate(gate.clone());
2995 builder.with_default_namespace("tenant-x");
2996 let reg = builder.build().expect("registry builds");
2997
2998 reg.dispatch("list", serde_json::json!({"namespace": "tenant-y"}))
3000 .await
3001 .unwrap();
3002 reg.dispatch("list", Value::Null).await.unwrap();
3004 let err = reg
3006 .dispatch("list", serde_json::json!({"namespace": ""}))
3007 .await
3008 .unwrap_err();
3009 assert!(
3010 matches!(err, RuntimeError::InvalidInput(_)),
3011 "empty namespace must return InvalidInput, got {err:?}"
3012 );
3013
3014 let seen = gate.seen.lock().unwrap().clone();
3015 assert_eq!(seen, vec!["tenant-y", "tenant-x"]);
3016 }
3017
3018 #[tokio::test]
3019 async fn dispatch_falls_back_to_local_when_no_default_set() {
3020 let gate = Arc::new(NamespaceCapturingGate::default());
3022 let mut builder = VerbRegistryBuilder::new();
3023 builder.register(AlphaPack);
3024 builder.with_gate(gate.clone());
3025 let reg = builder.build().expect("registry builds");
3026
3027 reg.dispatch("list", Value::Null).await.unwrap();
3028 let seen = gate.seen.lock().unwrap().clone();
3029 assert_eq!(seen, vec!["local"]);
3030 }
3031
3032 #[tokio::test]
3038 async fn namespace_null_rejected_not_coerced() {
3039 let cases: Vec<(&str, Value)> = vec![
3040 ("null", Value::Null),
3041 ("number", serde_json::json!(42)),
3042 ("boolean", serde_json::json!(true)),
3043 ("array", serde_json::json!(["local"])),
3044 ("object", serde_json::json!({"ns": "local"})),
3045 ];
3046
3047 for (label, ns_value) in cases {
3048 let gate = Arc::new(NamespaceCapturingGate::default());
3049 let mut builder = VerbRegistryBuilder::new();
3050 builder.register(AlphaPack);
3051 builder.with_gate(gate.clone());
3052 builder.with_default_namespace("tenant-x");
3053 let reg = builder.build().expect("registry builds");
3054
3055 let err = reg
3056 .dispatch("list", serde_json::json!({"namespace": ns_value}))
3057 .await
3058 .unwrap_err();
3059 assert!(
3060 matches!(err, RuntimeError::InvalidInput(_)),
3061 "case {label}: expected InvalidInput, got {err:?}"
3062 );
3063
3064 let seen = gate.seen.lock().unwrap().clone();
3068 assert!(
3069 seen.is_empty(),
3070 "case {label}: gate must not be consulted for malformed namespace, saw {seen:?}"
3071 );
3072 }
3073 }
3074
3075 use khive_gate::{AuditDecision, AuditEvent, Obligation};
3078
3079 #[derive(Default, Debug)]
3081 struct AuditCapturingGate {
3082 events: std::sync::Mutex<Vec<AuditEvent>>,
3083 deny_verb: Option<&'static str>,
3084 }
3085
3086 impl Gate for AuditCapturingGate {
3087 fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
3088 let decision = if Some(req.verb.as_str()) == self.deny_verb {
3089 GateDecision::deny("test deny")
3090 } else {
3091 GateDecision::allow_with(vec![Obligation::Audit {
3092 tag: format!("{}.check", req.verb),
3093 }])
3094 };
3095 let ev = AuditEvent::from_check(req, &decision, self.impl_name());
3097 self.events.lock().unwrap().push(ev);
3098 Ok(decision)
3099 }
3100
3101 fn impl_name(&self) -> &'static str {
3102 "AuditCapturingGate"
3103 }
3104 }
3105
3106 #[tokio::test]
3107 async fn dispatch_emits_one_audit_event_per_call() {
3108 let gate = Arc::new(AuditCapturingGate::default());
3109 let mut builder = VerbRegistryBuilder::new();
3110 builder.register(AlphaPack);
3111 builder.with_gate(gate.clone());
3112 let reg = builder.build().expect("registry builds");
3113
3114 reg.dispatch("list", Value::Null).await.unwrap();
3115 reg.dispatch("create", Value::Null).await.unwrap();
3116
3117 let evs = gate.events.lock().unwrap();
3118 assert_eq!(evs.len(), 2, "exactly one audit event per dispatch call");
3119 }
3120
3121 #[tokio::test]
3122 async fn dispatch_audit_event_allow_carries_obligations() {
3123 let gate = Arc::new(AuditCapturingGate::default());
3124 let mut builder = VerbRegistryBuilder::new();
3125 builder.register(AlphaPack);
3126 builder.with_gate(gate.clone());
3127 let reg = builder.build().expect("registry builds");
3128
3129 reg.dispatch("list", Value::Null).await.unwrap();
3130
3131 let evs = gate.events.lock().unwrap();
3132 let ev = &evs[0];
3133 assert_eq!(ev.verb, "list");
3134 assert_eq!(ev.decision, AuditDecision::Allow);
3135 assert!(ev.deny_reason.is_none());
3136 assert_eq!(ev.obligations.len(), 1);
3137 assert_eq!(ev.gate_impl, "AuditCapturingGate");
3138 }
3139
3140 #[tokio::test]
3141 async fn dispatch_audit_event_deny_carries_reason() {
3142 let gate = Arc::new(AuditCapturingGate {
3143 events: Default::default(),
3144 deny_verb: Some("create"),
3145 });
3146 let mut builder = VerbRegistryBuilder::new();
3147 builder.register(AlphaPack);
3148 builder.with_gate(gate.clone());
3149 let reg = builder.build().expect("registry builds");
3150
3151 let err = reg.dispatch("create", Value::Null).await.unwrap_err();
3154 assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
3155
3156 let evs = gate.events.lock().unwrap();
3157 let ev = &evs[0];
3158 assert_eq!(ev.verb, "create");
3159 assert_eq!(ev.decision, AuditDecision::Deny);
3160 assert_eq!(ev.deny_reason.as_deref(), Some("test deny"));
3161 assert!(ev.obligations.is_empty());
3162 }
3163
3164 #[tokio::test]
3165 async fn dispatch_audit_event_fields_match_gate_request() {
3166 let gate = Arc::new(AuditCapturingGate::default());
3167 let mut builder = VerbRegistryBuilder::new();
3168 builder.register(AlphaPack);
3169 builder.with_gate(gate.clone());
3170 builder.with_default_namespace("tenant-z");
3171 let reg = builder.build().expect("registry builds");
3172
3173 reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
3174 .await
3175 .unwrap();
3176
3177 let evs = gate.events.lock().unwrap();
3178 let ev = &evs[0];
3179 assert_eq!(ev.namespace, "tenant-q");
3181 assert_eq!(ev.verb, "list");
3182 assert_eq!(ev.actor.kind, "anonymous");
3183 }
3184
3185 #[derive(Default, Debug)]
3189 struct ActorCapturingGate {
3190 requests: std::sync::Mutex<Vec<GateRequest>>,
3191 }
3192
3193 impl Gate for ActorCapturingGate {
3194 fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
3195 self.requests.lock().unwrap().push(req.clone());
3196 Ok(GateDecision::allow())
3197 }
3198 }
3199
3200 #[tokio::test]
3204 async fn gate_request_carries_configured_actor_when_actor_id_is_set() {
3205 let gate = Arc::new(ActorCapturingGate::default());
3206 let mut builder = VerbRegistryBuilder::new();
3207 builder.register(AlphaPack);
3208 builder.with_gate(gate.clone());
3209 builder.with_actor_id(Some("team-abc:implementer".to_string()));
3210 let reg = builder.build().expect("registry builds");
3211
3212 reg.dispatch("list", Value::Null).await.unwrap();
3213
3214 let reqs = gate.requests.lock().unwrap();
3215 assert_eq!(reqs.len(), 1);
3216 let req = &reqs[0];
3217 assert_eq!(
3218 req.actor.kind, "actor",
3219 "gate request must carry kind='actor' when actor_id is configured"
3220 );
3221 assert_eq!(
3222 req.actor.id, "team-abc:implementer",
3223 "gate request must carry the configured actor id"
3224 );
3225 }
3226
3227 #[tokio::test]
3230 async fn gate_request_carries_anonymous_when_no_actor_id_configured() {
3231 let gate = Arc::new(ActorCapturingGate::default());
3232 let mut builder = VerbRegistryBuilder::new();
3233 builder.register(AlphaPack);
3234 builder.with_gate(gate.clone());
3235 let reg = builder.build().expect("registry builds");
3237
3238 reg.dispatch("list", Value::Null).await.unwrap();
3239
3240 let reqs = gate.requests.lock().unwrap();
3241 assert_eq!(reqs.len(), 1);
3242 let req = &reqs[0];
3243 assert_eq!(
3244 req.actor.kind, "anonymous",
3245 "gate request must carry anonymous actor when no actor_id is configured"
3246 );
3247 assert_eq!(req.actor.id, "local");
3248 }
3249
3250 struct TokenCapturingPack {
3253 actors: Arc<std::sync::Mutex<Vec<khive_gate::ActorRef>>>,
3254 }
3255
3256 impl Pack for TokenCapturingPack {
3257 const NAME: &'static str = "alpha";
3258 const NOTE_KINDS: &'static [&'static str] = &[];
3259 const ENTITY_KINDS: &'static [&'static str] = &[];
3260 const HANDLERS: &'static [HandlerDef] = AlphaPack::HANDLERS;
3261 }
3262
3263 #[async_trait]
3264 impl PackRuntime for TokenCapturingPack {
3265 fn name(&self) -> &str {
3266 Self::NAME
3267 }
3268 fn note_kinds(&self) -> &'static [&'static str] {
3269 Self::NOTE_KINDS
3270 }
3271 fn entity_kinds(&self) -> &'static [&'static str] {
3272 Self::ENTITY_KINDS
3273 }
3274 fn handlers(&self) -> &'static [HandlerDef] {
3275 Self::HANDLERS
3276 }
3277 async fn dispatch(
3278 &self,
3279 verb: &str,
3280 _params: Value,
3281 _registry: &VerbRegistry,
3282 token: &NamespaceToken,
3283 ) -> Result<Value, RuntimeError> {
3284 self.actors.lock().unwrap().push(token.actor().clone());
3285 Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
3286 }
3287 }
3288
3289 #[tokio::test]
3298 async fn gate_actor_and_token_actor_are_identical_when_actor_id_is_set() {
3299 let gate = Arc::new(ActorCapturingGate::default());
3300 let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3301 let pack = TokenCapturingPack {
3302 actors: actors.clone(),
3303 };
3304 let mut builder = VerbRegistryBuilder::new();
3305 builder.register(pack);
3306 builder.with_gate(gate.clone());
3307 builder.with_actor_id(Some("actor-alpha".to_string()));
3308 let reg = builder.build().expect("registry builds");
3309
3310 reg.dispatch("list", Value::Null).await.unwrap();
3311
3312 let reqs = gate.requests.lock().unwrap();
3313 let gate_actor = reqs[0].actor.clone();
3314 drop(reqs);
3315
3316 let captured = actors.lock().unwrap();
3317 let token_actor = captured[0].clone();
3318
3319 assert_eq!(
3320 gate_actor.kind, token_actor.kind,
3321 "gate request actor and storage token actor must carry the same kind"
3322 );
3323 assert_eq!(
3324 gate_actor.id, token_actor.id,
3325 "gate request actor and storage token actor must carry the same id"
3326 );
3327 assert_eq!(gate_actor.id, "actor-alpha");
3328 }
3329
3330 #[tokio::test]
3333 async fn gate_actor_and_token_actor_are_identical_when_anonymous() {
3334 let gate = Arc::new(ActorCapturingGate::default());
3335 let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3336 let pack = TokenCapturingPack {
3337 actors: actors.clone(),
3338 };
3339 let mut builder = VerbRegistryBuilder::new();
3340 builder.register(pack);
3341 builder.with_gate(gate.clone());
3342 let reg = builder.build().expect("registry builds");
3343
3344 reg.dispatch("list", Value::Null).await.unwrap();
3345
3346 let reqs = gate.requests.lock().unwrap();
3347 let gate_actor = reqs[0].actor.clone();
3348 drop(reqs);
3349
3350 let captured = actors.lock().unwrap();
3351 let token_actor = captured[0].clone();
3352
3353 assert_eq!(gate_actor.kind, token_actor.kind);
3354 assert_eq!(gate_actor.id, token_actor.id);
3355 assert_eq!(gate_actor.id, "local");
3356 }
3357
3358 #[tokio::test]
3365 async fn dispatch_as_threads_verified_actor_into_token() {
3366 let gate = Arc::new(ActorCapturingGate::default());
3367 let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3368 let pack = TokenCapturingPack {
3369 actors: actors.clone(),
3370 };
3371 let mut builder = VerbRegistryBuilder::new();
3372 builder.register(pack);
3373 builder.with_gate(gate.clone());
3374 let reg = builder.build().expect("registry builds");
3375
3376 reg.dispatch_as(
3377 "list",
3378 Value::Null,
3379 VerifiedActor::new("gateway:principal-42").unwrap(),
3380 )
3381 .await
3382 .unwrap();
3383
3384 let reqs = gate.requests.lock().unwrap();
3385 assert_eq!(reqs[0].actor.kind, "actor");
3386 assert_eq!(reqs[0].actor.id, "gateway:principal-42");
3387 drop(reqs);
3388
3389 let captured = actors.lock().unwrap();
3390 assert_eq!(captured[0].kind, "actor");
3391 assert_eq!(
3392 captured[0].id, "gateway:principal-42",
3393 "the storage token actor must be the verified_actor supplied to dispatch_as, \
3394 matching exactly what pack handlers read as the acting principal"
3395 );
3396 }
3397
3398 #[tokio::test]
3403 async fn dispatch_as_does_not_change_plain_dispatch_behavior() {
3404 let gate = Arc::new(ActorCapturingGate::default());
3405 let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3406 let pack = TokenCapturingPack {
3407 actors: actors.clone(),
3408 };
3409 let mut builder = VerbRegistryBuilder::new();
3410 builder.register(pack);
3411 builder.with_gate(gate.clone());
3412 builder.with_actor_id(Some("baked-actor".to_string()));
3413 let reg = builder.build().expect("registry builds");
3414
3415 reg.dispatch_as(
3416 "list",
3417 Value::Null,
3418 VerifiedActor::new("verified-actor").unwrap(),
3419 )
3420 .await
3421 .unwrap();
3422 reg.dispatch("list", Value::Null).await.unwrap();
3423
3424 let captured = actors.lock().unwrap();
3425 assert_eq!(captured.len(), 2);
3426 assert_eq!(captured[0].id, "verified-actor", "dispatch_as call");
3427 assert_eq!(
3428 captured[1].id, "baked-actor",
3429 "a later plain dispatch() call must still use the registry's baked \
3430 actor_id, unaffected by the prior dispatch_as call"
3431 );
3432 }
3433
3434 #[tokio::test]
3441 async fn dispatch_as_ignores_actor_key_in_params() {
3442 let gate = Arc::new(ActorCapturingGate::default());
3443 let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3444 let pack = TokenCapturingPack {
3445 actors: actors.clone(),
3446 };
3447 let mut builder = VerbRegistryBuilder::new();
3448 builder.register(pack);
3449 builder.with_gate(gate.clone());
3450 let reg = builder.build().expect("registry builds");
3451
3452 reg.dispatch_as(
3453 "list",
3454 serde_json::json!({"actor": "spoofed-actor"}),
3455 VerifiedActor::new("verified-actor").unwrap(),
3456 )
3457 .await
3458 .unwrap();
3459
3460 let captured = actors.lock().unwrap();
3461 assert_eq!(
3462 captured[0].id, "verified-actor",
3463 "an 'actor' key inside params must never override the verified_actor \
3464 argument threaded through dispatch_as"
3465 );
3466 }
3467
3468 #[test]
3471 fn verified_actor_rejects_empty_identifier() {
3472 let err = VerifiedActor::new("").unwrap_err();
3473 assert!(
3474 matches!(err, RuntimeError::InvalidInput(_)),
3475 "expected InvalidInput, got {err:?}"
3476 );
3477 }
3478
3479 #[test]
3482 fn verified_actor_rejects_whitespace_only_identifier() {
3483 let err = VerifiedActor::new(" ").unwrap_err();
3484 assert!(
3485 matches!(err, RuntimeError::InvalidInput(_)),
3486 "expected InvalidInput, got {err:?}"
3487 );
3488 }
3489
3490 #[tokio::test]
3504 async fn rego_gate_missing_entrypoint_returns_permission_denied() {
3505 use khive_gate_rego::RegoGate;
3506
3507 let policy = r#"
3513 package khive.gate
3514 import rego.v1
3515 verdict := "allow"
3516 "#;
3517 let gate = Arc::new(RegoGate::from_policy_str(policy).expect("policy compiles"));
3518
3519 let mut builder = VerbRegistryBuilder::new();
3520 builder.register(AlphaPack);
3521 builder.with_gate(gate);
3522 let reg = builder.build().expect("registry builds");
3523
3524 let err = reg.dispatch("create", Value::Null).await.unwrap_err();
3525 assert!(
3526 matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
3527 "expected PermissionDenied for missing rego entrypoint, got {err:?}"
3528 );
3529 }
3530
3531 use std::sync::{Mutex as StdMutex, Once, OnceLock};
3543
3544 use serial_test::serial;
3545 use tracing::field::{Field, Visit};
3546
3547 #[derive(Clone, Debug, Default)]
3548 struct CapturedEvent {
3549 message: Option<String>,
3550 audit_event: Option<String>,
3551 }
3552
3553 #[derive(Default)]
3554 struct CapturedEventVisitor(CapturedEvent);
3555
3556 impl Visit for CapturedEventVisitor {
3557 fn record_str(&mut self, field: &Field, value: &str) {
3558 match field.name() {
3559 "message" => self.0.message = Some(value.to_string()),
3560 "audit_event" => self.0.audit_event = Some(value.to_string()),
3561 _ => {}
3562 }
3563 }
3564
3565 fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
3566 let formatted = format!("{value:?}");
3572 let cleaned = formatted
3573 .trim_start_matches('"')
3574 .trim_end_matches('"')
3575 .to_string();
3576 match field.name() {
3577 "message" => self.0.message = Some(cleaned),
3578 "audit_event" => self.0.audit_event = Some(cleaned),
3579 _ => {}
3580 }
3581 }
3582 }
3583
3584 struct CaptureSubscriber {
3597 events: Arc<StdMutex<Vec<CapturedEvent>>>,
3598 }
3599
3600 impl CaptureSubscriber {
3601 fn new(events: Arc<StdMutex<Vec<CapturedEvent>>>) -> Self {
3602 Self { events }
3603 }
3604 }
3605
3606 impl tracing::Subscriber for CaptureSubscriber {
3607 fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
3608 true
3609 }
3610 fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
3611 tracing::span::Id::from_u64(1)
3612 }
3613 fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
3614 fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
3615 fn event(&self, event: &tracing::Event<'_>) {
3616 let mut visitor = CapturedEventVisitor::default();
3617 event.record(&mut visitor);
3618 self.events.lock().unwrap().push(visitor.0);
3619 }
3620 fn enter(&self, _: &tracing::span::Id) {}
3621 fn exit(&self, _: &tracing::span::Id) {}
3622 }
3623
3624 static GLOBAL_CAPTURE: OnceLock<Arc<StdMutex<Vec<CapturedEvent>>>> = OnceLock::new();
3634 static GLOBAL_INIT: Once = Once::new();
3635
3636 fn global_capture() -> Arc<StdMutex<Vec<CapturedEvent>>> {
3637 GLOBAL_INIT.call_once(|| {
3638 let buffer = Arc::new(StdMutex::new(Vec::new()));
3639 let subscriber = CaptureSubscriber::new(Arc::clone(&buffer));
3640 let _ = tracing::subscriber::set_global_default(subscriber);
3645 let _ = GLOBAL_CAPTURE.set(buffer);
3646 });
3647 Arc::clone(GLOBAL_CAPTURE.get().expect("global capture initialized"))
3648 }
3649
3650 fn capture_dispatch_events<Fut>(future: Fut) -> Vec<CapturedEvent>
3655 where
3656 Fut: std::future::Future<Output = ()>,
3657 {
3658 let buffer = global_capture();
3659 buffer.lock().unwrap().clear();
3660
3661 let rt = tokio::runtime::Builder::new_current_thread()
3662 .enable_all()
3663 .build()
3664 .expect("build current-thread tokio runtime");
3665 rt.block_on(future);
3666
3667 let result = buffer.lock().unwrap().clone();
3668 result
3669 }
3670
3671 fn gate_check_events_for(events: &[CapturedEvent], gate_impl: &str) -> Vec<CapturedEvent> {
3678 events
3679 .iter()
3680 .filter(|e| e.message.as_deref() == Some("gate.check"))
3681 .filter(|e| {
3682 e.audit_event
3683 .as_deref()
3684 .and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
3685 .and_then(|v| {
3686 v.get("gate_impl")
3687 .and_then(|g| g.as_str().map(|s| s.to_string()))
3688 })
3689 .as_deref()
3690 == Some(gate_impl)
3691 })
3692 .cloned()
3693 .collect()
3694 }
3695
3696 #[test]
3697 #[serial]
3698 fn dispatch_tracing_emits_one_gate_check_event_on_allow() {
3699 #[derive(Debug)]
3700 struct TracingAllowGate;
3701 impl Gate for TracingAllowGate {
3702 fn check(&self, _: &GateRequest) -> Result<GateDecision, GateError> {
3703 Ok(GateDecision::allow())
3704 }
3705 fn impl_name(&self) -> &'static str {
3706 "TracingAllowGate"
3707 }
3708 }
3709
3710 let events = capture_dispatch_events(async {
3711 let mut builder = VerbRegistryBuilder::new();
3712 builder.register(AlphaPack);
3713 builder.with_gate(Arc::new(TracingAllowGate));
3714 builder.with_default_namespace("tenant-default");
3715 let reg = builder.build().expect("registry builds");
3716 reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
3717 .await
3718 .unwrap();
3719 });
3720
3721 let gate_events = gate_check_events_for(&events, "TracingAllowGate");
3722 assert_eq!(
3723 gate_events.len(),
3724 1,
3725 "exactly one gate.check tracing event per dispatch (allow); got {gate_events:?}"
3726 );
3727 let payload = gate_events[0]
3728 .audit_event
3729 .as_ref()
3730 .expect("gate.check event must carry an audit_event field");
3731 let audit: khive_gate::AuditEvent =
3732 serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
3733 assert_eq!(audit.decision, AuditDecision::Allow);
3734 assert_eq!(audit.verb, "list");
3735 assert_eq!(audit.namespace, "tenant-q");
3736 assert_eq!(audit.gate_impl, "TracingAllowGate");
3737 assert!(
3738 audit.deny_reason.is_none(),
3739 "deny_reason must be None on Allow"
3740 );
3741 }
3742
3743 use crate::runtime::NamespaceToken;
3746 use async_trait::async_trait;
3747 use khive_storage::{
3748 BatchWriteSummary, Event, EventFilter, EventStore, Page, PageRequest, SubstrateKind,
3749 };
3750 use khive_types::EventOutcome;
3751
3752 #[derive(Default, Debug)]
3754 struct MemoryEventStore {
3755 events: std::sync::Mutex<Vec<Event>>,
3756 }
3757
3758 #[async_trait]
3759 impl EventStore for MemoryEventStore {
3760 async fn append_event(&self, event: Event) -> khive_storage::StorageResult<()> {
3761 self.events.lock().unwrap().push(event);
3762 Ok(())
3763 }
3764 async fn append_events(
3765 &self,
3766 events: Vec<Event>,
3767 ) -> khive_storage::StorageResult<BatchWriteSummary> {
3768 let attempted = events.len() as u64;
3769 let affected = attempted;
3770 self.events.lock().unwrap().extend(events);
3771 Ok(BatchWriteSummary {
3772 attempted,
3773 affected,
3774 failed: 0,
3775 first_error: String::new(),
3776 })
3777 }
3778 async fn get_event(&self, id: uuid::Uuid) -> khive_storage::StorageResult<Option<Event>> {
3779 Ok(self
3780 .events
3781 .lock()
3782 .unwrap()
3783 .iter()
3784 .find(|e| e.id == id)
3785 .cloned())
3786 }
3787 async fn query_events(
3788 &self,
3789 _filter: EventFilter,
3790 _page: PageRequest,
3791 ) -> khive_storage::StorageResult<Page<Event>> {
3792 let items = self.events.lock().unwrap().clone();
3793 let total = items.len() as u64;
3794 Ok(Page {
3795 items,
3796 total: Some(total),
3797 })
3798 }
3799 async fn count_events(&self, _filter: EventFilter) -> khive_storage::StorageResult<u64> {
3800 Ok(self.events.lock().unwrap().len() as u64)
3801 }
3802 }
3803
3804 #[tokio::test]
3805 async fn allow_all_gate_default_remains_backward_compatible() {
3806 let mut builder = VerbRegistryBuilder::new();
3808 builder.register(AlphaPack);
3809 let reg = builder.build().expect("registry builds");
3810
3811 let res = reg.dispatch("list", Value::Null).await.unwrap();
3812 assert_eq!(
3813 res["pack"], "alpha",
3814 "AllowAllGate must allow every verb — backward compat guarantee"
3815 );
3816 let res = reg.dispatch("create", Value::Null).await.unwrap();
3817 assert_eq!(res["pack"], "alpha");
3818 }
3819
3820 #[tokio::test]
3821 async fn deny_gate_returns_permission_denied_pack_never_invoked() {
3822 #[derive(Debug)]
3823 struct AlwaysDenyGate;
3824 impl Gate for AlwaysDenyGate {
3825 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
3826 Ok(GateDecision::deny("test: always deny"))
3827 }
3828 }
3829
3830 #[derive(Debug)]
3832 struct TrackedPack {
3833 invoked: Arc<AtomicUsize>,
3834 }
3835
3836 impl khive_types::Pack for TrackedPack {
3837 const NAME: &'static str = "tracked";
3838 const NOTE_KINDS: &'static [&'static str] = &[];
3839 const ENTITY_KINDS: &'static [&'static str] = &[];
3840 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
3841 name: "guarded",
3842 description: "a guarded verb",
3843 visibility: Visibility::Verb,
3844 category: VerbCategory::Assertive,
3845 params: &[],
3846 }];
3847 }
3848
3849 #[async_trait]
3850 impl PackRuntime for TrackedPack {
3851 fn name(&self) -> &str {
3852 Self::NAME
3853 }
3854 fn note_kinds(&self) -> &'static [&'static str] {
3855 Self::NOTE_KINDS
3856 }
3857 fn entity_kinds(&self) -> &'static [&'static str] {
3858 Self::ENTITY_KINDS
3859 }
3860 fn handlers(&self) -> &'static [HandlerDef] {
3861 Self::HANDLERS
3862 }
3863 async fn dispatch(
3864 &self,
3865 _verb: &str,
3866 _params: Value,
3867 _registry: &VerbRegistry,
3868 _token: &NamespaceToken,
3869 ) -> Result<Value, RuntimeError> {
3870 self.invoked.fetch_add(1, Ordering::SeqCst);
3871 Ok(serde_json::json!({"invoked": true}))
3872 }
3873 }
3874
3875 let invoked = Arc::new(AtomicUsize::new(0));
3876 let mut builder = VerbRegistryBuilder::new();
3877 builder.register(TrackedPack {
3878 invoked: invoked.clone(),
3879 });
3880 builder.with_gate(Arc::new(AlwaysDenyGate));
3881 let reg = builder.build().expect("registry builds");
3882
3883 let err = reg.dispatch("guarded", Value::Null).await.unwrap_err();
3884 assert!(
3885 matches!(err, RuntimeError::PermissionDenied { ref verb, ref reason } if verb == "guarded" && reason.contains("always deny")),
3886 "expected PermissionDenied with verb=guarded and reason, got: {err:?}"
3887 );
3888 assert_eq!(
3889 invoked.load(Ordering::SeqCst),
3890 0,
3891 "pack dispatch MUST NOT be invoked when gate denies"
3892 );
3893 }
3894
3895 #[tokio::test]
3896 async fn audit_event_persists_to_event_store_on_allow() {
3897 let store = Arc::new(MemoryEventStore::default());
3898 let mut builder = VerbRegistryBuilder::new();
3899 builder.register(AlphaPack);
3900 builder.with_event_store(store.clone());
3901 let reg = builder.build().expect("registry builds");
3902
3903 reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
3904 .await
3905 .unwrap();
3906
3907 let count = store.count_events(EventFilter::default()).await.unwrap();
3908 assert_eq!(count, 1, "one audit event persisted to EventStore on allow");
3909
3910 let page = store
3911 .query_events(
3912 EventFilter::default(),
3913 PageRequest {
3914 limit: 10,
3915 offset: 0,
3916 },
3917 )
3918 .await
3919 .unwrap();
3920 let ev = &page.items[0];
3921 assert_eq!(ev.verb, "list");
3922 assert_eq!(ev.namespace, "test-ns");
3923 assert_eq!(ev.substrate, SubstrateKind::Event);
3924 assert_eq!(ev.outcome, EventOutcome::Success);
3925 }
3926
3927 #[tokio::test]
3928 async fn audit_event_duration_us_reflects_measured_dispatch_time() {
3929 let store = Arc::new(MemoryEventStore::default());
3935 let mut builder = VerbRegistryBuilder::new();
3936 builder.register(SleepingPack);
3937 builder.with_event_store(store.clone());
3938 let reg = builder.build().expect("registry builds");
3939
3940 reg.dispatch("slow_op", serde_json::json!({}))
3941 .await
3942 .unwrap();
3943
3944 let page = store
3945 .query_events(
3946 EventFilter::default(),
3947 PageRequest {
3948 limit: 10,
3949 offset: 0,
3950 },
3951 )
3952 .await
3953 .unwrap();
3954 assert_eq!(page.items.len(), 1);
3955 let ev = &page.items[0];
3956 assert!(
3957 ev.duration_us >= 10_000,
3958 "duration_us must reflect the ~20ms measured dispatch time, got {}",
3959 ev.duration_us
3960 );
3961 }
3962
3963 #[tokio::test]
3964 async fn dispatch_unknown_verb_allowed_by_gate_still_persists_audit_row() {
3965 let store = Arc::new(MemoryEventStore::default());
3970 let mut builder = VerbRegistryBuilder::new();
3971 builder.register(AlphaPack);
3972 builder.with_event_store(store.clone());
3973 let reg = builder.build().expect("registry builds");
3974
3975 let result = reg.dispatch("no_such_verb", serde_json::json!({})).await;
3976 assert!(result.is_err(), "unknown verb must still return an error");
3977
3978 let count = store.count_events(EventFilter::default()).await.unwrap();
3979 assert_eq!(
3980 count, 1,
3981 "an allowed-but-unknown verb must still persist one audit row"
3982 );
3983 let page = store
3984 .query_events(
3985 EventFilter::default(),
3986 PageRequest {
3987 limit: 10,
3988 offset: 0,
3989 },
3990 )
3991 .await
3992 .unwrap();
3993 assert_eq!(page.items[0].duration_us, 0);
3994 assert_eq!(page.items[0].outcome, EventOutcome::Error);
3998 }
3999
4000 #[tokio::test]
4001 async fn audit_event_persists_to_event_store_on_deny() {
4002 #[derive(Debug)]
4003 struct AlwaysDenyGate;
4004 impl Gate for AlwaysDenyGate {
4005 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4006 Ok(GateDecision::deny("denied by test"))
4007 }
4008 }
4009
4010 let store = Arc::new(MemoryEventStore::default());
4011 let mut builder = VerbRegistryBuilder::new();
4012 builder.register(AlphaPack);
4013 builder.with_gate(Arc::new(AlwaysDenyGate));
4014 builder.with_event_store(store.clone());
4015 let reg = builder.build().expect("registry builds");
4016
4017 let err = reg
4019 .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4020 .await
4021 .unwrap_err();
4022 assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
4023
4024 let count = store.count_events(EventFilter::default()).await.unwrap();
4025 assert_eq!(count, 1, "one audit event persisted to EventStore on deny");
4026
4027 let page = store
4028 .query_events(
4029 EventFilter::default(),
4030 PageRequest {
4031 limit: 10,
4032 offset: 0,
4033 },
4034 )
4035 .await
4036 .unwrap();
4037 let ev = &page.items[0];
4038 assert_eq!(ev.verb, "list");
4039 assert_eq!(ev.outcome, EventOutcome::Denied);
4040 }
4041
4042 #[tokio::test]
4043 async fn gate_error_does_not_persist_to_event_store() {
4044 #[derive(Debug)]
4045 struct FailingGate;
4046 impl Gate for FailingGate {
4047 fn check(&self, _req: &GateRequest) -> Result<GateDecision, khive_gate::GateError> {
4048 Err(khive_gate::GateError::Internal("gate broken".into()))
4049 }
4050 }
4051
4052 let store = Arc::new(MemoryEventStore::default());
4053 let mut builder = VerbRegistryBuilder::new();
4054 builder.register(AlphaPack);
4055 builder.with_gate(Arc::new(FailingGate));
4056 builder.with_event_store(store.clone());
4057 let reg = builder.build().expect("registry builds");
4058
4059 let res = reg.dispatch("list", Value::Null).await.unwrap();
4061 assert_eq!(
4062 res["pack"], "alpha",
4063 "gate error must fail-open, not block dispatch"
4064 );
4065
4066 let count = store.count_events(EventFilter::default()).await.unwrap();
4067 assert_eq!(
4068 count, 0,
4069 "gate infrastructure error must NOT produce an audit event in EventStore"
4070 );
4071 }
4072
4073 #[tokio::test]
4074 async fn no_event_store_configured_tracing_only() {
4075 let mut builder = VerbRegistryBuilder::new();
4079 builder.register(AlphaPack);
4080 let reg = builder.build().expect("registry builds");
4081
4082 let res = reg.dispatch("list", Value::Null).await.unwrap();
4083 assert_eq!(res["pack"], "alpha");
4084 }
4085
4086 #[test]
4087 #[serial]
4088 fn dispatch_tracing_emits_gate_check_event_with_deny_payload() {
4089 #[derive(Debug)]
4090 struct TracingDenyGate;
4091 impl Gate for TracingDenyGate {
4092 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4093 Ok(GateDecision::deny("denied by test gate"))
4094 }
4095 fn impl_name(&self) -> &'static str {
4096 "TracingDenyGate"
4097 }
4098 }
4099
4100 let events = capture_dispatch_events(async {
4101 let mut builder = VerbRegistryBuilder::new();
4102 builder.register(AlphaPack);
4103 builder.with_gate(Arc::new(TracingDenyGate));
4104 let reg = builder.build().expect("registry builds");
4105 let _ = reg.dispatch("create", serde_json::Value::Null).await;
4108 });
4109
4110 let gate_events = gate_check_events_for(&events, "TracingDenyGate");
4111 assert_eq!(
4112 gate_events.len(),
4113 1,
4114 "exactly one gate.check tracing event per dispatch (deny); got {gate_events:?}"
4115 );
4116 let payload = gate_events[0]
4117 .audit_event
4118 .as_ref()
4119 .expect("gate.check event must carry an audit_event field on Deny");
4120 let audit: khive_gate::AuditEvent =
4121 serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
4122 assert_eq!(audit.decision, AuditDecision::Deny);
4123 assert_eq!(audit.deny_reason.as_deref(), Some("denied by test gate"));
4124 assert_eq!(audit.gate_impl, "TracingDenyGate");
4125 let payload_json: serde_json::Value =
4129 serde_json::from_str(payload).expect("payload must be valid JSON");
4130 assert_eq!(
4131 payload_json["obligations"],
4132 serde_json::Value::Array(Vec::new()),
4133 "obligations must be `[]` on Deny on the tracing payload, not omitted"
4134 );
4135 }
4136
4137 #[tokio::test]
4144 async fn audit_envelope_round_trips_deny_reason_and_gate_impl_through_event_store() {
4145 #[derive(Debug)]
4146 struct DenyGateWithName;
4147 impl Gate for DenyGateWithName {
4148 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4149 Ok(GateDecision::deny("policy: write forbidden for anon"))
4150 }
4151 fn impl_name(&self) -> &'static str {
4152 "DenyGateWithName"
4153 }
4154 }
4155
4156 let store = Arc::new(MemoryEventStore::default());
4157 let mut builder = VerbRegistryBuilder::new();
4158 builder.register(AlphaPack);
4159 builder.with_gate(Arc::new(DenyGateWithName));
4160 builder.with_event_store(store.clone());
4161 let reg = builder.build().expect("registry builds");
4162
4163 let err = reg
4165 .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4166 .await
4167 .unwrap_err();
4168 assert!(
4169 matches!(err, RuntimeError::PermissionDenied { .. }),
4170 "expected PermissionDenied, got {err:?}"
4171 );
4172
4173 let page = store
4175 .query_events(
4176 EventFilter::default(),
4177 PageRequest {
4178 limit: 10,
4179 offset: 0,
4180 },
4181 )
4182 .await
4183 .unwrap();
4184 assert_eq!(
4185 page.items.len(),
4186 1,
4187 "one audit event must be persisted on deny"
4188 );
4189
4190 let ev = &page.items[0];
4191 assert_eq!(ev.outcome, EventOutcome::Denied);
4192
4193 let data = &ev.payload;
4195
4196 let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4197 .expect("Event.payload must deserialize to AuditEvent");
4198
4199 assert_eq!(
4200 audit.deny_reason.as_deref(),
4201 Some("policy: write forbidden for anon"),
4202 "deny_reason must be preserved through EventStore"
4203 );
4204 assert_eq!(
4205 audit.gate_impl, "DenyGateWithName",
4206 "gate_impl must be preserved through EventStore"
4207 );
4208 assert_eq!(
4209 audit.decision,
4210 khive_gate::AuditDecision::Deny,
4211 "decision field must be preserved through EventStore"
4212 );
4213 }
4214
4215 #[tokio::test]
4216 async fn audit_envelope_round_trips_obligations_through_event_store() {
4217 use khive_gate::Obligation;
4218
4219 #[derive(Debug)]
4220 struct ObligationGate;
4221 impl Gate for ObligationGate {
4222 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4223 Ok(GateDecision::allow_with(vec![Obligation::Audit {
4224 tag: "billing.meter".into(),
4225 }]))
4226 }
4227 fn impl_name(&self) -> &'static str {
4228 "ObligationGate"
4229 }
4230 }
4231
4232 let store = Arc::new(MemoryEventStore::default());
4233 let mut builder = VerbRegistryBuilder::new();
4234 builder.register(AlphaPack);
4235 builder.with_gate(Arc::new(ObligationGate));
4236 builder.with_event_store(store.clone());
4237 let reg = builder.build().expect("registry builds");
4238
4239 reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4240 .await
4241 .unwrap();
4242
4243 let page = store
4244 .query_events(
4245 EventFilter::default(),
4246 PageRequest {
4247 limit: 10,
4248 offset: 0,
4249 },
4250 )
4251 .await
4252 .unwrap();
4253 assert_eq!(page.items.len(), 1);
4254
4255 let ev = &page.items[0];
4256 assert_eq!(ev.outcome, EventOutcome::Success);
4257
4258 let data = &ev.payload;
4259
4260 let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4261 .expect("Event.payload must deserialize to AuditEvent");
4262
4263 assert_eq!(audit.gate_impl, "ObligationGate");
4264 assert_eq!(
4265 audit.obligations.len(),
4266 1,
4267 "obligations must be preserved through EventStore"
4268 );
4269 match &audit.obligations[0] {
4270 Obligation::Audit { tag } => assert_eq!(tag, "billing.meter"),
4271 other => panic!("expected Audit obligation, got {other:?}"),
4272 }
4273 }
4274
4275 #[tokio::test]
4283 async fn sql_backed_audit_envelope_round_trips_deny_reason_gate_impl_and_obligations() {
4284 #[derive(Debug)]
4285 struct SqlTestDenyGate;
4286 impl Gate for SqlTestDenyGate {
4287 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4288 Ok(GateDecision::deny("sql-path: write denied"))
4289 }
4290 fn impl_name(&self) -> &'static str {
4291 "SqlTestDenyGate"
4292 }
4293 }
4294
4295 let rt = KhiveRuntime::memory().expect("in-memory runtime");
4299 let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
4300 let sql_store = rt
4301 .events(&test_tok)
4302 .expect("events_for_namespace must succeed");
4303
4304 let mut builder = VerbRegistryBuilder::new();
4305 builder.register(AlphaPack);
4306 builder.with_gate(Arc::new(SqlTestDenyGate));
4307 builder.with_event_store(sql_store.clone());
4308 let reg = builder.build().expect("registry builds");
4309
4310 let err = reg
4312 .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4313 .await
4314 .unwrap_err();
4315 assert!(
4316 matches!(err, RuntimeError::PermissionDenied { .. }),
4317 "expected PermissionDenied, got {err:?}"
4318 );
4319
4320 let page = sql_store
4322 .query_events(
4323 EventFilter::default(),
4324 PageRequest {
4325 limit: 10,
4326 offset: 0,
4327 },
4328 )
4329 .await
4330 .unwrap();
4331 assert_eq!(
4332 page.items.len(),
4333 1,
4334 "one audit event must be persisted on deny through SqlEventStore"
4335 );
4336
4337 let ev = &page.items[0];
4338 assert_eq!(ev.outcome, EventOutcome::Denied);
4339
4340 let data = &ev.payload;
4344
4345 let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4346 .expect("Event.payload must deserialize to AuditEvent after SQL round-trip");
4347
4348 assert_eq!(
4349 audit.deny_reason.as_deref(),
4350 Some("sql-path: write denied"),
4351 "deny_reason must survive the SQL text round-trip"
4352 );
4353 assert_eq!(
4354 audit.gate_impl, "SqlTestDenyGate",
4355 "gate_impl must survive the SQL text round-trip"
4356 );
4357 assert_eq!(
4358 audit.decision,
4359 khive_gate::AuditDecision::Deny,
4360 "decision field must survive the SQL text round-trip"
4361 );
4362 assert!(
4365 audit.obligations.is_empty(),
4366 "obligations must be preserved as empty [] through SQL round-trip"
4367 );
4368 }
4369
4370 #[tokio::test]
4382 async fn sql_backed_audit_envelope_round_trips_non_empty_obligations() {
4383 use khive_gate::Obligation;
4384
4385 #[derive(Debug)]
4386 struct SqlTestAllowWithObligationGate;
4387 impl Gate for SqlTestAllowWithObligationGate {
4388 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4389 Ok(GateDecision::allow_with(vec![Obligation::Audit {
4390 tag: "sql-path-billing.meter".into(),
4391 }]))
4392 }
4393 fn impl_name(&self) -> &'static str {
4394 "SqlTestAllowWithObligationGate"
4395 }
4396 }
4397
4398 let rt = KhiveRuntime::memory().expect("in-memory runtime");
4399 let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
4400 let sql_store = rt
4401 .events(&test_tok)
4402 .expect("events_for_namespace must succeed");
4403
4404 let mut builder = VerbRegistryBuilder::new();
4405 builder.register(AlphaPack);
4406 builder.with_gate(Arc::new(SqlTestAllowWithObligationGate));
4407 builder.with_event_store(sql_store.clone());
4408 let reg = builder.build().expect("registry builds");
4409
4410 reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4412 .await
4413 .expect("dispatch must succeed when gate allows");
4414
4415 let page = sql_store
4417 .query_events(
4418 EventFilter::default(),
4419 PageRequest {
4420 limit: 10,
4421 offset: 0,
4422 },
4423 )
4424 .await
4425 .unwrap();
4426 assert_eq!(
4427 page.items.len(),
4428 1,
4429 "one audit event must be persisted on allow through SqlEventStore"
4430 );
4431
4432 let ev = &page.items[0];
4433 assert_eq!(ev.outcome, EventOutcome::Success);
4434
4435 let data = &ev.payload;
4436
4437 let obligations_raw = data
4442 .get("obligations")
4443 .expect("Event.data JSON must contain 'obligations' key");
4444 let obligations_arr = obligations_raw
4445 .as_array()
4446 .expect("'obligations' must be a JSON array");
4447 assert!(
4448 !obligations_arr.is_empty(),
4449 "raw Event.data['obligations'] must be non-empty after SQL round-trip"
4450 );
4451
4452 let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4455 .expect("Event.data must deserialize to AuditEvent after SQL round-trip");
4456
4457 assert_eq!(
4458 audit.gate_impl, "SqlTestAllowWithObligationGate",
4459 "gate_impl must survive the SQL text round-trip"
4460 );
4461 assert_eq!(
4462 audit.decision,
4463 khive_gate::AuditDecision::Allow,
4464 "decision field must survive the SQL text round-trip"
4465 );
4466 assert_eq!(
4467 audit.obligations.len(),
4468 1,
4469 "obligations must be non-empty after SQL round-trip (not silently defaulted to [])"
4470 );
4471 match &audit.obligations[0] {
4472 Obligation::Audit { tag } => assert_eq!(
4473 tag, "sql-path-billing.meter",
4474 "Audit obligation tag must survive the SQL text round-trip"
4475 ),
4476 other => panic!("expected Audit obligation, got {other:?}"),
4477 }
4478 }
4479
4480 #[tokio::test]
4488 async fn audit_event_payload_shape_for_create_verb() {
4489 let store = Arc::new(MemoryEventStore::default());
4490 let mut builder = VerbRegistryBuilder::new();
4491 builder.register(AlphaPack);
4492 builder.with_event_store(store.clone());
4493 builder.with_default_namespace("test-ns");
4494 let reg = builder.build().expect("registry builds");
4495
4496 reg.dispatch("create", serde_json::json!({"namespace": "test-ns"}))
4499 .await
4500 .unwrap();
4501
4502 let count = store.count_events(EventFilter::default()).await.unwrap();
4503 assert_eq!(count, 1, "exactly one audit event for one dispatch");
4504
4505 let page = store
4506 .query_events(
4507 EventFilter::default(),
4508 PageRequest {
4509 limit: 10,
4510 offset: 0,
4511 },
4512 )
4513 .await
4514 .unwrap();
4515 let ev = &page.items[0];
4516
4517 assert_eq!(ev.verb, "create", "ev.verb must be the dispatched verb");
4519 assert_eq!(
4520 ev.outcome,
4521 EventOutcome::Success,
4522 "ev.outcome must be Success on allow"
4523 );
4524 assert_eq!(
4525 ev.namespace, "test-ns",
4526 "ev.namespace must match the dispatch namespace"
4527 );
4528
4529 let data = &ev.payload;
4531
4532 let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4533 .expect("ev.payload must deserialize to AuditEvent");
4534
4535 assert_eq!(
4536 audit.decision,
4537 khive_gate::AuditDecision::Allow,
4538 "AuditEvent.decision must be Allow"
4539 );
4540 assert_eq!(audit.verb, "create", "AuditEvent.verb must be 'create'");
4541 assert_eq!(
4542 audit.namespace, "test-ns",
4543 "AuditEvent.namespace must be preserved"
4544 );
4545 assert_eq!(
4546 audit.gate_impl, "AllowAllGate",
4547 "AuditEvent.gate_impl must name the gate implementation"
4548 );
4549 assert!(
4550 audit.deny_reason.is_none(),
4551 "AuditEvent.deny_reason must be None on Allow"
4552 );
4553 let payload_json: serde_json::Value =
4555 serde_json::from_value(data.clone()).expect("data must be valid JSON");
4556 assert_eq!(
4557 payload_json["obligations"],
4558 serde_json::Value::Array(Vec::new()),
4559 "obligations must be [] on AllowAllGate"
4560 );
4561 }
4562
4563 struct EmbeddingAwarePack {
4570 models: Vec<String>,
4571 }
4572
4573 impl khive_types::Pack for EmbeddingAwarePack {
4574 const NAME: &'static str = "embedding_aware";
4575 const NOTE_KINDS: &'static [&'static str] = &[];
4576 const ENTITY_KINDS: &'static [&'static str] = &["widget"];
4577 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
4578 name: "create",
4579 description: "create a widget (embedding-aware stub)",
4580 visibility: Visibility::Verb,
4581 category: VerbCategory::Commissive,
4582 params: &[],
4583 }];
4584 }
4585
4586 #[async_trait]
4587 impl PackRuntime for EmbeddingAwarePack {
4588 fn name(&self) -> &str {
4589 Self::NAME
4590 }
4591 fn note_kinds(&self) -> &'static [&'static str] {
4592 Self::NOTE_KINDS
4593 }
4594 fn entity_kinds(&self) -> &'static [&'static str] {
4595 Self::ENTITY_KINDS
4596 }
4597 fn handlers(&self) -> &'static [HandlerDef] {
4598 Self::HANDLERS
4599 }
4600 fn registered_embedding_model_names(&self) -> Vec<String> {
4601 self.models.clone()
4602 }
4603 async fn dispatch(
4604 &self,
4605 verb: &str,
4606 _params: Value,
4607 _registry: &VerbRegistry,
4608 _token: &NamespaceToken,
4609 ) -> Result<Value, RuntimeError> {
4610 Ok(serde_json::json!({ "pack": "embedding_aware", "verb": verb }))
4611 }
4612 }
4613
4614 struct FailingProbePack;
4617
4618 impl khive_types::Pack for FailingProbePack {
4619 const NAME: &'static str = "failing_probe";
4620 const NOTE_KINDS: &'static [&'static str] = &[];
4621 const ENTITY_KINDS: &'static [&'static str] = &[];
4622 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
4623 name: "probe",
4624 description: "always fails",
4625 visibility: Visibility::Verb,
4626 category: VerbCategory::Assertive,
4627 params: &[],
4628 }];
4629 }
4630
4631 #[async_trait]
4632 impl PackRuntime for FailingProbePack {
4633 fn name(&self) -> &str {
4634 Self::NAME
4635 }
4636 fn note_kinds(&self) -> &'static [&'static str] {
4637 Self::NOTE_KINDS
4638 }
4639 fn entity_kinds(&self) -> &'static [&'static str] {
4640 Self::ENTITY_KINDS
4641 }
4642 fn handlers(&self) -> &'static [HandlerDef] {
4643 Self::HANDLERS
4644 }
4645 async fn dispatch(
4646 &self,
4647 _verb: &str,
4648 _params: Value,
4649 _registry: &VerbRegistry,
4650 _token: &NamespaceToken,
4651 ) -> Result<Value, RuntimeError> {
4652 Err(RuntimeError::InvalidInput("boom".into()))
4653 }
4654 }
4655
4656 #[tokio::test]
4657 async fn resource_cost_unit_present_on_non_embedding_successful_dispatch() {
4658 let store = Arc::new(MemoryEventStore::default());
4659 let mut builder = VerbRegistryBuilder::new();
4660 builder.register(AlphaPack);
4661 builder.with_event_store(store.clone());
4662 let reg = builder.build().expect("registry builds");
4663
4664 reg.dispatch("list", serde_json::json!({})).await.unwrap();
4665
4666 let page = store
4667 .query_events(
4668 EventFilter::default(),
4669 PageRequest {
4670 limit: 10,
4671 offset: 0,
4672 },
4673 )
4674 .await
4675 .unwrap();
4676 assert_eq!(page.items.len(), 1);
4677 assert_eq!(
4678 page.items[0].payload["resource"],
4679 serde_json::json!({"work_class": "interactive", "cost_unit": 1}),
4680 "non-embedding-bearing verb's resource.cost_unit must be base_weight(verb) alone"
4681 );
4682 }
4683
4684 #[tokio::test]
4685 async fn resource_cost_unit_scales_with_registered_model_count_for_create() {
4686 let store = Arc::new(MemoryEventStore::default());
4687 let mut builder = VerbRegistryBuilder::new();
4688 builder.register(EmbeddingAwarePack {
4689 models: vec!["all-minilm-l6-v2".into(), "paraphrase".into()],
4690 });
4691 builder.with_event_store(store.clone());
4692 let reg = builder.build().expect("registry builds");
4693
4694 reg.dispatch("create", serde_json::json!({"kind": "widget"}))
4695 .await
4696 .unwrap();
4697
4698 let page = store
4699 .query_events(
4700 EventFilter::default(),
4701 PageRequest {
4702 limit: 10,
4703 offset: 0,
4704 },
4705 )
4706 .await
4707 .unwrap();
4708 assert_eq!(
4710 page.items[0].payload["resource"],
4711 serde_json::json!({"work_class": "interactive", "cost_unit": 3}),
4712 );
4713 }
4714
4715 #[tokio::test]
4716 async fn resource_cost_unit_zero_registered_models_is_base_weight_only() {
4717 let store = Arc::new(MemoryEventStore::default());
4718 let mut builder = VerbRegistryBuilder::new();
4719 builder.register(EmbeddingAwarePack { models: vec![] });
4720 builder.with_event_store(store.clone());
4721 let reg = builder.build().expect("registry builds");
4722
4723 reg.dispatch("create", serde_json::json!({"kind": "widget"}))
4724 .await
4725 .unwrap();
4726
4727 let page = store
4728 .query_events(
4729 EventFilter::default(),
4730 PageRequest {
4731 limit: 10,
4732 offset: 0,
4733 },
4734 )
4735 .await
4736 .unwrap();
4737 assert_eq!(
4738 page.items[0].payload["resource"]["cost_unit"], 1,
4739 "zero registered embedding models must vanish the term, not error or omit"
4740 );
4741 }
4742
4743 #[tokio::test]
4744 async fn resource_work_class_present_cost_unit_absent_when_dispatch_returns_error() {
4745 let store = Arc::new(MemoryEventStore::default());
4746 let mut builder = VerbRegistryBuilder::new();
4747 builder.register(FailingProbePack);
4748 builder.with_event_store(store.clone());
4749 let reg = builder.build().expect("registry builds");
4750
4751 let err = reg
4752 .dispatch("probe", serde_json::json!({}))
4753 .await
4754 .unwrap_err();
4755 assert!(matches!(err, RuntimeError::InvalidInput(_)));
4756
4757 let page = store
4758 .query_events(
4759 EventFilter::default(),
4760 PageRequest {
4761 limit: 10,
4762 offset: 0,
4763 },
4764 )
4765 .await
4766 .unwrap();
4767 assert_eq!(page.items.len(), 1);
4768 assert_eq!(page.items[0].outcome, EventOutcome::Error);
4769 assert_eq!(
4774 page.items[0].payload["resource"],
4775 serde_json::json!({"work_class": "interactive"}),
4776 "resource must carry work_class with cost_unit OMITTED (never 0) on an \
4777 errored dispatch: {:?}",
4778 page.items[0].payload
4779 );
4780 }
4781
4782 #[tokio::test]
4783 async fn resource_work_class_present_cost_unit_absent_when_no_pack_owns_the_verb() {
4784 let store = Arc::new(MemoryEventStore::default());
4785 let mut builder = VerbRegistryBuilder::new();
4786 builder.register(AlphaPack);
4787 builder.with_event_store(store.clone());
4788 let reg = builder.build().expect("registry builds");
4789
4790 let _ = reg
4791 .dispatch("no_such_verb_resource_test", serde_json::json!({}))
4792 .await;
4793
4794 let page = store
4795 .query_events(
4796 EventFilter::default(),
4797 PageRequest {
4798 limit: 10,
4799 offset: 0,
4800 },
4801 )
4802 .await
4803 .unwrap();
4804 assert_eq!(page.items.len(), 1);
4805 assert_eq!(
4806 page.items[0].payload["resource"],
4807 serde_json::json!({"work_class": "interactive"})
4808 );
4809 }
4810
4811 #[tokio::test]
4812 async fn resource_work_class_present_cost_unit_absent_on_denied_dispatch() {
4813 #[derive(Debug)]
4814 struct AlwaysDenyGate;
4815 impl Gate for AlwaysDenyGate {
4816 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4817 Ok(GateDecision::deny("test: always deny"))
4818 }
4819 }
4820 let store = Arc::new(MemoryEventStore::default());
4821 let mut builder = VerbRegistryBuilder::new();
4822 builder.register(AlphaPack);
4823 builder.with_gate(Arc::new(AlwaysDenyGate));
4824 builder.with_event_store(store.clone());
4825 let reg = builder.build().expect("registry builds");
4826
4827 let _ = reg.dispatch("list", serde_json::json!({})).await;
4828
4829 let page = store
4830 .query_events(
4831 EventFilter::default(),
4832 PageRequest {
4833 limit: 10,
4834 offset: 0,
4835 },
4836 )
4837 .await
4838 .unwrap();
4839 assert_eq!(page.items.len(), 1);
4840 assert_eq!(page.items[0].outcome, EventOutcome::Denied);
4841 assert_eq!(
4842 page.items[0].payload["resource"],
4843 serde_json::json!({"work_class": "interactive"})
4844 );
4845 }
4846
4847 #[tokio::test]
4848 async fn resource_cost_unit_present_on_link_singleton_success() {
4849 let store = Arc::new(MemoryEventStore::default());
4850 let edge_id = uuid::Uuid::new_v4();
4851 let source_id = uuid::Uuid::new_v4();
4852 let target_id = uuid::Uuid::new_v4();
4853 let edge_json = serde_json::json!({
4854 "id": edge_id,
4855 "namespace": "local",
4856 "source_id": source_id,
4857 "target_id": target_id,
4858 "relation": "depends_on",
4859 "weight": 1.0,
4860 });
4861 let mut builder = VerbRegistryBuilder::new();
4862 builder.register(LinkResultPack::ok(edge_json));
4863 builder.with_event_store(store.clone());
4864 let reg = builder.build().expect("registry builds");
4865
4866 reg.dispatch(
4867 "link",
4868 serde_json::json!({
4869 "source_id": source_id,
4870 "target_id": target_id,
4871 "relation": "depends_on",
4872 }),
4873 )
4874 .await
4875 .unwrap();
4876
4877 let page = store
4878 .query_events(
4879 EventFilter::default(),
4880 PageRequest {
4881 limit: 10,
4882 offset: 0,
4883 },
4884 )
4885 .await
4886 .unwrap();
4887 assert_eq!(
4888 page.items[0].payload["resource"],
4889 serde_json::json!({"work_class": "interactive", "cost_unit": 1}),
4890 "link has no embedding-bearing path -> base_weight(link) alone, even on the v2-enriched singleton path"
4891 );
4892 }
4893
4894 #[tokio::test]
4895 async fn resource_work_class_present_cost_unit_absent_on_link_dispatch_failure() {
4896 let store = Arc::new(MemoryEventStore::default());
4897 let mut builder = VerbRegistryBuilder::new();
4898 builder.register(LinkResultPack::err("target endpoint not found"));
4899 builder.with_event_store(store.clone());
4900 let reg = builder.build().expect("registry builds");
4901
4902 let _ = reg
4903 .dispatch(
4904 "link",
4905 serde_json::json!({
4906 "source_id": "note:alpha",
4907 "target_id": "note:missing",
4908 "relation": "depends_on",
4909 }),
4910 )
4911 .await;
4912
4913 let page = store
4914 .query_events(
4915 EventFilter::default(),
4916 PageRequest {
4917 limit: 10,
4918 offset: 0,
4919 },
4920 )
4921 .await
4922 .unwrap();
4923 assert_eq!(page.items.len(), 1);
4924 assert_eq!(
4925 page.items[0].payload["resource"],
4926 serde_json::json!({"work_class": "interactive"})
4927 );
4928 }
4929
4930 #[tokio::test]
4932 async fn audit_event_threads_target_id_from_dispatch_args() {
4933 let store = Arc::new(MemoryEventStore::default());
4934 let target = uuid::Uuid::new_v4();
4935 let mut builder = VerbRegistryBuilder::new();
4936 builder.register(AlphaPack);
4937 builder.with_event_store(store.clone());
4938 builder.with_default_namespace("test-ns");
4939 let reg = builder.build().expect("registry builds");
4940
4941 reg.dispatch(
4942 "create",
4943 serde_json::json!({"namespace": "test-ns", "target_id": target}),
4944 )
4945 .await
4946 .unwrap();
4947
4948 let page = store
4949 .query_events(
4950 EventFilter::default(),
4951 PageRequest {
4952 offset: 0,
4953 limit: 10,
4954 },
4955 )
4956 .await
4957 .unwrap();
4958 assert_eq!(
4959 page.items[0].target_id,
4960 Some(target),
4961 "#282: audit event must carry target_id from dispatch params"
4962 );
4963 }
4964
4965 struct LinkResultPack {
4971 result: std::sync::Mutex<Option<Result<Value, RuntimeError>>>,
4972 }
4973
4974 impl LinkResultPack {
4975 fn ok(value: Value) -> Self {
4976 Self {
4977 result: std::sync::Mutex::new(Some(Ok(value))),
4978 }
4979 }
4980 fn err(message: &str) -> Self {
4981 Self {
4982 result: std::sync::Mutex::new(Some(Err(RuntimeError::InvalidInput(
4983 message.to_string(),
4984 )))),
4985 }
4986 }
4987 }
4988
4989 impl khive_types::Pack for LinkResultPack {
4990 const NAME: &'static str = "kg";
4991 const NOTE_KINDS: &'static [&'static str] = &[];
4992 const ENTITY_KINDS: &'static [&'static str] = &[];
4993 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
4994 name: "link",
4995 description: "test link handler",
4996 visibility: Visibility::Verb,
4997 category: VerbCategory::Commissive,
4998 params: &[],
4999 }];
5000 }
5001
5002 #[async_trait]
5003 impl PackRuntime for LinkResultPack {
5004 fn name(&self) -> &str {
5005 Self::NAME
5006 }
5007 fn note_kinds(&self) -> &'static [&'static str] {
5008 Self::NOTE_KINDS
5009 }
5010 fn entity_kinds(&self) -> &'static [&'static str] {
5011 Self::ENTITY_KINDS
5012 }
5013 fn handlers(&self) -> &'static [HandlerDef] {
5014 Self::HANDLERS
5015 }
5016 async fn dispatch(
5017 &self,
5018 _verb: &str,
5019 _params: Value,
5020 _registry: &VerbRegistry,
5021 _token: &NamespaceToken,
5022 ) -> Result<Value, RuntimeError> {
5023 self.result
5024 .lock()
5025 .unwrap()
5026 .take()
5027 .expect("LinkResultPack dispatch called more than once in a test")
5028 }
5029 }
5030
5031 #[tokio::test]
5032 async fn link_audit_enriches_successful_singleton_with_edge_v2() {
5033 let store = Arc::new(MemoryEventStore::default());
5034 let edge_id = uuid::Uuid::new_v4();
5035 let source_id = uuid::Uuid::new_v4();
5036 let target_id = uuid::Uuid::new_v4();
5037 let edge_json = serde_json::json!({
5038 "id": edge_id,
5039 "namespace": "local",
5040 "source_id": source_id,
5041 "target_id": target_id,
5042 "relation": "depends_on",
5043 "weight": 1.0,
5044 });
5045 let mut builder = VerbRegistryBuilder::new();
5046 builder.register(LinkResultPack::ok(edge_json));
5047 builder.with_event_store(store.clone());
5048 builder.with_default_namespace("test-ns");
5049 let reg = builder.build().expect("registry builds");
5050
5051 reg.dispatch(
5052 "link",
5053 serde_json::json!({
5054 "source_id": source_id,
5055 "target_id": target_id,
5056 "relation": "depends_on",
5057 }),
5058 )
5059 .await
5060 .unwrap();
5061
5062 let count = store.count_events(EventFilter::default()).await.unwrap();
5063 assert_eq!(
5064 count, 1,
5065 "exactly one deferred audit row must be persisted for a successful singleton link"
5066 );
5067 let page = store
5068 .query_events(
5069 EventFilter::default(),
5070 PageRequest {
5071 limit: 10,
5072 offset: 0,
5073 },
5074 )
5075 .await
5076 .unwrap();
5077 let ev = &page.items[0];
5078 assert_eq!(ev.verb, "link");
5079 assert_eq!(ev.outcome, EventOutcome::Success);
5080 assert_eq!(
5081 ev.payload_schema_version, 2,
5082 "successful singleton link uses audit schema v2"
5083 );
5084 assert_eq!(
5085 ev.target_id,
5086 Some(edge_id),
5087 "target_id must be the created/resolved edge id, not a raw caller arg"
5088 );
5089 assert_eq!(ev.payload["edge_id"], serde_json::json!(edge_id));
5090 assert_eq!(ev.payload["source_id"], serde_json::json!(source_id));
5091 assert_eq!(ev.payload["target_id"], serde_json::json!(target_id));
5092 assert_eq!(ev.payload["relation"], "depends_on");
5093 assert_eq!(ev.payload["weight"], 1.0);
5094 assert_eq!(ev.payload["verb"], "link");
5096 assert_eq!(ev.payload["decision"], "allow");
5097 assert!(ev.payload.get("gate_impl").is_some());
5098 }
5099
5100 #[tokio::test]
5101 async fn link_audit_falls_back_to_v1_when_dispatch_fails() {
5102 let store = Arc::new(MemoryEventStore::default());
5103 let mut builder = VerbRegistryBuilder::new();
5104 builder.register(LinkResultPack::err("target endpoint not found"));
5105 builder.with_event_store(store.clone());
5106 builder.with_default_namespace("test-ns");
5107 let reg = builder.build().expect("registry builds");
5108
5109 let err = reg
5110 .dispatch(
5111 "link",
5112 serde_json::json!({
5113 "source_id": "note:alpha",
5114 "target_id": "note:missing",
5115 "relation": "depends_on",
5116 }),
5117 )
5118 .await
5119 .unwrap_err();
5120 assert!(
5121 matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("not found")),
5122 "the original dispatch error must be returned unchanged"
5123 );
5124
5125 let page = store
5126 .query_events(
5127 EventFilter::default(),
5128 PageRequest {
5129 limit: 10,
5130 offset: 0,
5131 },
5132 )
5133 .await
5134 .unwrap();
5135 assert_eq!(
5136 page.items.len(),
5137 1,
5138 "a v1 fallback audit row must still be persisted on dispatch failure"
5139 );
5140 let ev = &page.items[0];
5141 assert_eq!(
5142 ev.payload_schema_version, 1,
5143 "failed link keeps the v1 audit shape"
5144 );
5145 assert_eq!(
5148 ev.outcome,
5149 EventOutcome::Error,
5150 "outcome reflects the dispatch result (Err), not the gate decision (Allow)"
5151 );
5152 assert!(
5153 ev.duration_us >= 0,
5154 "duration_us must still be populated (measured, not the Event::new \
5155 default sentinel) on a failed dispatch"
5156 );
5157 assert!(
5158 ev.target_id.is_none(),
5159 "non-UUID caller-supplied ids do not spuriously populate target_id"
5160 );
5161 assert!(
5162 ev.payload.get("edge_id").is_none(),
5163 "v1 fallback must not carry edge enrichment fields"
5164 );
5165 let _: khive_gate::AuditEvent = serde_json::from_value(ev.payload.clone())
5166 .expect("v1 fallback payload must deserialize as AuditEvent");
5167 }
5168
5169 #[tokio::test]
5170 async fn link_audit_falls_back_to_v1_when_result_missing_edge_fields() {
5171 let store = Arc::new(MemoryEventStore::default());
5172 let target_arg = uuid::Uuid::new_v4();
5173 let mut builder = VerbRegistryBuilder::new();
5174 builder.register(LinkResultPack::ok(serde_json::json!({"ok": true})));
5175 builder.with_event_store(store.clone());
5176 builder.with_default_namespace("test-ns");
5177 let reg = builder.build().expect("registry builds");
5178
5179 reg.dispatch(
5180 "link",
5181 serde_json::json!({
5182 "source_id": uuid::Uuid::new_v4(),
5183 "target_id": target_arg,
5184 "relation": "depends_on",
5185 }),
5186 )
5187 .await
5188 .unwrap();
5189
5190 let page = store
5191 .query_events(
5192 EventFilter::default(),
5193 PageRequest {
5194 limit: 10,
5195 offset: 0,
5196 },
5197 )
5198 .await
5199 .unwrap();
5200 assert_eq!(page.items.len(), 1);
5201 let ev = &page.items[0];
5202 assert_eq!(
5203 ev.payload_schema_version, 1,
5204 "an unparsable success result falls back to v1 rather than dropping the audit row"
5205 );
5206 assert_eq!(ev.outcome, EventOutcome::Success);
5207 assert_eq!(
5208 ev.target_id,
5209 Some(target_arg),
5210 "v1 fallback still extracts target_id from the raw dispatch args"
5211 );
5212 assert!(ev.payload.get("edge_id").is_none());
5213 }
5214
5215 #[tokio::test]
5216 async fn link_audit_bulk_links_get_no_enrichment() {
5217 let store = Arc::new(MemoryEventStore::default());
5218 let mut builder = VerbRegistryBuilder::new();
5219 builder.register(LinkResultPack::ok(serde_json::json!({
5220 "attempted": 2, "created": 2, "skipped": 0, "failed": 0
5221 })));
5222 builder.with_event_store(store.clone());
5223 builder.with_default_namespace("test-ns");
5224 let reg = builder.build().expect("registry builds");
5225
5226 reg.dispatch(
5227 "link",
5228 serde_json::json!({
5229 "links": [
5230 {"source_id": "a", "target_id": "b", "relation": "depends_on"},
5231 {"source_id": "c", "target_id": "d", "relation": "depends_on"},
5232 ],
5233 }),
5234 )
5235 .await
5236 .unwrap();
5237
5238 let count = store.count_events(EventFilter::default()).await.unwrap();
5239 assert_eq!(
5240 count, 1,
5241 "bulk `links` gets exactly one v1 audit row (deferred until dispatch \
5242 resolves like every other Allow-outcome row since ADR-103 Stage 1, \
5243 but never v2-enriched — enrichment is singleton-`link`-only)"
5244 );
5245 let page = store
5246 .query_events(
5247 EventFilter::default(),
5248 PageRequest {
5249 limit: 10,
5250 offset: 0,
5251 },
5252 )
5253 .await
5254 .unwrap();
5255 let ev = &page.items[0];
5256 assert_eq!(
5257 ev.payload_schema_version, 1,
5258 "bulk link mode is out of scope for #676's events.target_id enrichment"
5259 );
5260 assert!(ev.target_id.is_none());
5261 }
5262
5263 #[test]
5264 fn link_audit_success_from_result_extracts_edge_fields() {
5265 let gate_req = GateRequest::new(
5266 ActorRef::anonymous(),
5267 Namespace::local(),
5268 "link",
5269 serde_json::json!({}),
5270 );
5271 let decision = GateDecision::Allow {
5272 obligations: vec![],
5273 };
5274 let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
5275
5276 let edge_id = uuid::Uuid::new_v4();
5277 let source_id = uuid::Uuid::new_v4();
5278 let target_id = uuid::Uuid::new_v4();
5279 let result = serde_json::json!({
5280 "id": edge_id,
5281 "source_id": source_id,
5282 "target_id": target_id,
5283 "relation": "depends_on",
5284 "weight": 0.5,
5285 });
5286
5287 let (returned_id, payload) = link_audit_success_from_result(audit, &result)
5288 .expect("well-formed edge JSON must produce an enriched payload");
5289 assert_eq!(returned_id, edge_id);
5290 assert_eq!(payload["edge_id"], serde_json::json!(edge_id));
5291 assert_eq!(payload["relation"], "depends_on");
5292 assert_eq!(payload["weight"], 0.5);
5293 assert_eq!(
5294 payload["verb"], "link",
5295 "v1 AuditEvent fields must flatten into the v2 payload"
5296 );
5297 }
5298
5299 #[test]
5300 fn link_audit_success_from_result_rejects_incomplete_or_malformed_result() {
5301 let gate_req = GateRequest::new(
5302 ActorRef::anonymous(),
5303 Namespace::local(),
5304 "link",
5305 serde_json::json!({}),
5306 );
5307 let decision = GateDecision::Allow {
5308 obligations: vec![],
5309 };
5310 let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
5311
5312 assert!(
5313 link_audit_success_from_result(
5314 audit.clone(),
5315 &serde_json::json!({"id": uuid::Uuid::new_v4()}),
5316 )
5317 .is_none(),
5318 "missing source_id/target_id/relation/weight must not enrich"
5319 );
5320 assert!(
5321 link_audit_success_from_result(audit, &serde_json::json!({"id": "not-a-uuid"}))
5322 .is_none(),
5323 "a non-UUID id must not enrich"
5324 );
5325 }
5326
5327 async fn first_event(store: &Arc<MemoryEventStore>) -> Event {
5337 let page = store
5338 .query_events(
5339 EventFilter::default(),
5340 PageRequest {
5341 limit: 10,
5342 offset: 0,
5343 },
5344 )
5345 .await
5346 .unwrap();
5347 assert_eq!(
5348 page.items.len(),
5349 1,
5350 "expected exactly one persisted audit event"
5351 );
5352 page.items[0].clone()
5353 }
5354
5355 #[tokio::test]
5356 async fn dispatch_with_identity_stamps_request_id_on_success() {
5357 let store = Arc::new(MemoryEventStore::default());
5358 let mut builder = VerbRegistryBuilder::new();
5359 builder.register(AlphaPack);
5360 builder.with_event_store(store.clone());
5361 let reg = builder.build().expect("registry builds");
5362
5363 reg.dispatch_with_identity(
5364 "list",
5365 serde_json::json!({"namespace": "test-ns"}),
5366 Some(RequestIdentity {
5367 request_id: Some(101),
5368 ..Default::default()
5369 }),
5370 )
5371 .await
5372 .unwrap();
5373
5374 let ev = first_event(&store).await;
5375 assert_eq!(ev.outcome, EventOutcome::Success);
5376 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(101));
5377 }
5378
5379 #[tokio::test]
5380 async fn dispatch_with_identity_stamps_request_id_on_dispatch_error() {
5381 let store = Arc::new(MemoryEventStore::default());
5382 let mut builder = VerbRegistryBuilder::new();
5383 builder.register(FailingProbePack);
5384 builder.with_event_store(store.clone());
5385 let reg = builder.build().expect("registry builds");
5386
5387 let err = reg
5388 .dispatch_with_identity(
5389 "probe",
5390 serde_json::json!({"namespace": "test-ns"}),
5391 Some(RequestIdentity {
5392 request_id: Some(102),
5393 ..Default::default()
5394 }),
5395 )
5396 .await
5397 .unwrap_err();
5398 assert!(matches!(err, RuntimeError::InvalidInput(_)));
5399
5400 let ev = first_event(&store).await;
5401 assert_eq!(ev.outcome, EventOutcome::Error);
5402 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(102));
5403 }
5404
5405 #[tokio::test]
5406 async fn dispatch_with_identity_stamps_request_id_on_denied() {
5407 #[derive(Debug)]
5408 struct AlwaysDenyGate;
5409 impl Gate for AlwaysDenyGate {
5410 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
5411 Ok(GateDecision::deny("denied by test"))
5412 }
5413 }
5414
5415 let store = Arc::new(MemoryEventStore::default());
5416 let mut builder = VerbRegistryBuilder::new();
5417 builder.register(AlphaPack);
5418 builder.with_gate(Arc::new(AlwaysDenyGate));
5419 builder.with_event_store(store.clone());
5420 let reg = builder.build().expect("registry builds");
5421
5422 let err = reg
5423 .dispatch_with_identity(
5424 "list",
5425 serde_json::json!({"namespace": "test-ns"}),
5426 Some(RequestIdentity {
5427 request_id: Some(103),
5428 ..Default::default()
5429 }),
5430 )
5431 .await
5432 .unwrap_err();
5433 assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
5434
5435 let ev = first_event(&store).await;
5436 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(103));
5437 }
5438
5439 #[tokio::test]
5440 async fn dispatch_with_identity_stamps_request_id_on_link_v2_success() {
5441 let store = Arc::new(MemoryEventStore::default());
5442 let edge_id = uuid::Uuid::new_v4();
5443 let source_id = uuid::Uuid::new_v4();
5444 let target_id = uuid::Uuid::new_v4();
5445 let edge_json = serde_json::json!({
5446 "id": edge_id,
5447 "namespace": "local",
5448 "source_id": source_id,
5449 "target_id": target_id,
5450 "relation": "depends_on",
5451 "weight": 1.0,
5452 });
5453 let mut builder = VerbRegistryBuilder::new();
5454 builder.register(LinkResultPack::ok(edge_json));
5455 builder.with_event_store(store.clone());
5456 builder.with_default_namespace("test-ns");
5457 let reg = builder.build().expect("registry builds");
5458
5459 reg.dispatch_with_identity(
5460 "link",
5461 serde_json::json!({
5462 "source_id": source_id,
5463 "target_id": target_id,
5464 "relation": "depends_on",
5465 }),
5466 Some(RequestIdentity {
5467 namespace: "test-ns".to_string(),
5468 request_id: Some(104),
5469 ..Default::default()
5470 }),
5471 )
5472 .await
5473 .unwrap();
5474
5475 let ev = first_event(&store).await;
5476 assert_eq!(
5477 ev.payload_schema_version, 2,
5478 "successful singleton link uses audit schema v2"
5479 );
5480 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(104));
5481 }
5482
5483 #[tokio::test]
5484 async fn dispatch_with_identity_stamps_request_id_on_link_v1_fallback() {
5485 let store = Arc::new(MemoryEventStore::default());
5486 let mut builder = VerbRegistryBuilder::new();
5487 builder.register(LinkResultPack::err("target endpoint not found"));
5488 builder.with_event_store(store.clone());
5489 builder.with_default_namespace("test-ns");
5490 let reg = builder.build().expect("registry builds");
5491
5492 let err = reg
5493 .dispatch_with_identity(
5494 "link",
5495 serde_json::json!({
5496 "source_id": "note:alpha",
5497 "target_id": "note:missing",
5498 "relation": "depends_on",
5499 }),
5500 Some(RequestIdentity {
5501 namespace: "test-ns".to_string(),
5502 request_id: Some(105),
5503 ..Default::default()
5504 }),
5505 )
5506 .await
5507 .unwrap_err();
5508 assert!(matches!(err, RuntimeError::InvalidInput(_)));
5509
5510 let ev = first_event(&store).await;
5511 assert_eq!(
5512 ev.payload_schema_version, 1,
5513 "failed link keeps the v1 audit shape"
5514 );
5515 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(105));
5516 }
5517
5518 #[tokio::test]
5519 async fn dispatch_with_identity_stamps_request_id_on_unknown_verb() {
5520 let store = Arc::new(MemoryEventStore::default());
5521 let mut builder = VerbRegistryBuilder::new();
5522 builder.register(AlphaPack);
5523 builder.with_event_store(store.clone());
5524 let reg = builder.build().expect("registry builds");
5525
5526 let err = reg
5527 .dispatch_with_identity(
5528 "no_such_verb",
5529 serde_json::json!({}),
5530 Some(RequestIdentity {
5531 namespace: Namespace::local().as_str().to_string(),
5532 request_id: Some(106),
5533 ..Default::default()
5534 }),
5535 )
5536 .await
5537 .unwrap_err();
5538 assert!(matches!(err, RuntimeError::InvalidInput(_)));
5539
5540 let ev = first_event(&store).await;
5541 assert_eq!(ev.outcome, EventOutcome::Error);
5542 assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(106));
5543 }
5544
5545 #[tokio::test]
5546 async fn dispatch_with_identity_omits_request_id_key_when_absent() {
5547 let store = Arc::new(MemoryEventStore::default());
5548 let mut builder = VerbRegistryBuilder::new();
5549 builder.register(AlphaPack);
5550 builder.with_event_store(store.clone());
5551 let reg = builder.build().expect("registry builds");
5552
5553 reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
5555 .await
5556 .unwrap();
5557
5558 let ev = first_event(&store).await;
5559 let resource = ev.payload["resource"]
5560 .as_object()
5561 .expect("resource must be an object");
5562 assert!(
5563 !resource.contains_key("request_id"),
5564 "request_id key must be entirely absent when no id is supplied, \
5565 not present as null or 0: got {resource:?}"
5566 );
5567 }
5568}
5569
5570#[cfg(test)]
5573mod dep_tests {
5574 use super::*;
5575 use async_trait::async_trait;
5576 use khive_types::Pack;
5577 use serde_json::Value;
5578
5579 struct KgDepPack;
5580 struct MemoryDepPack;
5581 struct ADepPack;
5582 struct BDepPack;
5583
5584 impl Pack for KgDepPack {
5585 const NAME: &'static str = "kg_dep";
5586 const NOTE_KINDS: &'static [&'static str] = &["observation"];
5587 const ENTITY_KINDS: &'static [&'static str] = &["concept"];
5588 const HANDLERS: &'static [HandlerDef] = &[];
5589 }
5590
5591 impl Pack for MemoryDepPack {
5592 const NAME: &'static str = "memory_dep";
5593 const NOTE_KINDS: &'static [&'static str] = &["memory"];
5594 const ENTITY_KINDS: &'static [&'static str] = &[];
5595 const HANDLERS: &'static [HandlerDef] = &[];
5596 const REQUIRES: &'static [&'static str] = &["kg_dep"];
5597 }
5598
5599 impl Pack for ADepPack {
5600 const NAME: &'static str = "pack_a";
5601 const NOTE_KINDS: &'static [&'static str] = &[];
5602 const ENTITY_KINDS: &'static [&'static str] = &[];
5603 const HANDLERS: &'static [HandlerDef] = &[];
5604 const REQUIRES: &'static [&'static str] = &["pack_b"];
5605 }
5606
5607 impl Pack for BDepPack {
5608 const NAME: &'static str = "pack_b";
5609 const NOTE_KINDS: &'static [&'static str] = &[];
5610 const ENTITY_KINDS: &'static [&'static str] = &[];
5611 const HANDLERS: &'static [HandlerDef] = &[];
5612 const REQUIRES: &'static [&'static str] = &["pack_a"];
5613 }
5614
5615 #[async_trait]
5616 impl PackRuntime for KgDepPack {
5617 fn name(&self) -> &str {
5618 Self::NAME
5619 }
5620 fn note_kinds(&self) -> &'static [&'static str] {
5621 Self::NOTE_KINDS
5622 }
5623 fn entity_kinds(&self) -> &'static [&'static str] {
5624 Self::ENTITY_KINDS
5625 }
5626 fn handlers(&self) -> &'static [HandlerDef] {
5627 Self::HANDLERS
5628 }
5629 async fn dispatch(
5630 &self,
5631 verb: &str,
5632 _: Value,
5633 _: &VerbRegistry,
5634 _: &NamespaceToken,
5635 ) -> Result<Value, RuntimeError> {
5636 Err(RuntimeError::InvalidInput(format!(
5637 "KgDepPack has no verbs: {verb}"
5638 )))
5639 }
5640 }
5641
5642 #[async_trait]
5643 impl PackRuntime for MemoryDepPack {
5644 fn name(&self) -> &str {
5645 Self::NAME
5646 }
5647 fn note_kinds(&self) -> &'static [&'static str] {
5648 Self::NOTE_KINDS
5649 }
5650 fn entity_kinds(&self) -> &'static [&'static str] {
5651 Self::ENTITY_KINDS
5652 }
5653 fn handlers(&self) -> &'static [HandlerDef] {
5654 Self::HANDLERS
5655 }
5656 fn requires(&self) -> &'static [&'static str] {
5657 Self::REQUIRES
5658 }
5659 async fn dispatch(
5660 &self,
5661 verb: &str,
5662 _: Value,
5663 _: &VerbRegistry,
5664 _: &NamespaceToken,
5665 ) -> Result<Value, RuntimeError> {
5666 Err(RuntimeError::InvalidInput(format!(
5667 "MemoryDepPack has no verbs: {verb}"
5668 )))
5669 }
5670 }
5671
5672 #[async_trait]
5673 impl PackRuntime for ADepPack {
5674 fn name(&self) -> &str {
5675 Self::NAME
5676 }
5677 fn note_kinds(&self) -> &'static [&'static str] {
5678 Self::NOTE_KINDS
5679 }
5680 fn entity_kinds(&self) -> &'static [&'static str] {
5681 Self::ENTITY_KINDS
5682 }
5683 fn handlers(&self) -> &'static [HandlerDef] {
5684 Self::HANDLERS
5685 }
5686 fn requires(&self) -> &'static [&'static str] {
5687 Self::REQUIRES
5688 }
5689 async fn dispatch(
5690 &self,
5691 verb: &str,
5692 _: Value,
5693 _: &VerbRegistry,
5694 _: &NamespaceToken,
5695 ) -> Result<Value, RuntimeError> {
5696 Err(RuntimeError::InvalidInput(format!(
5697 "ADepPack has no verbs: {verb}"
5698 )))
5699 }
5700 }
5701
5702 #[async_trait]
5703 impl PackRuntime for BDepPack {
5704 fn name(&self) -> &str {
5705 Self::NAME
5706 }
5707 fn note_kinds(&self) -> &'static [&'static str] {
5708 Self::NOTE_KINDS
5709 }
5710 fn entity_kinds(&self) -> &'static [&'static str] {
5711 Self::ENTITY_KINDS
5712 }
5713 fn handlers(&self) -> &'static [HandlerDef] {
5714 Self::HANDLERS
5715 }
5716 fn requires(&self) -> &'static [&'static str] {
5717 Self::REQUIRES
5718 }
5719 async fn dispatch(
5720 &self,
5721 verb: &str,
5722 _: Value,
5723 _: &VerbRegistry,
5724 _: &NamespaceToken,
5725 ) -> Result<Value, RuntimeError> {
5726 Err(RuntimeError::InvalidInput(format!(
5727 "BDepPack has no verbs: {verb}"
5728 )))
5729 }
5730 }
5731
5732 #[test]
5733 fn test_pack_deps_happy_path() {
5734 let mut builder = VerbRegistryBuilder::new();
5735 builder.register(MemoryDepPack);
5736 builder.register(KgDepPack);
5737 let reg = builder
5738 .build()
5739 .expect("kg_dep satisfies memory_dep dependency");
5740 assert_eq!(reg.pack_requires("memory_dep").unwrap(), &["kg_dep"]);
5741 let names = reg.pack_names();
5742 let kg_pos = names.iter().position(|&n| n == "kg_dep").unwrap();
5743 let mem_pos = names.iter().position(|&n| n == "memory_dep").unwrap();
5744 assert!(
5745 kg_pos < mem_pos,
5746 "kg_dep must be loaded before memory_dep; order: {names:?}"
5747 );
5748 }
5749
5750 #[test]
5751 fn test_pack_deps_missing() {
5752 let mut builder = VerbRegistryBuilder::new();
5753 builder.register(MemoryDepPack);
5754 let err = match builder.build() {
5755 Ok(_) => panic!("expected Err, got Ok"),
5756 Err(e) => e,
5757 };
5758 assert!(
5759 matches!(err, RuntimeError::MissingPackDependency(_)),
5760 "expected MissingPackDependency, got {err:?}"
5761 );
5762 let msg = err.to_string();
5763 assert!(
5764 msg.contains("memory_dep"),
5765 "error must name the dependent pack: {msg}"
5766 );
5767 assert!(
5768 msg.contains("kg_dep"),
5769 "error must name the missing dep: {msg}"
5770 );
5771 }
5772
5773 #[test]
5774 fn test_pack_deps_circular() {
5775 let mut builder = VerbRegistryBuilder::new();
5776 builder.register(ADepPack);
5777 builder.register(BDepPack);
5778 let err = match builder.build() {
5779 Ok(_) => panic!("expected Err, got Ok"),
5780 Err(e) => e,
5781 };
5782 assert!(
5783 matches!(err, RuntimeError::CircularPackDependency(_)),
5784 "expected CircularPackDependency, got {err:?}"
5785 );
5786 let msg = err.to_string();
5787 assert!(msg.contains("pack_a"), "error must name pack_a: {msg}");
5788 assert!(msg.contains("pack_b"), "error must name pack_b: {msg}");
5789 }
5790
5791 #[test]
5792 fn test_pack_deps_no_deps() {
5793 struct NoDepsA;
5794 struct NoDepsB;
5795
5796 impl Pack for NoDepsA {
5797 const NAME: &'static str = "no_deps_a";
5798 const NOTE_KINDS: &'static [&'static str] = &[];
5799 const ENTITY_KINDS: &'static [&'static str] = &[];
5800 const HANDLERS: &'static [HandlerDef] = &[];
5801 }
5802
5803 impl Pack for NoDepsB {
5804 const NAME: &'static str = "no_deps_b";
5805 const NOTE_KINDS: &'static [&'static str] = &[];
5806 const ENTITY_KINDS: &'static [&'static str] = &[];
5807 const HANDLERS: &'static [HandlerDef] = &[];
5808 }
5809
5810 #[async_trait]
5811 impl PackRuntime for NoDepsA {
5812 fn name(&self) -> &str {
5813 Self::NAME
5814 }
5815 fn note_kinds(&self) -> &'static [&'static str] {
5816 Self::NOTE_KINDS
5817 }
5818 fn entity_kinds(&self) -> &'static [&'static str] {
5819 Self::ENTITY_KINDS
5820 }
5821 fn handlers(&self) -> &'static [HandlerDef] {
5822 Self::HANDLERS
5823 }
5824 async fn dispatch(
5825 &self,
5826 verb: &str,
5827 _: Value,
5828 _: &VerbRegistry,
5829 _: &NamespaceToken,
5830 ) -> Result<Value, RuntimeError> {
5831 Err(RuntimeError::InvalidInput(format!("NoDepsA: {verb}")))
5832 }
5833 }
5834
5835 #[async_trait]
5836 impl PackRuntime for NoDepsB {
5837 fn name(&self) -> &str {
5838 Self::NAME
5839 }
5840 fn note_kinds(&self) -> &'static [&'static str] {
5841 Self::NOTE_KINDS
5842 }
5843 fn entity_kinds(&self) -> &'static [&'static str] {
5844 Self::ENTITY_KINDS
5845 }
5846 fn handlers(&self) -> &'static [HandlerDef] {
5847 Self::HANDLERS
5848 }
5849 async fn dispatch(
5850 &self,
5851 verb: &str,
5852 _: Value,
5853 _: &VerbRegistry,
5854 _: &NamespaceToken,
5855 ) -> Result<Value, RuntimeError> {
5856 Err(RuntimeError::InvalidInput(format!("NoDepsB: {verb}")))
5857 }
5858 }
5859
5860 let mut builder = VerbRegistryBuilder::new();
5861 builder.register(NoDepsA);
5862 builder.register(NoDepsB);
5863 let reg = builder.build().expect("packs with REQUIRES=&[] build");
5864 assert_eq!(reg.pack_requires("no_deps_a").unwrap(), &[] as &[&str]);
5865 assert_eq!(reg.pack_requires("no_deps_b").unwrap(), &[] as &[&str]);
5866 }
5867}
5868
5869#[cfg(test)]
5872mod hook_tests {
5873 use super::*;
5874 use async_trait::async_trait;
5875 use khive_types::Pack;
5876 use std::sync::atomic::{AtomicUsize, Ordering};
5877 use std::sync::Mutex as StdMutex;
5878
5879 struct SimplePack;
5880
5881 impl Pack for SimplePack {
5882 const NAME: &'static str = "simple";
5883 const NOTE_KINDS: &'static [&'static str] = &[];
5884 const ENTITY_KINDS: &'static [&'static str] = &[];
5885 const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
5886 name: "ping",
5887 description: "ping",
5888 visibility: Visibility::Verb,
5889 category: VerbCategory::Assertive,
5890 params: &[],
5891 }];
5892 }
5893
5894 #[async_trait]
5895 impl PackRuntime for SimplePack {
5896 fn name(&self) -> &str {
5897 SimplePack::NAME
5898 }
5899 fn note_kinds(&self) -> &'static [&'static str] {
5900 SimplePack::NOTE_KINDS
5901 }
5902 fn entity_kinds(&self) -> &'static [&'static str] {
5903 SimplePack::ENTITY_KINDS
5904 }
5905 fn handlers(&self) -> &'static [HandlerDef] {
5906 SimplePack::HANDLERS
5907 }
5908 async fn dispatch(
5909 &self,
5910 verb: &str,
5911 _params: Value,
5912 _registry: &VerbRegistry,
5913 _token: &NamespaceToken,
5914 ) -> Result<Value, RuntimeError> {
5915 Ok(serde_json::json!({ "verb": verb }))
5916 }
5917 }
5918
5919 #[derive(Default)]
5921 struct CountingHook {
5922 calls: AtomicUsize,
5923 last_verb: StdMutex<String>,
5924 }
5925
5926 #[async_trait]
5927 impl DispatchHook for CountingHook {
5928 async fn on_dispatch(&self, view: &EventView) {
5929 self.calls.fetch_add(1, Ordering::SeqCst);
5930 *self.last_verb.lock().unwrap() = view.event.verb.clone();
5931 }
5932 }
5933
5934 #[tokio::test]
5935 async fn dispatch_hook_fires_on_successful_dispatch() {
5936 let hook = Arc::new(CountingHook::default());
5937 let mut builder = VerbRegistryBuilder::new();
5938 builder.register(SimplePack);
5939 builder.with_dispatch_hook(hook.clone());
5940 let reg = builder.build().expect("registry builds");
5941
5942 reg.dispatch("ping", Value::Null).await.unwrap();
5943
5944 assert_eq!(
5945 hook.calls.load(Ordering::SeqCst),
5946 1,
5947 "hook must fire once per successful dispatch"
5948 );
5949 assert_eq!(
5950 hook.last_verb.lock().unwrap().as_str(),
5951 "ping",
5952 "hook event must carry the dispatched verb"
5953 );
5954 }
5955
5956 #[tokio::test]
5957 async fn dispatch_hook_fires_multiple_times() {
5958 let hook = Arc::new(CountingHook::default());
5959 let mut builder = VerbRegistryBuilder::new();
5960 builder.register(SimplePack);
5961 builder.with_dispatch_hook(hook.clone());
5962 let reg = builder.build().expect("registry builds");
5963
5964 reg.dispatch("ping", Value::Null).await.unwrap();
5965 reg.dispatch("ping", Value::Null).await.unwrap();
5966 reg.dispatch("ping", Value::Null).await.unwrap();
5967
5968 assert_eq!(
5969 hook.calls.load(Ordering::SeqCst),
5970 3,
5971 "hook must fire once per successful dispatch"
5972 );
5973 }
5974
5975 #[tokio::test]
5976 async fn dispatch_hook_does_not_fire_on_unknown_verb() {
5977 let hook = Arc::new(CountingHook::default());
5978 let mut builder = VerbRegistryBuilder::new();
5979 builder.register(SimplePack);
5980 builder.with_dispatch_hook(hook.clone());
5981 let reg = builder.build().expect("registry builds");
5982
5983 let _ = reg.dispatch("nonexistent", Value::Null).await;
5984
5985 assert_eq!(
5986 hook.calls.load(Ordering::SeqCst),
5987 0,
5988 "hook must NOT fire for unknown verb (dispatch returns error)"
5989 );
5990 }
5991
5992 #[tokio::test]
5993 async fn dispatch_hook_does_not_fire_on_gate_deny() {
5994 use khive_gate::{Gate, GateDecision, GateError};
5995
5996 #[derive(Debug)]
5997 struct AlwaysDenyGate;
5998 impl Gate for AlwaysDenyGate {
5999 fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
6000 Ok(GateDecision::deny("test deny"))
6001 }
6002 }
6003
6004 let hook = Arc::new(CountingHook::default());
6005 let mut builder = VerbRegistryBuilder::new();
6006 builder.register(SimplePack);
6007 builder.with_gate(Arc::new(AlwaysDenyGate));
6008 builder.with_dispatch_hook(hook.clone());
6009 let reg = builder.build().expect("registry builds");
6010
6011 let err = reg.dispatch("ping", Value::Null).await.unwrap_err();
6012 assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
6013
6014 assert_eq!(
6015 hook.calls.load(Ordering::SeqCst),
6016 0,
6017 "hook must NOT fire when gate denies dispatch"
6018 );
6019 }
6020
6021 #[tokio::test]
6022 async fn dispatch_hook_event_carries_namespace_from_params() {
6023 let hook = Arc::new(CountingHook::default());
6024
6025 #[derive(Default)]
6026 struct NsCapturingHook {
6027 ns: StdMutex<String>,
6028 }
6029
6030 #[async_trait]
6031 impl DispatchHook for NsCapturingHook {
6032 async fn on_dispatch(&self, view: &EventView) {
6033 *self.ns.lock().unwrap() = view.event.namespace.clone();
6034 }
6035 }
6036
6037 let ns_hook = Arc::new(NsCapturingHook::default());
6038 let mut builder = VerbRegistryBuilder::new();
6039 builder.register(SimplePack);
6040 builder.with_dispatch_hook(ns_hook.clone());
6041 let reg = builder.build().expect("registry builds");
6042
6043 reg.dispatch("ping", serde_json::json!({"namespace": "tenant-abc"}))
6044 .await
6045 .unwrap();
6046
6047 assert_eq!(
6048 ns_hook.ns.lock().unwrap().as_str(),
6049 "tenant-abc",
6050 "dispatch hook event must carry the resolved namespace"
6051 );
6052
6053 drop(hook);
6055 }
6056
6057 #[tokio::test]
6058 async fn no_dispatch_hook_configured_dispatch_succeeds() {
6059 let mut builder = VerbRegistryBuilder::new();
6061 builder.register(SimplePack);
6062 let reg = builder.build().expect("registry builds");
6064
6065 let res = reg.dispatch("ping", Value::Null).await.unwrap();
6066 assert_eq!(res["verb"], "ping");
6067 }
6068}
6069
6070#[cfg(test)]
6073mod help_tests {
6074 use super::*;
6075 use async_trait::async_trait;
6076 use khive_types::Pack;
6077 use std::sync::{
6078 atomic::{AtomicUsize, Ordering},
6079 Arc,
6080 };
6081
6082 static CREATE_PARAMS: [ParamDef; 2] = [
6087 ParamDef {
6088 name: "kind",
6089 param_type: "string",
6090 required: true,
6091 description: "Granular kind (concept | document | ...).",
6092 },
6093 ParamDef {
6094 name: "name",
6095 param_type: "string",
6096 required: false,
6097 description: "Human-readable name.",
6098 },
6099 ];
6100
6101 static RECALL_PARAMS: [ParamDef; 2] = [
6102 ParamDef {
6103 name: "query",
6104 param_type: "string",
6105 required: true,
6106 description: "Semantic recall query.",
6107 },
6108 ParamDef {
6109 name: "limit",
6110 param_type: "integer",
6111 required: false,
6112 description: "Maximum memories to return.",
6113 },
6114 ];
6115
6116 static EMBED_PARAMS: [ParamDef; 0] = [];
6119
6120 struct HelpPack {
6121 invocations: Arc<AtomicUsize>,
6122 }
6123
6124 impl Pack for HelpPack {
6125 const NAME: &'static str = "helptest";
6126 const NOTE_KINDS: &'static [&'static str] = &[];
6127 const ENTITY_KINDS: &'static [&'static str] = &[];
6128 const HANDLERS: &'static [HandlerDef] = &[
6129 HandlerDef {
6130 name: "create",
6131 description: "Create an entity or note",
6132 visibility: Visibility::Verb,
6133 category: VerbCategory::Commissive,
6134 params: &CREATE_PARAMS,
6135 },
6136 HandlerDef {
6137 name: "recall",
6138 description: "Recall memory notes with decay-aware hybrid ranking",
6139 visibility: Visibility::Verb,
6140 category: VerbCategory::Assertive,
6141 params: &RECALL_PARAMS,
6142 },
6143 HandlerDef {
6146 name: "recall.embed",
6147 description: "Return the embedding vector used by memory recall",
6148 visibility: Visibility::Subhandler,
6149 category: VerbCategory::Assertive,
6150 params: &EMBED_PARAMS,
6151 },
6152 ];
6153 }
6154
6155 #[async_trait]
6156 impl PackRuntime for HelpPack {
6157 fn name(&self) -> &str {
6158 HelpPack::NAME
6159 }
6160 fn note_kinds(&self) -> &'static [&'static str] {
6161 HelpPack::NOTE_KINDS
6162 }
6163 fn entity_kinds(&self) -> &'static [&'static str] {
6164 HelpPack::ENTITY_KINDS
6165 }
6166 fn handlers(&self) -> &'static [HandlerDef] {
6167 HelpPack::HANDLERS
6168 }
6169 async fn dispatch(
6170 &self,
6171 verb: &str,
6172 _params: Value,
6173 _registry: &VerbRegistry,
6174 _token: &NamespaceToken,
6175 ) -> Result<Value, RuntimeError> {
6176 self.invocations.fetch_add(1, Ordering::SeqCst);
6177 Ok(serde_json::json!({ "pack": "helptest", "verb": verb }))
6178 }
6179 }
6180
6181 fn build_help_registry(invocations: Arc<AtomicUsize>) -> VerbRegistry {
6182 let mut builder = VerbRegistryBuilder::new();
6183 builder.register(HelpPack { invocations });
6184 builder.build().expect("help registry builds")
6185 }
6186
6187 #[tokio::test]
6190 async fn test_help_true_returns_schema_for_kg_create() {
6191 let invocations = Arc::new(AtomicUsize::new(0));
6192 let reg = build_help_registry(invocations.clone());
6193
6194 let result = reg
6195 .dispatch("create", serde_json::json!({ "help": true }))
6196 .await
6197 .expect("help=true must succeed for a known verb");
6198
6199 assert_eq!(result["verb"], "create", "envelope must name the verb");
6201 assert_eq!(
6202 result["pack"], "helptest",
6203 "envelope must name the owning pack"
6204 );
6205 assert!(
6206 result["description"].as_str().is_some(),
6207 "description must be a string"
6208 );
6209
6210 let params = result["params"]
6212 .as_array()
6213 .expect("params must be a JSON array");
6214 assert!(!params.is_empty(), "params array must not be empty");
6215
6216 let kind_param = params.iter().find(|p| p["name"] == "kind");
6218 assert!(
6219 kind_param.is_some(),
6220 "params array must include the 'kind' parameter"
6221 );
6222 let kind_param = kind_param.unwrap();
6223 assert_eq!(
6224 kind_param["required"],
6225 serde_json::json!(true),
6226 "'kind' must be required"
6227 );
6228 assert_eq!(kind_param["type"], "string", "'kind' type must be 'string'");
6229 }
6230
6231 #[tokio::test]
6233 async fn test_help_true_returns_schema_for_recall() {
6234 let invocations = Arc::new(AtomicUsize::new(0));
6235 let reg = build_help_registry(invocations.clone());
6236
6237 let result = reg
6238 .dispatch("recall", serde_json::json!({ "help": true }))
6239 .await
6240 .expect("help=true must succeed for recall");
6241
6242 assert_eq!(result["verb"], "recall");
6243 assert_eq!(result["pack"], "helptest");
6244
6245 let params = result["params"]
6246 .as_array()
6247 .expect("params must be a JSON array");
6248
6249 let query_param = params.iter().find(|p| p["name"] == "query");
6251 assert!(query_param.is_some(), "params must include 'query'");
6252 let query_param = query_param.unwrap();
6253 assert_eq!(
6254 query_param["required"],
6255 serde_json::json!(true),
6256 "'query' must be required"
6257 );
6258
6259 let limit_param = params.iter().find(|p| p["name"] == "limit");
6261 assert!(limit_param.is_some(), "params must include 'limit'");
6262 let limit_param = limit_param.unwrap();
6263 assert_eq!(
6264 limit_param["required"],
6265 serde_json::json!(false),
6266 "'limit' must be optional"
6267 );
6268 }
6269
6270 #[tokio::test]
6273 async fn test_help_true_does_not_execute_the_verb() {
6274 let invocations = Arc::new(AtomicUsize::new(0));
6275 let reg = build_help_registry(invocations.clone());
6276
6277 reg.dispatch("create", serde_json::json!({ "help": true }))
6279 .await
6280 .expect("help=true must succeed");
6281 reg.dispatch("recall", serde_json::json!({ "help": true }))
6282 .await
6283 .expect("help=true must succeed");
6284
6285 assert_eq!(
6286 invocations.load(Ordering::SeqCst),
6287 0,
6288 "pack dispatch MUST NOT be invoked when help=true; \
6289 got {} invocation(s)",
6290 invocations.load(Ordering::SeqCst)
6291 );
6292
6293 reg.dispatch("create", serde_json::json!({}))
6295 .await
6296 .expect("normal dispatch must succeed");
6297 assert_eq!(
6298 invocations.load(Ordering::SeqCst),
6299 1,
6300 "pack dispatch must fire exactly once for a normal call"
6301 );
6302 }
6303
6304 #[tokio::test]
6313 async fn help_true_on_subhandler_returns_callable_via_mcp_false() {
6314 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6315
6316 let result = reg
6317 .dispatch("recall.embed", serde_json::json!({ "help": true }))
6318 .await
6319 .expect("help=true on subhandler must succeed (no permission check on help path)");
6320
6321 assert_eq!(
6322 result["callable_via_mcp"],
6323 serde_json::json!(false),
6324 "subhandler help must carry callable_via_mcp: false"
6325 );
6326 assert_eq!(
6327 result["visibility"], "internal",
6328 "subhandler help must carry visibility: internal"
6329 );
6330 assert_eq!(result["verb"], "recall.embed");
6333 assert_eq!(result["pack"], "helptest");
6334 }
6335
6336 #[tokio::test]
6338 async fn help_true_on_public_verb_does_not_have_callable_via_mcp_false() {
6339 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6340
6341 let result = reg
6342 .dispatch("create", serde_json::json!({ "help": true }))
6343 .await
6344 .expect("help=true on public verb must succeed");
6345
6346 assert_ne!(
6348 result.get("callable_via_mcp"),
6349 Some(&serde_json::json!(false)),
6350 "public verb help must NOT carry callable_via_mcp: false"
6351 );
6352 assert_ne!(
6354 result.get("visibility"),
6355 Some(&serde_json::json!("internal")),
6356 "public verb help must NOT carry visibility: internal"
6357 );
6358 }
6359
6360 #[tokio::test]
6362 async fn help_true_on_unknown_verb_returns_error() {
6363 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6364
6365 let err = reg
6366 .dispatch("nonexistent_verb", serde_json::json!({ "help": true }))
6367 .await
6368 .unwrap_err();
6369
6370 assert!(
6371 matches!(err, RuntimeError::InvalidInput(_)),
6372 "help=true on unknown verb must return InvalidInput, got {err:?}"
6373 );
6374 let msg = err.to_string();
6375 assert!(
6376 msg.contains("nonexistent_verb"),
6377 "error must name the unknown verb: {msg}"
6378 );
6379 }
6380
6381 #[tokio::test]
6383 async fn help_true_on_subhandler_includes_params_field() {
6384 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6385
6386 let result = reg
6387 .dispatch("recall.embed", serde_json::json!({ "help": true }))
6388 .await
6389 .expect("help=true on subhandler must succeed");
6390
6391 let params = result
6393 .get("params")
6394 .expect("subhandler help must include 'params' field");
6395 assert!(
6396 params.is_array(),
6397 "subhandler help params must be a JSON array"
6398 );
6399 }
6400
6401 #[tokio::test]
6406 async fn help_true_unknown_verb_available_list_excludes_subhandlers() {
6407 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6408
6409 let err = reg
6410 .dispatch("not_a_verb", serde_json::json!({ "help": true }))
6411 .await
6412 .unwrap_err();
6413
6414 let msg = err.to_string();
6415 assert!(
6418 !msg.contains("recall.embed"),
6419 "unknown-verb help error must not advertise subhandler recall.embed: {msg}"
6420 );
6421 assert!(
6423 msg.contains("create"),
6424 "unknown-verb help error must still list public verb 'create': {msg}"
6425 );
6426 assert!(
6427 msg.contains("recall"),
6428 "unknown-verb help error must still list public verb 'recall': {msg}"
6429 );
6430 }
6431
6432 #[tokio::test]
6434 async fn dispatch_unknown_verb_available_list_excludes_subhandlers() {
6435 let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6436
6437 let err = reg
6438 .dispatch("not_a_verb", serde_json::json!({}))
6439 .await
6440 .unwrap_err();
6441
6442 let msg = err.to_string();
6443 assert!(
6446 !msg.contains("recall.embed"),
6447 "dispatch unknown-verb error must not advertise subhandler recall.embed: {msg}"
6448 );
6449 assert!(
6451 msg.contains("create"),
6452 "dispatch unknown-verb error must still list public verb 'create': {msg}"
6453 );
6454 assert!(
6455 msg.contains("recall"),
6456 "dispatch unknown-verb error must still list public verb 'recall': {msg}"
6457 );
6458 }
6459
6460 struct SchemaPack {
6464 pack_name: &'static str,
6465 statements: &'static [&'static str],
6466 }
6467
6468 impl Pack for SchemaPack {
6469 const NAME: &'static str = "schema-pack";
6470 const NOTE_KINDS: &'static [&'static str] = &[];
6471 const ENTITY_KINDS: &'static [&'static str] = &[];
6472 const HANDLERS: &'static [HandlerDef] = &[];
6473 }
6474
6475 #[async_trait]
6476 impl PackRuntime for SchemaPack {
6477 fn name(&self) -> &str {
6478 self.pack_name
6479 }
6480 fn note_kinds(&self) -> &'static [&'static str] {
6481 &[]
6482 }
6483 fn entity_kinds(&self) -> &'static [&'static str] {
6484 &[]
6485 }
6486 fn handlers(&self) -> &'static [HandlerDef] {
6487 &[]
6488 }
6489 fn schema_plan(&self) -> SchemaPlan {
6490 SchemaPlan {
6491 pack: self.pack_name,
6492 statements: self.statements,
6493 }
6494 }
6495 async fn dispatch(
6496 &self,
6497 verb: &str,
6498 _params: Value,
6499 _registry: &VerbRegistry,
6500 _token: &NamespaceToken,
6501 ) -> Result<Value, RuntimeError> {
6502 Ok(serde_json::json!({ "pack": self.pack_name, "verb": verb }))
6503 }
6504 }
6505
6506 #[test]
6509 fn all_schema_plans_named_returns_correct_pairs() {
6510 let mut builder = VerbRegistryBuilder::new();
6511 builder.register_boxed(Box::new(SchemaPack {
6512 pack_name: "alpha",
6513 statements: &["CREATE TABLE IF NOT EXISTS t_alpha (id INTEGER PRIMARY KEY)"],
6514 }));
6515 builder.register_boxed(Box::new(SchemaPack {
6516 pack_name: "beta",
6517 statements: &[],
6518 }));
6519 let reg = builder.build().expect("registry builds");
6520
6521 let named = reg.all_schema_plans_named();
6522 assert_eq!(named.len(), 2);
6523
6524 let alpha_entry = named.iter().find(|(n, _)| *n == "alpha");
6525 let beta_entry = named.iter().find(|(n, _)| *n == "beta");
6526
6527 assert!(alpha_entry.is_some(), "alpha must appear in named plans");
6528 assert!(beta_entry.is_some(), "beta must appear in named plans");
6529
6530 let (_, alpha_plan) = alpha_entry.unwrap();
6531 assert_eq!(alpha_plan.statements.len(), 1);
6532 assert!(!alpha_plan.is_empty());
6533
6534 let (_, beta_plan) = beta_entry.unwrap();
6535 assert!(beta_plan.is_empty());
6536 }
6537
6538 #[tokio::test]
6555 async fn apply_schema_plans_with_map_routes_to_correct_backend() {
6556 use khive_storage::types::{SqlStatement, SqlValue};
6557
6558 let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
6559 let pack_backend =
6560 khive_db::StorageBackend::memory().expect("pack-specific memory backend");
6561
6562 let mut builder = VerbRegistryBuilder::new();
6563 builder.register_boxed(Box::new(SchemaPack {
6564 pack_name: "routed",
6565 statements: &["CREATE TABLE IF NOT EXISTS t_routed (id INTEGER PRIMARY KEY)"],
6566 }));
6567 let reg = builder.build().expect("registry builds");
6568
6569 let mut backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6570 backend_map.insert("routed", &pack_backend);
6571
6572 reg.apply_schema_plans_with_map(&backend_map, &default_backend)
6573 .expect("schema application must not collide");
6574
6575 let mut writer = pack_backend.sql().writer().await.expect("writer");
6577 let result = writer
6578 .execute(SqlStatement {
6579 sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
6580 params: vec![SqlValue::Integer(1)],
6581 label: None,
6582 })
6583 .await;
6584 assert!(
6585 result.is_ok(),
6586 "t_routed must exist on pack_backend after routing: {result:?}"
6587 );
6588
6589 let mut default_writer = default_backend.sql().writer().await.expect("writer");
6591 let default_result = default_writer
6592 .execute(SqlStatement {
6593 sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
6594 params: vec![SqlValue::Integer(2)],
6595 label: None,
6596 })
6597 .await;
6598 assert!(
6599 default_result.is_err(),
6600 "t_routed must NOT exist on default_backend (table should not be there)"
6601 );
6602 }
6603
6604 #[tokio::test]
6607 async fn apply_schema_plans_with_map_falls_back_to_default_for_unmapped_packs() {
6608 use khive_storage::types::{SqlStatement, SqlValue};
6609
6610 let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
6611
6612 let mut builder = VerbRegistryBuilder::new();
6613 builder.register_boxed(Box::new(SchemaPack {
6614 pack_name: "unmapped",
6615 statements: &["CREATE TABLE IF NOT EXISTS t_unmapped (id INTEGER PRIMARY KEY)"],
6616 }));
6617 let reg = builder.build().expect("registry builds");
6618
6619 let backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6620 reg.apply_schema_plans_with_map(&backend_map, &default_backend)
6621 .expect("schema application must not collide");
6622
6623 let mut writer = default_backend.sql().writer().await.expect("writer");
6625 let result = writer
6626 .execute(SqlStatement {
6627 sql: "INSERT INTO t_unmapped (id) VALUES (?1)".into(),
6628 params: vec![SqlValue::Integer(1)],
6629 label: None,
6630 })
6631 .await;
6632 assert!(
6633 result.is_ok(),
6634 "t_unmapped must exist on default_backend for unmapped pack: {result:?}"
6635 );
6636 }
6637
6638 #[test]
6643 fn apply_schema_plans_with_map_collision_is_an_error() {
6644 let backend = khive_db::StorageBackend::memory().expect("memory backend");
6645 let empty_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6646
6647 let mut builder = VerbRegistryBuilder::new();
6648 builder.register_boxed(Box::new(SchemaPack {
6649 pack_name: "pack_alpha",
6650 statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
6651 }));
6652 builder.register_boxed(Box::new(SchemaPack {
6653 pack_name: "pack_beta",
6654 statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
6655 }));
6656 let registry = builder.build().expect("registry builds");
6657
6658 let result = registry.apply_schema_plans_with_map(&empty_map, &backend);
6659 assert!(
6660 result.is_err(),
6661 "two packs declaring the same table on the same backend must produce a collision error"
6662 );
6663 let err = result.unwrap_err();
6664 let msg = err.to_string();
6665 assert!(
6666 msg.contains("pack_alpha"),
6667 "collision error must name first pack; got: {msg}"
6668 );
6669 assert!(
6670 msg.contains("pack_beta"),
6671 "collision error must name second pack; got: {msg}"
6672 );
6673 assert!(
6674 msg.contains("collision_table"),
6675 "collision error must name the table; got: {msg}"
6676 );
6677 }
6678}