khive_runtime/pack/catalog.rs
1//! Registry catalog, kind hooks, and pack lifecycle integration.
2
3use std::any::Any;
4use std::collections::HashMap;
5use std::sync::Arc;
6
7use khive_types::NotePropertyPolicySpec;
8use serde_json::Value;
9
10use crate::context::ContextContributor;
11use crate::error::RuntimeError;
12use crate::operations::{LinkSpec, Resolved};
13use crate::runtime::NamespaceToken;
14use crate::validation::ValidationRule;
15use crate::KhiveRuntime;
16
17use super::request_identity::extract_table_names;
18#[cfg(doc)]
19use super::VerbPresentationPolicy;
20use super::{
21 EdgeEndpointRule, EntityTypeDef, HandlerDef, KindHook, NoteEmbeddingPolicySpec, NoteKindSpec,
22 PackByIdResolver, PackColumnAddition, PackRuntime, SchemaPlan, VerbCategory, VerbRegistry,
23 Visibility, GENERIC_CRUD_PACK,
24};
25
26#[derive(Clone)]
27pub(crate) struct CanonicalNoteHookLookup {
28 packs: Arc<Vec<Box<dyn PackRuntime>>>,
29}
30
31impl std::fmt::Debug for CanonicalNoteHookLookup {
32 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 formatter
34 .debug_struct("CanonicalNoteHookLookup")
35 .finish_non_exhaustive()
36 }
37}
38
39impl CanonicalNoteHookLookup {
40 pub(crate) fn has_hook(&self, kind: &str) -> bool {
41 find_registered_kind_hook(self.packs.as_ref(), kind).is_some()
42 }
43}
44
45fn find_registered_kind_hook(
46 packs: &[Box<dyn PackRuntime>],
47 kind: &str,
48) -> Option<Arc<dyn KindHook>> {
49 for pack in packs {
50 let owns = pack.note_kinds().contains(&kind) || pack.entity_kinds().contains(&kind);
51 if owns {
52 if let Some(hook) = pack.kind_hook(kind) {
53 return Some(hook);
54 }
55 }
56 }
57 None
58}
59
60impl VerbRegistry {
61 /// Registered pack-level by-ID resolvers, in registration order.
62 ///
63 /// Each element is `(pack_name, resolver)`. The kg `get` and `delete` handlers
64 /// iterate this slice to probe pack-private tables when the standard KG
65 /// substrates (entity/note/edge/event) return `None` for a given UUID.
66 pub fn resolvers(&self) -> &[(String, Box<dyn PackByIdResolver>)] {
67 &self.resolvers
68 }
69
70 /// The daemon-warm recently-referenced ring (unified-verb draft ADR,
71 /// Slice 1). Consumed by `resolve_reference` (Layer 0 stage 2) and by the
72 /// `resolve` verb handler; admitted-to by every successful by-id
73 /// dispatch (see the admission block in `dispatch_with_identity`).
74 pub fn reference_ring(&self) -> &Arc<crate::reference_ring::ReferenceRing> {
75 &self.reference_ring
76 }
77
78 /// Find a kind hook among the registered packs.
79 ///
80 /// Walks packs in registration order; the first pack that both owns the
81 /// kind (declares it in `note_kinds()` or `entity_kinds()`) and returns
82 /// a hook from `kind_hook(kind)` wins. Returns `None` if the kind is
83 /// unknown to all packs or no owning pack registered a hook.
84 pub fn find_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
85 find_registered_kind_hook(self.packs.as_ref(), kind)
86 }
87
88 pub(crate) fn canonical_note_hook_lookup(&self) -> CanonicalNoteHookLookup {
89 CanonicalNoteHookLookup {
90 packs: Arc::clone(&self.packs),
91 }
92 }
93
94 /// Every `(entity kind, hook)` pair for which the owning pack declares
95 /// the entity kind and registers a `KindHook` — the entity-scoped
96 /// subset of [`Self::find_kind_hook`]'s ownership check, computed once.
97 ///
98 /// `khive-runtime` does not hold a `VerbRegistry` (ownership runs the
99 /// other way: packs are constructed FROM a runtime handle), so
100 /// `KhiveRuntime::install_entity_kind_hooks` is the extension point
101 /// that carries this aggregate to the runtime layer — the transport
102 /// calls this after the registry is built, same timing as
103 /// [`Self::all_edge_rules`]. `Arc<dyn KindHook>` values returned here
104 /// hold no reference back to the pack or registry that produced them
105 /// (every production `kind_hook()` implementation constructs a fresh,
106 /// stateless hook per call), so installing this aggregate on the
107 /// runtime creates no ownership cycle.
108 pub fn entity_kind_hooks(&self) -> crate::runtime::EntityKindHooks {
109 let mut hooks = Vec::new();
110 for pack in self.packs.iter() {
111 for kind in pack.entity_kinds().iter().copied() {
112 if let Some(hook) = pack.kind_hook(kind) {
113 hooks.push((kind.to_string(), hook));
114 }
115 }
116 }
117 hooks
118 }
119
120 /// Run the owning kind's shared-note-update normalizer/validator, if it declares one.
121 ///
122 /// Compatibility wrapper for callers that only need normalization and
123 /// validation. Writers use [`Self::prepare_note_update_policy`] and attach
124 /// its returned policy so kind-specific property removals reach storage.
125 ///
126 /// The ordering lives here, at the single dispatch site, rather than in a
127 /// [`KindHook`] method a pack could override: a pack implements the two
128 /// halves and cannot express a sequence, so it cannot replace the
129 /// validator by overriding the sequence. See ADR-017.
130 pub async fn prepare_note_update_hook(
131 &self,
132 runtime: &KhiveRuntime,
133 token: &NamespaceToken,
134 note: &khive_storage::Note,
135 args: &mut Value,
136 ) -> Result<(), RuntimeError> {
137 self.prepare_note_update_policy(runtime, token, note, args)
138 .await
139 .map(|_| ())
140 }
141
142 /// Normalize and validate a note update, then carry the owning kind's
143 /// property policy into the shared prepared write. Writers must attach the
144 /// returned policy to their `NotePatch` or snapshot update preparation;
145 /// [`Self::prepare_note_update_hook`] remains the validation-only wrapper.
146 pub async fn prepare_note_update_policy(
147 &self,
148 runtime: &KhiveRuntime,
149 token: &NamespaceToken,
150 note: &khive_storage::Note,
151 args: &mut Value,
152 ) -> Result<crate::NoteUpdatePolicy, RuntimeError> {
153 crate::curation::normalize_note_update_tags(args)?;
154 if let Some(hook) = self.find_kind_hook(¬e.kind) {
155 hook.normalize_note_update(runtime, token, note, args)
156 .await?;
157 let properties = args.get("properties").filter(|value| !value.is_null());
158 hook.validate_note_update(runtime, token, note, properties)
159 .await?;
160 return Ok(crate::NoteUpdatePolicy::for_kind(
161 ¬e.kind,
162 hook.note_update_null_clearing_properties(),
163 ));
164 }
165 Ok(crate::NoteUpdatePolicy::default())
166 }
167
168 /// Run the owning kind's shared-note-update property validator, if it
169 /// declares one.
170 ///
171 /// Kept as the validation-only compatibility seam for callers that do not
172 /// own a mutable request object. Canonical and atomic CRUD use
173 /// [`Self::prepare_note_update_hook`] instead, so a hook's
174 /// [`KindHook::normalize_note_update`] can run before its validation does.
175 /// Reaching a hook through this seam therefore runs the validator alone:
176 /// that is the point of it, and it is why callers that CAN supply a
177 /// mutable request should not use it.
178 pub async fn validate_note_update_hook(
179 &self,
180 runtime: &KhiveRuntime,
181 token: &NamespaceToken,
182 note: &khive_storage::Note,
183 properties: Option<&Value>,
184 ) -> Result<(), RuntimeError> {
185 if let Some(hook) = self.find_kind_hook(¬e.kind) {
186 hook.validate_note_update(runtime, token, note, properties)
187 .await?;
188 }
189 Ok(())
190 }
191
192 /// Run shared-link validators grouped by the owning source-note kind.
193 ///
194 /// Supplying the whole proposed batch lets a kind hook reject an invariant
195 /// violation formed only by multiple entries in that batch. Sources that
196 /// are not live notes, or whose kind has no hook, remain the canonical
197 /// endpoint validator's responsibility.
198 pub async fn validate_link_hooks(
199 &self,
200 runtime: &KhiveRuntime,
201 token: &NamespaceToken,
202 specs: &[LinkSpec],
203 ) -> Result<(), RuntimeError> {
204 let mut specs_by_kind: HashMap<String, Vec<LinkSpec>> = HashMap::new();
205 for spec in specs {
206 let Some(Resolved::Note(source)) = runtime.resolve_by_id(token, spec.source_id).await?
207 else {
208 continue;
209 };
210 specs_by_kind
211 .entry(source.kind)
212 .or_default()
213 .push(spec.clone());
214 }
215 for (kind, kind_specs) in specs_by_kind {
216 if let Some(hook) = self.find_kind_hook(&kind) {
217 hook.validate_links(runtime, token, &kind_specs).await?;
218 }
219 }
220 Ok(())
221 }
222
223 /// Whether any registered pack declares a handler with this verb name.
224 ///
225 /// A non-dispatch capability check: callers that would otherwise pay a
226 /// guaranteed-failed `dispatch` (and its audit write) when an optional
227 /// pack is absent can probe first and skip the call entirely.
228 pub fn has_verb(&self, verb: &str) -> bool {
229 self.handler_by_name.contains_key(verb) && !self.is_verb_disabled(verb)
230 }
231
232 /// Whether operator policy disables this registered public handler.
233 pub fn is_verb_disabled(&self, verb: &str) -> bool {
234 self.disabled_verbs.contains(verb)
235 }
236
237 pub(crate) fn unknown_verb_error(&self, verb: &str) -> RuntimeError {
238 RuntimeError::UnknownVerb(format!(
239 "unknown verb {verb:?}; available: {}",
240 self.available_verbs.join(", ")
241 ))
242 }
243
244 /// Advisory metadata for synchronous planning and MCP initialization.
245 pub fn mounted_verb_snapshot(&self) -> Vec<Value> {
246 self.packs
247 .iter()
248 .flat_map(|pack| {
249 pack.mounted_catalog_snapshot()
250 .into_iter()
251 .map(|verb| verb.describe(pack.name()))
252 })
253 .collect()
254 }
255
256 pub async fn mounted_verb_catalog(&self) -> Result<Vec<Value>, RuntimeError> {
257 let mut catalog = Vec::new();
258 for pack in self.packs.iter() {
259 for definition in pack.mounted_catalog().await? {
260 catalog.push(definition.describe(pack.name()));
261 }
262 }
263 Ok(catalog)
264 }
265
266 /// Apply section evidence through the installed brain instance. Callers must
267 /// validate their domain target and authorize their own operation first;
268 /// this trusted Rust hook adds no handler to dispatch or the wire catalog.
269 pub async fn apply_profile_section_feedback(
270 &self,
271 token: &NamespaceToken,
272 profile_id: &str,
273 section_signals: Value,
274 target_attribution: Option<String>,
275 ) -> Result<Value, RuntimeError> {
276 let brain = self
277 .packs
278 .iter()
279 .find(|pack| pack.name() == "brain")
280 .ok_or_else(|| {
281 RuntimeError::InvalidInput(
282 "profile section feedback requires the brain pack".into(),
283 )
284 })?;
285 brain
286 .apply_profile_section_feedback(token, profile_id, section_signals, target_attribution)
287 .await
288 }
289
290 /// All MCP-exposed handlers across all registered packs (`Visibility::Verb` only).
291 ///
292 /// Disabled verbs and subhandlers (`Visibility::Subhandler`) are excluded — subhandlers are internal
293 /// pipeline steps not surfaced on the MCP wire. Returned with `'static`
294 /// lifetime since pack handlers are `&'static [HandlerDef]` constants.
295 pub fn all_verbs(&self) -> Vec<&'static HandlerDef> {
296 self.registered_handlers()
297 .map(|(_, handler)| handler)
298 .filter(|handler| self.is_enabled_verb(handler))
299 .collect()
300 }
301
302 /// All MCP-exposed handlers paired with the name of the pack that owns them
303 /// (`Visibility::Verb` only).
304 ///
305 /// Subhandlers (`Visibility::Subhandler`) are excluded from the MCP catalog
306 /// Use `all_handlers_with_names` when internal handlers must
307 /// also be enumerated (e.g. runtime introspection).
308 pub fn all_verbs_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
309 self.registered_handlers()
310 .map(|(pack, handler)| (pack.name(), handler))
311 .filter(|(_, handler)| self.is_enabled_verb(handler))
312 .collect()
313 }
314
315 /// All handler definitions across all registered packs, including disabled verbs and subhandlers.
316 ///
317 /// Unlike `all_verbs`, this includes `Visibility::Subhandler` entries. Useful
318 /// for runtime introspection (e.g. `list_handlers`) and tooling that needs
319 /// the complete handler surface.
320 pub fn all_handlers_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
321 self.registered_handlers()
322 .map(|(pack, handler)| (pack.name(), handler))
323 .collect()
324 }
325
326 fn registered_handlers(
327 &self,
328 ) -> impl Iterator<Item = (&dyn PackRuntime, &'static HandlerDef)> + '_ {
329 self.packs.iter().flat_map(|pack| {
330 let pack = pack.as_ref();
331 pack.handlers().iter().map(move |handler| (pack, handler))
332 })
333 }
334
335 fn is_enabled_verb(&self, handler: &HandlerDef) -> bool {
336 matches!(handler.visibility, Visibility::Verb) && !self.is_verb_disabled(handler.name)
337 }
338
339 /// Collect declared kinds once, retaining their first-seen registration order.
340 fn collect_pack_kinds(
341 &self,
342 select: impl Fn(&dyn PackRuntime) -> &'static [&'static str],
343 ) -> Vec<&'static str> {
344 let mut seen = std::collections::HashSet::new();
345 self.packs
346 .iter()
347 .flat_map(|pack| select(pack.as_ref()).iter().copied())
348 .filter(|kind| seen.insert(*kind))
349 .collect()
350 }
351
352 /// Merged set of note kinds across all registered packs (deduplicated,
353 /// first-seen order preserved).
354 pub fn all_note_kinds(&self) -> Vec<&'static str> {
355 self.collect_pack_kinds(|pack| pack.note_kinds())
356 }
357
358 /// Note kinds owned by a pack, i.e. every kind in [`all_note_kinds`] that
359 /// is not one of the generic-CRUD pack's own kinds.
360 ///
361 /// [`GENERIC_CRUD_PACK`] declares the general-purpose note kinds the shared
362 /// CRUD verbs exist to serve (`observation`, `insight`, …); every other
363 /// pack's kinds are records that pack's own verbs create and maintain.
364 /// Derived from the packs' `NOTE_KINDS` constants, so a pack that adds or
365 /// drops a kind moves this set with it — nothing is hardcoded here but the
366 /// name of the generic pack itself.
367 ///
368 /// [`all_note_kinds`]: Self::all_note_kinds
369 pub fn pack_owned_note_kinds(&self) -> Vec<&'static str> {
370 let generic: std::collections::HashSet<&'static str> = self
371 .packs
372 .iter()
373 .filter(|p| p.name() == GENERIC_CRUD_PACK)
374 .flat_map(|p| p.note_kinds().iter().copied())
375 .collect();
376 let mut seen = std::collections::HashSet::new();
377 self.packs
378 .iter()
379 .filter(|p| p.name() != GENERIC_CRUD_PACK)
380 .flat_map(|p| p.note_kinds().iter().copied())
381 .filter(|k| !generic.contains(k) && seen.insert(*k))
382 .collect()
383 }
384
385 /// Merged set of entity kinds across all registered packs (deduplicated,
386 /// first-seen order preserved).
387 pub fn all_entity_kinds(&self) -> Vec<&'static str> {
388 self.collect_pack_kinds(|pack| pack.entity_kinds())
389 }
390
391 /// Merged set of brain profile consumer kinds requested by registered
392 /// packs (deduplicated, first-seen order preserved).
393 pub fn all_brain_consumer_kinds(&self) -> Vec<&'static str> {
394 self.collect_pack_kinds(|pack| pack.brain_consumer_kinds())
395 }
396
397 /// Names of packs in topological load order.
398 pub fn pack_names(&self) -> Vec<&str> {
399 self.packs.iter().map(|p| p.name()).collect()
400 }
401
402 /// Borrow a registered pack's shared host state without reconstructing
403 /// that pack. Missing packs, absent state, and type mismatches return None.
404 pub fn pack_host_state<T: Any + Send + Sync>(&self, name: &str) -> Option<Arc<T>> {
405 self.find_pack(name)?.host_state()?.downcast::<T>().ok()
406 }
407
408 /// Factory-reported version for a registered pack.
409 ///
410 /// Returns `None` for unknown packs and packs registered directly without
411 /// a factory; their package version is not available to the registry.
412 pub fn pack_version(&self, name: &str) -> Option<&'static str> {
413 self.pack_versions.get(name).copied()
414 }
415
416 /// Declared dependencies for a registered pack.
417 pub fn pack_requires(&self, name: &str) -> Option<&'static [&'static str]> {
418 self.find_pack(name).map(|pack| pack.requires())
419 }
420
421 /// Note kinds owned by a specific registered pack.
422 ///
423 /// Returns `None` if no pack with `name` is registered. The slice is
424 /// the pack's `NOTE_KINDS` constant — `'static` lifetime, no allocation.
425 pub fn pack_note_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
426 self.find_pack(name).map(|pack| pack.note_kinds())
427 }
428
429 /// Entity kinds owned by a specific registered pack.
430 ///
431 /// Returns `None` if no pack with `name` is registered. The slice is
432 /// the pack's `ENTITY_KINDS` constant — `'static` lifetime, no allocation.
433 pub fn pack_entity_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
434 self.find_pack(name).map(|pack| pack.entity_kinds())
435 }
436
437 /// Handlers declared by a specific registered pack.
438 ///
439 /// Returns `None` if no pack with `name` is registered. Each `HandlerDef`
440 /// carries name + description + visibility — sufficient for introspection clients.
441 pub fn pack_verbs(&self, name: &str) -> Option<&'static [HandlerDef]> {
442 self.find_pack(name).map(|pack| pack.handlers())
443 }
444
445 fn find_pack(&self, name: &str) -> Option<&dyn PackRuntime> {
446 self.packs
447 .iter()
448 .find(|pack| pack.name() == name)
449 .map(|pack| pack.as_ref())
450 }
451
452 /// Collect pack items in registration order without filtering or deduplication.
453 fn collect_pack_items<T, I>(&self, select: impl Fn(&dyn PackRuntime) -> I) -> Vec<T>
454 where
455 I: IntoIterator<Item = T>,
456 {
457 self.packs
458 .iter()
459 .flat_map(|pack| select(pack.as_ref()))
460 .collect()
461 }
462
463 /// Optional context sources in the registry's topological pack order.
464 ///
465 /// Omits packs returning `None`, without caching, deduplicating or invoking contributors.
466 pub fn context_contributors(&self) -> Vec<Arc<dyn ContextContributor>> {
467 self.collect_pack_items(|pack| pack.context_contributor())
468 }
469
470 /// All pack-declared edge endpoint rules across registered packs.
471 ///
472 /// Order follows topological pack registration; duplicates are *not* deduplicated —
473 /// validation only checks membership, and an exact-duplicate rule is a
474 /// harmless restatement.
475 pub fn all_edge_rules(&self) -> Vec<EdgeEndpointRule> {
476 self.attributed_edge_rules
477 .iter()
478 .map(|(_, rule)| *rule)
479 .collect()
480 }
481
482 /// Pack-declared endpoint rules paired with their owning pack names.
483 ///
484 /// Ownership is captured when the registry is built. Order and duplicate
485 /// declarations match [`Self::all_edge_rules`] exactly; built-in substrate
486 /// rules are excluded from both accessors. Names borrow the immutable
487 /// registry snapshot, so registry clones retain the same attribution.
488 pub fn all_edge_rules_with_packs(&self) -> Vec<(&str, EdgeEndpointRule)> {
489 self.attributed_edge_rules
490 .iter()
491 .map(|(pack, rule)| (pack.as_str(), *rule))
492 .collect()
493 }
494
495 /// All pack-declared entity-type subtypes across registered packs.
496 ///
497 /// Order follows topological pack registration; duplicates are *not*
498 /// deduplicated here — same posture as [`all_edge_rules`](Self::all_edge_rules).
499 /// Consumers compose this with `EntityTypeRegistry::builtin()` via
500 /// `EntityTypeRegistry::with_extra` to get the boot-time composed registry.
501 pub fn all_entity_types(&self) -> Vec<EntityTypeDef> {
502 self.collect_pack_items(|pack| pack.entity_types().iter().cloned())
503 }
504
505 /// Collect all `NoteKindSpec` declarations from every loaded pack.
506 ///
507 /// Used by the runtime for lifecycle introspection and future enforcement.
508 pub fn all_note_kind_specs(&self) -> Vec<&'static NoteKindSpec> {
509 self.collect_pack_items(|pack| pack.note_kind_specs().iter())
510 }
511
512 /// Copy the validated boot snapshot in topological pack/declaration order.
513 ///
514 /// This metadata-only accessor never calls packs or changes write enforcement.
515 pub fn all_note_property_policies(&self) -> Vec<NotePropertyPolicySpec> {
516 self.note_property_policies.as_ref().clone()
517 }
518
519 /// Collect pack-declared embedding policies for registered note kinds.
520 pub fn all_note_embedding_policies(&self) -> Vec<NoteEmbeddingPolicySpec> {
521 self.collect_pack_items(|pack| pack.note_embedding_policies().iter().copied())
522 }
523
524 /// All pack-contributed validation rules across registered packs.
525 ///
526 /// Returns references into the pack-owned `'static` slices — no allocation
527 /// beyond the outer `Vec`. Rule IDs are namespaced by pack; callers can
528 /// group by `rule.id.split_once('/')` to attribute rules to their packs.
529 pub fn all_validation_rules(&self) -> Vec<&'static ValidationRule> {
530 self.collect_pack_items(|pack| pack.validation_rules().iter())
531 }
532
533 /// Pack-auxiliary schema plans for all registered packs.
534 ///
535 /// Returns one `SchemaPlan` per pack. Callers (typically the runtime
536 /// bootstrap) apply each plan to the pack's assigned backend. Empty plans
537 /// are included so the caller can iterate uniformly; callers that want to
538 /// skip empty plans should check `plan.is_empty()`. Schema application must
539 /// use [`Self::all_schema_plans_with_columns`] to retain column upgrades.
540 pub fn all_schema_plans(&self) -> Vec<SchemaPlan> {
541 self.packs.iter().map(|p| p.schema_plan()).collect()
542 }
543
544 /// Schema plans paired with the same owning pack's nullable-column upgrades.
545 ///
546 /// Callers applying plans directly must pass both entries to
547 /// `StorageBackend::apply_pack_ddl_statements_with_columns`.
548 pub fn all_schema_plans_with_columns(
549 &self,
550 ) -> Vec<(SchemaPlan, &'static [PackColumnAddition])> {
551 self.packs
552 .iter()
553 .map(|pack| (pack.schema_plan(), pack.schema_column_additions()))
554 .collect()
555 }
556
557 /// Invoke `PackRuntime::register_embedders` on every registered pack.
558 ///
559 /// Called by the transport during startup, after the registry is built and
560 /// before the first verb dispatch, so that custom embedding providers
561 /// contributed by packs are reachable via `KhiveRuntime::embedder(name)`.
562 ///
563 /// This low-level hook dispatcher does not finalize configured peers.
564 /// Hosts use [`Self::initialize_embedding_engines`] for serialized,
565 /// idempotent registration and fallible metadata binding before serving.
566 pub fn call_register_embedders(&self, runtime: &KhiveRuntime) {
567 for pack in self.packs.iter() {
568 pack.register_embedders(runtime);
569 }
570 }
571
572 /// Register pack providers and finalize configured engines before serving.
573 ///
574 /// Repeated aliases of one runtime registry initialize once. Distinct
575 /// no-embedding runtimes bind an empty list without running provider hooks;
576 /// their main/core registry is unaffected. No model services are built.
577 pub fn initialize_embedding_engines(
578 &self,
579 runtimes: &[&KhiveRuntime],
580 ) -> Result<(), RuntimeError> {
581 let mut seen = std::collections::HashSet::new();
582 for runtime in runtimes {
583 if seen.insert(runtime.embedding_registry_identity()) {
584 runtime.initialize_embedding_engines(|| self.call_register_embedders(runtime))?;
585 }
586 }
587 Ok(())
588 }
589
590 /// Invoke `PackRuntime::register_entity_type_validator` on every registered pack.
591 ///
592 /// Called by the transport during startup, after the registry is built and
593 /// before the first verb dispatch, so that entity-type validation at the
594 /// runtime layer is active for all write paths including direct `create_many`
595 /// callers that bypass the handler layer.
596 ///
597 /// Packs whose `register_entity_type_validator` is the default no-op pay
598 /// no overhead.
599 ///
600 /// Composes [`all_entity_types`](Self::all_entity_types) once and passes
601 /// the same aggregate to every pack, mirroring how `install_edge_rules`
602 /// installs one `all_edge_rules()` aggregate for the whole registry.
603 pub fn call_register_entity_type_validators(&self, runtime: &KhiveRuntime) {
604 let entity_types = self.all_entity_types();
605 for pack in self.packs.iter() {
606 pack.register_entity_type_validator_with_types(runtime, &entity_types);
607 }
608 }
609
610 /// Invoke `PackRuntime::register_note_mutation_hook` on every registered pack.
611 ///
612 /// Called by the transport during startup, after the registry is built and
613 /// before the first verb dispatch, so that note-mutation notifications at
614 /// the runtime layer are active for all write paths — including KG's
615 /// `update`/`delete` verbs reaching a `kind="memory"` note, which have no
616 /// crate-level dependency on `khive-pack-memory`.
617 ///
618 /// Packs whose `register_note_mutation_hook` is the default no-op pay no
619 /// overhead.
620 pub fn call_register_note_mutation_hooks(&self, runtime: &KhiveRuntime) {
621 for pack in self.packs.iter() {
622 pack.register_note_mutation_hook(runtime);
623 }
624 }
625
626 /// Install pack-owned note-search candidate sources before warm-up or
627 /// dispatch, following the same registration timing as mutation hooks.
628 pub fn call_register_note_search_ann_providers(&self, runtime: &KhiveRuntime) {
629 for pack in self.packs.iter() {
630 pack.register_note_search_ann_provider(runtime);
631 }
632 }
633
634 /// Invoke `PackRuntime::register_note_write_validator` on every registered pack.
635 ///
636 /// Called by the transport during startup with the same timing as
637 /// `call_register_note_mutation_hooks`, so note-write validation is active
638 /// at the runtime layer for every write path — the generic `create` verb,
639 /// direct Rust callers, and proposal apply, none of which dispatch a pack
640 /// hook of their own on the note-write.
641 pub fn call_register_note_write_validators(&self, runtime: &KhiveRuntime) {
642 for pack in self.packs.iter() {
643 pack.register_note_write_validator(runtime);
644 }
645 }
646
647 /// Invoke `PackRuntime::warm` on every registered pack.
648 /// Called by the daemon at boot (in a background task) so expensive in-memory
649 /// state (ANN indexes) is pre-loaded without blocking request serving.
650 pub async fn call_warm_all(&self) {
651 for pack in self.packs.iter() {
652 pack.warm().await;
653 }
654 }
655
656 /// Resolve the presentation policy for a verb name.
657 ///
658 /// Uses the first registered handler (including subhandlers) with this name
659 /// and returns its declared [`VerbPresentationPolicy`].
660 /// Returns `Standard` for unknown verbs — unknown verbs will fail at
661 /// dispatch anyway, so the fallback here is safe.
662 pub fn presentation_policy_for(&self, verb: &str) -> khive_types::VerbPresentationPolicy {
663 self.handler_by_name
664 .get(verb)
665 .map_or(khive_types::VerbPresentationPolicy::Standard, |handler| {
666 handler.presentation_policy()
667 })
668 }
669
670 /// Resolve the declared [`VerbCategory`] for a verb name.
671 ///
672 /// Uses the first registered handler (including subhandlers) with this name
673 /// and returns its speech-act category. Returns `None` for
674 /// an unregistered verb name, so a caller deciding transport-level
675 /// behavior (e.g. whether a post-dispatch condition is safe to retry)
676 /// can fail closed on an unknown verb instead of guessing a category.
677 pub fn verb_category(&self, verb: &str) -> Option<VerbCategory> {
678 self.handler_by_name
679 .get(verb)
680 .map(|handler| handler.category)
681 }
682
683 /// Verbs classified [`VerbCategory::Assertive`] that nonetheless schedule
684 /// can schedule a persisted write on a successful dispatch, so a caller re-issuing
685 /// a call in this list after a lost response duplicates that write:
686 ///
687 /// - `memory.recall` schedules `brain.record_serve`, which inserts a
688 /// serve-ledger row keyed in part on a `served_at` timestamp captured
689 /// fresh at dispatch time — a second dispatch inserts a second row
690 /// rather than colliding with the first.
691 /// - `search` (the `kg` pack's bare verb) appends a `search_executed`
692 /// event with a freshly generated id and no natural key at all.
693 /// - `telemetry.emit` can append a durable stream record with a fresh
694 /// identity and sequence, depending on the configured channel policy.
695 /// - `tool.check` appends a `tool_check_decided` receipt with a fresh
696 /// event id for every evaluated decision (ADR-180 Amendment 6).
697 ///
698 /// The speech-act category alone cannot rule this out — it describes
699 /// what the verb tells the *caller*, not what it schedules against
700 /// storage. Adding a verb here (or removing one because its side effect
701 /// was made idempotent) is a correctness decision requiring the same
702 /// scrutiny as the categorization itself.
703 pub const SIDE_EFFECTING_ASSERTIVE_VERBS: &'static [&'static str] =
704 &["memory.recall", "search", "telemetry.emit", "tool.check"];
705
706 /// Whether a response lost to the daemon frame budget may be truthfully
707 /// advertised as safe to re-issue: the verb is [`VerbCategory::Assertive`]
708 /// (no institutional commitment was made) and is not on
709 /// `Self::SIDE_EFFECTING_ASSERTIVE_VERBS` (no persisted write to
710 /// duplicate on a second dispatch). An unregistered verb name resolves to
711 /// `None` from [`Self::verb_category`] and fails closed here.
712 ///
713 /// Used only by the MCP daemon's frame-budget omission decision; never
714 /// for permission checking or return-shape selection.
715 pub fn is_retry_safe_after_frame_omission(&self, verb: &str) -> bool {
716 matches!(self.verb_category(verb), Some(VerbCategory::Assertive))
717 && !Self::SIDE_EFFECTING_ASSERTIVE_VERBS.contains(&verb)
718 }
719
720 /// Returns `true` if the named verb exists and is tagged
721 /// `Visibility::Subhandler` (internal / operator-only).
722 ///
723 /// Used by the MCP server to gate subhandler invocation at the wire
724 /// boundary without blocking internal callers that invoke the same verbs
725 /// through the runtime directly.
726 pub fn is_subhandler_verb(&self, verb: &str) -> bool {
727 self.handler_by_name
728 .get(verb)
729 .is_some_and(|handler| matches!(handler.visibility, Visibility::Subhandler))
730 }
731
732 /// Apply all non-empty pack-auxiliary schema plans to the given backend.
733 ///
734 /// This is the centralized startup hook that replaced the previous lazy
735 /// per-pack self-bootstrap pattern. Each pack's `SchemaPlan` carries
736 /// idempotent `CREATE TABLE IF NOT EXISTS` DDL; calling this more than once
737 /// is safe. Plans with neither SQL nor column upgrades are skipped.
738 ///
739 /// Errors from individual plans are logged via `tracing::warn!` and not
740 /// propagated so that a single pack's schema failure does not prevent the
741 /// rest from loading. Serving hosts must instead use the fallible
742 /// [`Self::apply_schema_plans_with_map`] (with an empty map for one backend)
743 /// so a required schema failure cannot leave a pack's verbs unavailable.
744 pub fn apply_schema_plans(&self, backend: &khive_db::StorageBackend) {
745 if backend.is_read_only() {
746 tracing::info!(
747 "skipping pack schema plans because the backend is read-only; snapshot schema is used as-is"
748 );
749 return;
750 }
751 for (plan, additions) in self.all_schema_plans_with_columns() {
752 if plan.is_empty() && additions.is_empty() {
753 continue;
754 }
755 if let Err(e) =
756 backend.apply_pack_ddl_statements_with_columns(plan.statements, additions)
757 {
758 tracing::warn!(
759 pack = plan.pack,
760 error = %e,
761 "failed to apply pack schema plan at startup (non-fatal)"
762 );
763 }
764 }
765 }
766
767 /// Pack-auxiliary schema plans with their owning pack names.
768 ///
769 /// Returns `(pack_name, SchemaPlan)` pairs for every registered pack.
770 /// Used by the multi-backend boot path to apply each plan to the pack's
771 /// assigned backend rather than a single shared backend. Direct schema
772 /// application must use [`Self::all_schema_plans_with_columns`] so column
773 /// upgrades are retained.
774 pub fn all_schema_plans_named(&self) -> Vec<(&'static str, SchemaPlan)> {
775 self.packs
776 .iter()
777 .map(|p| {
778 let plan = p.schema_plan();
779 (plan.pack, plan)
780 })
781 .collect()
782 }
783
784 /// Apply pack-auxiliary schema plans using a per-pack backend map.
785 ///
786 /// For each plan and its owning pack's column additions, applies the full
787 /// plan to `backend_for_pack[plan.pack]` when present,
788 /// falling back to `default_backend` for any pack not in the map.
789 ///
790 /// Returns an error when two packs on the same backend declare the same
791 /// auxiliary table (ADR-028 §7 collision policy: boot failure naming both
792 /// packs and the conflicting table).
793 ///
794 /// Both single- and multi-backend hosts use this boot path (ADR-028).
795 /// An empty map selects the default backend for every pack. Read-only
796 /// backends validate declared columns without applying SQL or acquiring a
797 /// writer; missing or incompatible columns refuse boot with the pack name.
798 pub fn apply_schema_plans_with_map(
799 &self,
800 backend_for_pack: &HashMap<&str, &khive_db::StorageBackend>,
801 default_backend: &khive_db::StorageBackend,
802 ) -> Result<(), crate::PackSchemaCollisionError> {
803 // Track which pack first claimed each table on each backend.
804 // Backend identity is the raw pointer of the underlying connection pool Arc.
805 let mut claimed: HashMap<(*const (), String), &'static str> = HashMap::new();
806
807 let plans = self.all_schema_plans_with_columns();
808 // Check every declaration before applying any pack DDL. A collision
809 // must not leave earlier plans installed on a failed boot.
810 for (plan, additions) in &plans {
811 if plan.is_empty() && additions.is_empty() {
812 continue;
813 }
814 let pack_name = plan.pack;
815 let backend = backend_for_pack
816 .get(pack_name)
817 .copied()
818 .unwrap_or(default_backend);
819 let backend_ptr = std::sync::Arc::as_ptr(&backend.pool_arc()) as *const ();
820
821 // Collect DDL table ownership for the full plan set.
822 for stmt in plan.statements {
823 for table_name in extract_table_names(stmt) {
824 let key = (backend_ptr, table_name.clone());
825 match claimed.entry(key) {
826 std::collections::hash_map::Entry::Vacant(e) => {
827 e.insert(pack_name);
828 }
829 std::collections::hash_map::Entry::Occupied(e) => {
830 let prior_pack = *e.get();
831 return Err(crate::PackSchemaCollisionError {
832 pack_a: prior_pack,
833 pack_b: pack_name,
834 table: table_name,
835 });
836 }
837 }
838 }
839 }
840 for addition in *additions {
841 let table_name = addition.table.to_ascii_lowercase();
842 let key = (backend_ptr, table_name.clone());
843 match claimed.entry(key) {
844 std::collections::hash_map::Entry::Vacant(entry) => {
845 entry.insert(pack_name);
846 }
847 std::collections::hash_map::Entry::Occupied(entry) => {
848 let prior_pack = *entry.get();
849 // A pack's full CREATE and its upgrades declare the
850 // same table; this is one ownership claim.
851 if prior_pack != pack_name {
852 return Err(crate::PackSchemaCollisionError {
853 pack_a: prior_pack,
854 pack_b: pack_name,
855 table: table_name,
856 });
857 }
858 }
859 }
860 }
861 }
862
863 for (plan, additions) in plans {
864 if plan.is_empty() && additions.is_empty() {
865 continue;
866 }
867 let pack_name = plan.pack;
868 let backend = backend_for_pack
869 .get(pack_name)
870 .copied()
871 .unwrap_or(default_backend);
872 if backend.is_read_only() {
873 backend.validate_pack_schema_columns(additions).map_err(|error| {
874 crate::PackSchemaCollisionError {
875 pack_a: pack_name,
876 pack_b: pack_name,
877 table: format!("read-only schema validation failed: {error}; open the database writable to apply the pack schema upgrade"),
878 }
879 })?;
880 continue;
881 }
882
883 backend
884 .apply_pack_ddl_statements_with_columns(plan.statements, additions)
885 .map_err(|e| crate::PackSchemaCollisionError {
886 pack_a: pack_name,
887 pack_b: pack_name,
888 table: format!("DDL error: {e}"),
889 })?;
890 }
891 Ok(())
892 }
893}