khive_runtime/runtime.rs
1//! KhiveRuntime — composable handle to all storage capabilities.
2//!
3//! `RuntimeConfig`, `BackendId`, `NamespaceToken`, and embedding model helpers
4//! live in `super::config` and are re-exported from here.
5
6use std::collections::{HashMap, HashSet};
7use std::future::Future;
8use std::path::PathBuf;
9use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
10
11use khive_db::{ConnectionPool, StorageBackend};
12#[cfg(test)]
13use khive_gate::AllowAllGate;
14use khive_gate::GateRequest;
15use khive_storage::types::{SqlStatement, SqlValue};
16use khive_storage::{
17 AttachmentStore, EntityStore, Event, EventStore, GraphStore, NoteStore, SqlAccess, VectorStore,
18};
19use khive_types::{EdgeEndpointRule, EventKind, Namespace, SubstrateKind};
20use lattice_embed::{EmbeddingModel, EmbeddingService};
21
22use crate::config::{
23 build_embedder_registry, parse_embedding_model_alias, register_configured_embedding_models,
24 sanitize_key, vec_model_key,
25};
26use crate::error::{RuntimeError, RuntimeResult};
27use crate::note_search_ann::NoteSearchAnnProvider;
28use crate::pack::KindHook;
29
30#[path = "runtime/config_access.rs"]
31mod config_access;
32mod embedder_init;
33mod events_disk_policy;
34mod serving_policy;
35
36#[cfg(all(test, target_os = "macos"))]
37const IN_PROCESS_TEST_NOFILE_LIMIT: libc::rlim_t = 4096;
38#[cfg(all(test, target_os = "macos"))]
39static IN_PROCESS_TEST_NOFILE_INIT: std::sync::Once = std::sync::Once::new();
40
41#[cfg(all(test, target_os = "macos"))]
42fn ensure_in_process_test_nofile_limit() {
43 IN_PROCESS_TEST_NOFILE_INIT.call_once(|| {
44 let mut limits = libc::rlimit {
45 rlim_cur: 0,
46 rlim_max: 0,
47 };
48 // SAFETY: `limits` is writable, and only this test binary's soft
49 // limit may change; the inherited hard limit is preserved.
50 assert_eq!(unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limits) }, 0);
51 assert!(
52 limits.rlim_max >= IN_PROCESS_TEST_NOFILE_LIMIT,
53 "in-process SQLite tests require a hard open-file limit of at least {IN_PROCESS_TEST_NOFILE_LIMIT}"
54 );
55 if limits.rlim_cur < IN_PROCESS_TEST_NOFILE_LIMIT {
56 limits.rlim_cur = IN_PROCESS_TEST_NOFILE_LIMIT;
57 // SAFETY: the new soft limit does not exceed the observed hard
58 // limit, which is left unchanged.
59 assert_eq!(unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &limits) }, 0);
60 }
61 });
62}
63
64tokio::task_local! {
65 static REQUEST_EMBEDDER_EXCLUSIONS: Arc<HashSet<String>>;
66}
67
68/// Run one request with daemon-only embedding models excluded from registry access.
69pub fn scope_request_embedder_exclusions<F>(
70 excluded: Vec<String>,
71 future: F,
72) -> impl Future<Output = F::Output>
73where
74 F: Future,
75{
76 REQUEST_EMBEDDER_EXCLUSIONS.scope(Arc::new(excluded.into_iter().collect()), future)
77}
78
79/// Carry the current request's embedder exclusions into a spawned task.
80pub fn inherit_request_embedder_scope<F>(future: F) -> impl Future<Output = F::Output>
81where
82 F: Future,
83{
84 let exclusions = REQUEST_EMBEDDER_EXCLUSIONS.try_with(Arc::clone).ok();
85 async move {
86 match exclusions {
87 Some(exclusions) => REQUEST_EMBEDDER_EXCLUSIONS.scope(exclusions, future).await,
88 None => future.await,
89 }
90 }
91}
92
93fn request_excludes_embedder(name: &str) -> bool {
94 let canonical = parse_embedding_model_alias(name)
95 .map(|model| model.to_string())
96 .unwrap_or_else(|| name.to_string());
97 REQUEST_EMBEDDER_EXCLUSIONS
98 .try_with(|excluded| excluded.contains(&canonical))
99 .unwrap_or(false)
100}
101
102/// Callback type for pack-installed entity-type validators.
103///
104/// Receives `(kind, entity_type)` and returns the normalised type string,
105/// or `RuntimeError::InvalidInput` if the type is not registered for that kind.
106/// When `entity_type` is `None`, the implementation must return `Ok(None)`.
107pub type EntityTypeValidatorFn =
108 Arc<dyn Fn(&str, Option<&str>) -> Result<Option<String>, RuntimeError> + Send + Sync>;
109
110/// Pack-aggregated entity-kind update hooks: `(entity kind, hook)` for every
111/// kind whose owning pack both declares it and registers a `KindHook`.
112///
113/// Named rather than written inline because the runtime stores it behind an
114/// `Arc<RwLock<..>>` and passes it across the transport boundary, so the bare
115/// form appears three times and reads as noise at each one.
116pub type EntityKindHooks = Vec<(String, Arc<dyn KindHook>)>;
117
118/// Callback type for a pack-installed note-mutation hook.
119///
120/// Invoked by `update_note` (when the note's text/embedding actually
121/// changed) and `delete_note` (soft or hard) with `(note_kind, note_id)`,
122/// after the mutation has been durably applied. Returns a boxed future so
123/// the hook can await async cache-invalidation work (e.g.
124/// `khive-pack-memory`'s ANN warm-cache generation bump) without
125/// `khive-runtime` depending on any pack crate: dependencies point the
126/// other way, so the runtime exposes an extension point and the pack
127/// installs into it, same shape as `EntityTypeValidatorFn`, just async.
128pub type NoteMutationHookFn = Arc<
129 dyn Fn(String, uuid::Uuid) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>
130 + Send
131 + Sync,
132>;
133
134/// Callback type for a pack-installed note-write validator.
135///
136/// The pack that owns a note kind carrying derivable identity installs one so
137/// that the identity is a function of the authorization token rather than of
138/// caller input, on every write path including direct callers that bypass the
139/// handler layer — same rationale as [`EntityTypeValidatorFn`], which exists
140/// for exactly that reason on the entity side.
141///
142/// Kinds the installing pack does not own must be returned unchanged: the slot
143/// is single-occupancy (like `note_mutation_hook`), so a validator that
144/// rewrote foreign kinds would silently govern every other pack's notes.
145pub type NoteWriteValidatorFn = Arc<
146 dyn Fn(&str, &str, Option<serde_json::Value>) -> Result<Option<serde_json::Value>, RuntimeError>
147 + Send
148 + Sync,
149>;
150
151/// Immutable identity for a non-text vector store owned by a pack consumer.
152///
153/// This does not register an [`crate::EmbedderProvider`]. It gives a pack that
154/// performs its own governed inference a narrow path to a namespace-scoped
155/// Khive vector table while keeping model-key and dimension validation at the
156/// runtime boundary.
157#[derive(Clone, Debug, Eq, PartialEq)]
158pub struct NamedVectorIdentity {
159 model_key: String,
160 model_name: String,
161 dimensions: usize,
162}
163
164struct CachedNamedVectorStores {
165 identity: NamedVectorIdentity,
166 by_namespace: HashMap<String, Arc<dyn VectorStore>>,
167}
168
169fn check_cached_named_vector_identity(
170 cached: &NamedVectorIdentity,
171 requested: &NamedVectorIdentity,
172) -> RuntimeResult<()> {
173 if cached.dimensions() != requested.dimensions() {
174 return Err(RuntimeError::InvalidInput(format!(
175 "named vector model_key {:?} is already bound to {} dimensions, expected {}",
176 requested.model_key(),
177 cached.dimensions(),
178 requested.dimensions()
179 )));
180 }
181 if cached.model_name() != requested.model_name() {
182 return Err(RuntimeError::InvalidInput(format!(
183 "named vector model_key {:?} is already bound to a different active model identity",
184 requested.model_key()
185 )));
186 }
187 Ok(())
188}
189
190impl NamedVectorIdentity {
191 const MAX_MODEL_KEY_BYTES: usize = 128;
192 const MAX_MODEL_NAME_BYTES: usize = 512;
193
194 /// Validate and construct a named vector identity.
195 pub fn new(
196 model_key: impl Into<String>,
197 model_name: impl Into<String>,
198 dimensions: usize,
199 ) -> RuntimeResult<Self> {
200 let model_key = model_key.into();
201 let model_name = model_name.into();
202 if model_key.is_empty()
203 || model_key.len() > Self::MAX_MODEL_KEY_BYTES
204 || !model_key
205 .chars()
206 .all(|c| c.is_ascii_alphanumeric() || c == '_')
207 {
208 return Err(RuntimeError::InvalidInput(format!(
209 "named vector model_key must be 1..={} bytes of ASCII alphanumeric/underscore",
210 Self::MAX_MODEL_KEY_BYTES
211 )));
212 }
213 if model_name.trim().is_empty()
214 || model_name.trim() != model_name
215 || model_name.len() > Self::MAX_MODEL_NAME_BYTES
216 {
217 return Err(RuntimeError::InvalidInput(format!(
218 "named vector model_name must be 1..={} bytes with no surrounding whitespace",
219 Self::MAX_MODEL_NAME_BYTES
220 )));
221 }
222 if !(1..=8192).contains(&dimensions) {
223 return Err(RuntimeError::InvalidInput(format!(
224 "named vector dimensions must be in 1..=8192, got {dimensions}"
225 )));
226 }
227 Ok(Self {
228 model_key,
229 model_name,
230 dimensions,
231 })
232 }
233
234 pub fn model_key(&self) -> &str {
235 &self.model_key
236 }
237
238 pub fn model_name(&self) -> &str {
239 &self.model_name
240 }
241
242 pub fn dimensions(&self) -> usize {
243 self.dimensions
244 }
245}
246
247pub use crate::config::{
248 assert_captured_db_anchor_consistent, assert_db_anchor_consistent, expand_tilde,
249 parse_pack_list, resolve_db_anchor, resolve_project_actor_id, runtime_config_from_khive_config,
250 BackendId, BackendIdError, NamespaceToken, RuntimeConfig,
251};
252
253// ---- KhiveRuntime ----
254
255/// Composable runtime handle used by the MCP server.
256///
257/// Wraps a `StorageBackend` and provides namespace-scoped accessor methods
258/// Snapshot of the main runtime's embedder wiring (registry handle, default
259/// name, and the config model fields the default-resolution path reads).
260/// Carried by secondary-pack runtimes; consumed by [`KhiveRuntime::core`].
261#[derive(Clone)]
262struct CoreEmbedderState {
263 registry: Arc<std::sync::RwLock<crate::embedder_registry::EmbedderRegistry>>,
264 default_embedder_name: Arc<str>,
265 engines: Option<Vec<crate::engine_config::EngineConfig>>,
266 embedding_model: Option<EmbeddingModel>,
267 additional_embedding_models: Vec<EmbeddingModel>,
268}
269
270#[derive(Clone)]
271struct NoteKindEntry {
272 name: String,
273 embedding_policy: crate::NoteEmbeddingPolicy,
274 registered: bool,
275}
276
277/// An already-open serving backend eligible for operator diagnostics.
278/// Aliases of the same canonical database file share one entry.
279#[derive(Clone)]
280pub struct OpenedDiagnosticBackend {
281 pub backend_names: Vec<String>,
282 pub canonical_path: Option<PathBuf>,
283 pub pool: Arc<ConnectionPool>,
284}
285
286struct LateOpenedDiagnosticBackend {
287 backend_names: Vec<&'static str>,
288 pool: Weak<ConnectionPool>,
289}
290
291fn same_diagnostic_database(a: &OpenedDiagnosticBackend, b: &OpenedDiagnosticBackend) -> bool {
292 match (a.pool.canonical_path(), b.pool.canonical_path()) {
293 (Some(a), Some(b)) => a == b,
294 (None, None) => Arc::ptr_eq(&a.pool, &b.pool),
295 _ => false,
296 }
297}
298
299/// for each storage capability, plus a lazily-loaded embedder.
300#[derive(Clone)]
301pub struct KhiveRuntime {
302 pub(crate) visibility_receipts: Arc<crate::visibility_receipts::ReceiptCapability>,
303 pub(crate) visibility_cutover: Arc<crate::visibility_receipts::ReceiptCutover>,
304 core_visibility_cutover: Arc<crate::visibility_receipts::ReceiptCutover>,
305 backend: Arc<StorageBackend>,
306 /// Successful named-vector bindings and their namespace-scoped stores.
307 /// Shared by runtime clones so repeated reads do not enter the writer or
308 /// rescan the vector table after the first verified binding.
309 named_vector_stores: Arc<RwLock<HashMap<String, CachedNamedVectorStores>>>,
310 /// The main backend's cache, used when a secondary runtime creates a
311 /// `core()` handle. It must never reuse a secondary backend's store.
312 core_named_vector_stores: Option<Arc<RwLock<HashMap<String, CachedNamedVectorStores>>>>,
313 /// When `Some`, holds the main backend so that `core()` can return a
314 /// main-bound runtime handle without constructing a new connection.
315 /// `None` when this runtime is already bound to the main backend.
316 core_backend: Option<Arc<StorageBackend>>,
317 config: RuntimeConfig,
318 outbound_email_policy: crate::OutboundEmailPolicy,
319 /// All SQLite backends declared by the host process, including those
320 /// assigned to other packs. The code pack fences these from ingest.
321 declared_backend_db_paths: Arc<[PathBuf]>,
322 /// Pools opened during serving host composition, grouped by canonical
323 /// database file.
324 diagnostic_backends: Arc<[OpenedDiagnosticBackend]>,
325 /// Pools opened later by this serving runtime and its pack handles.
326 /// Weak references keep diagnostics from extending their lifetime.
327 late_diagnostic_backends: Arc<Mutex<Vec<LateOpenedDiagnosticBackend>>>,
328 /// ADR-118 exact-leg policy, sampled once at runtime construction.
329 /// Request-time memory/knowledge serving must never re-read the process
330 /// environment because tests and embedded runtimes share one process.
331 ann_fresh_tail_enabled: bool,
332 /// Pack-extensible embedder registry.
333 ///
334 /// Shared across clones via `Arc<RwLock<_>>` so that
335 /// [`register_embedder`](Self::register_embedder) after clone is visible
336 /// to all handles. Built-in lattice models are pre-registered during
337 /// construction; packs may add more via [`PackRuntime::register_embedders`].
338 embedder_registry: Arc<std::sync::RwLock<crate::embedder_registry::EmbedderRegistry>>,
339 default_embedder_name: Arc<str>,
340 /// The MAIN runtime's embedder wiring, carried by secondary-backend
341 /// runtimes so that `core()`-routed writes embed with the main runtime's
342 /// models even when this pack's own registry is empty (`no_embed`).
343 /// `None` on the main runtime, and on secondaries wired before the boot
344 /// path calls [`with_core_embedders_from`](Self::with_core_embedders_from)
345 /// — `core()` then falls back to this runtime's own embedder state.
346 core_embedders: Option<CoreEmbedderState>,
347 /// Pack-extensible edge endpoint rules. Shared across clones
348 /// via `Arc<RwLock<_>>`; installed once by the transport after the
349 /// `VerbRegistry` is built. Empty until installed
350 edge_rules: Arc<RwLock<Vec<EdgeEndpointRule>>>,
351 /// Pack-aggregated valid entity kinds and note-kind policy entries.
352 ///
353 /// Installed by the transport layer after building the `VerbRegistry`.
354 /// When non-empty, `create_entity`, `create_note_inner`, and `import_kg`
355 /// reject kinds not in these sets. When empty (no packs loaded, e.g.
356 /// bare runtime in unit tests), kind validation is skipped — the pack
357 /// handler layer is the primary enforcement point.
358 valid_entity_kinds: Arc<RwLock<Vec<String>>>,
359 valid_note_kinds: Arc<RwLock<Vec<NoteKindEntry>>>,
360 /// Pack-installed entity-type validator.
361 ///
362 /// When `Some`, `create_many` calls this function to validate and normalise
363 /// each `(kind, entity_type)` pair before writing. When `None` (bare runtime
364 /// without packs), entity-type validation is skipped — the pack handler layer
365 /// is the primary enforcement point, same as for `valid_entity_kinds`.
366 entity_type_validator: Arc<RwLock<Option<EntityTypeValidatorFn>>>,
367 /// Pack-installed note-mutation hook.
368 ///
369 /// When `Some`, `update_note` (on text change) and `delete_note` (soft
370 /// or hard) call this after the mutation is durably applied, so a pack
371 /// that caches derived state keyed by note content (e.g. `khive-pack-memory`'s
372 /// warm ANN index) can invalidate/advance its own generation counter even
373 /// when the mutation arrived through a different pack's verb (e.g. KG's
374 /// `update`/`delete` on a `kind="memory"` note) that has no dependency on
375 /// the reacting pack. `None` when no pack installs one (bare runtime, or
376 /// no pack cares about note-mutation notifications) — the call becomes a
377 /// no-op check of an `Option`.
378 note_mutation_hook: Arc<RwLock<Option<NoteMutationHookFn>>>,
379 /// Backend-matched, pack-installed ANN source for note search. Absent on a
380 /// bare runtime or when no memory graph provider serves this backend.
381 note_search_ann_provider: Arc<RwLock<Option<Arc<dyn NoteSearchAnnProvider>>>>,
382 /// Pack-installed note-write validator.
383 ///
384 /// When `Some`, every runtime note-materialisation site that accepts
385 /// caller-supplied `properties` routes them through this function before
386 /// the `Note` is built, so a pack-owned identity property is derived from
387 /// the authorization token instead of trusted from caller input. `None`
388 /// on a bare runtime (no packs) — the properties pass through unchanged.
389 note_write_validator: Arc<RwLock<Option<NoteWriteValidatorFn>>>,
390 /// Pack-installed entity-kind update-validation hooks (issue #2943).
391 ///
392 /// Every `(entity kind, hook)` pair for which an owning pack declares
393 /// the entity kind and registers a `KindHook`, aggregated once by
394 /// `VerbRegistry::entity_kind_hooks` and installed by the transport
395 /// after the registry is built — same timing and rationale as
396 /// `entity_type_validator`: `khive-runtime` does not hold a
397 /// `VerbRegistry`, so this is the extension point that lets
398 /// `prepare_guarded_entity_update` reach a pack's `KindHook` on the
399 /// generic entity `update` path, the counterpart to
400 /// `prepare_note_update_hook` on the note side. Empty until installed
401 /// (bare runtime, or no pack registers an entity-kind hook), which
402 /// leaves the dispatch a no-op.
403 entity_kind_hooks: Arc<RwLock<EntityKindHooks>>,
404 /// Pack-owned note kinds — every note kind declared by a pack other than
405 /// the generic-CRUD pack, installed by the transport from the registry
406 /// (see `VerbRegistry::pack_owned_note_kinds`). Records of these kinds are
407 /// maintained by their owning pack's own verbs, so `update`'s `properties`
408 /// patch is refused on them at the runtime layer and their owned identity
409 /// properties survive a `merge` unchanged. Empty until installed (bare
410 /// runtime), which leaves both rules inert.
411 pack_owned_note_kinds: Arc<RwLock<Vec<String>>>,
412 /// The immutable runtime-owned store/hydration pair (ADR-160 D3).
413 ///
414 /// Boot resolves one store and constructs exactly one shared hydrator,
415 /// then installs that same `Arc` on every runtime handle. The one-shot
416 /// slot rejects replacement so pack runtimes cannot silently split the
417 /// aggregate byte budget after startup. Bare runtimes leave it unset.
418 blob_hydrator: Arc<OnceLock<Arc<crate::blob::BlobHydrator>>>,
419 /// Pack-registered custom fusion executors (ADR-012), keyed by the name
420 /// carried in `FusionStrategy::Custom { name, .. }`.
421 ///
422 /// Unlike `entity_type_validator`/`note_mutation_hook` (single-occupancy —
423 /// one pack owns the slot), multiple packs each register their own named
424 /// strategy under this shared map, so it is keyed rather than a bare
425 /// `Option`. Empty until a pack calls
426 /// [`register_fusion_strategy`](Self::register_fusion_strategy); an
427 /// unregistered `Custom` name at dispatch time is
428 /// `RuntimeError::UnknownFusionStrategy`, never a silent fallback.
429 fusion_executors: Arc<RwLock<HashMap<String, Arc<dyn crate::fusion::FusionExecutor>>>>,
430}
431
432impl KhiveRuntime {
433 /// Create a new runtime with the given config.
434 ///
435 /// The config's `db_path` is used to open or create the SQLite backend.
436 /// This direct constructor is intended for fresh/current single-backend
437 /// databases and tests. Production and multi-backend hosts must use the
438 /// async khive-mcp/kkernel builders so secondary inventory and any
439 /// application-assisted V21 cutover complete before serving. The
440 /// [`from_backend`](Self::from_backend) seam is likewise only for an
441 /// already-prepared backend.
442 pub fn new(mut config: RuntimeConfig) -> RuntimeResult<Self> {
443 let wal_ceiling = config.resolve_wal_ceiling_policy(false)?;
444 let disk_guard = config.resolve_disk_guard_policy(false)?;
445 // Refuse a missing lock directory before the constructor below creates
446 // the database's parent directory.
447 let volume_lock_dir = if config.db_path.is_some() {
448 let configured = config.volume_lock_dir.clone();
449 Some(khive_db::require_volume_lock_dir(configured)?)
450 } else {
451 None
452 };
453 Self::new_with_file_backend(config, true, |path| {
454 StorageBackend::sqlite_with_max_readers_and_policies(
455 path,
456 None,
457 wal_ceiling,
458 disk_guard.expect("file-backed disk policy"),
459 volume_lock_dir.expect("file-backed volume-lock directory"),
460 )
461 })
462 }
463
464 /// Construct a fixture runtime with a small concurrent reader pool.
465 #[cfg(any(test, feature = "test-internals"))]
466 pub fn new_for_test(mut config: RuntimeConfig) -> RuntimeResult<Self> {
467 let wal_ceiling = config.resolve_wal_ceiling_policy(false)?;
468 let disk_guard = config.resolve_disk_guard_policy(false)?;
469 Self::new_with_file_backend(config, true, |path| {
470 StorageBackend::sqlite_for_test_with_policies(
471 path,
472 wal_ceiling,
473 disk_guard.expect("file-backed disk policy"),
474 )
475 })
476 }
477
478 pub(crate) fn new_with_file_backend(
479 mut config: RuntimeConfig,
480 create_parent: bool,
481 open_file: impl FnOnce(&std::path::Path) -> Result<StorageBackend, khive_db::SqliteError>,
482 ) -> RuntimeResult<Self> {
483 config.prepare_engines()?;
484 #[cfg(unix)]
485 crate::events_split::socket_path::validate_configured_events_socket(&config)?;
486 #[cfg(all(test, target_os = "macos"))]
487 ensure_in_process_test_nofile_limit();
488 let backend = match &config.db_path {
489 Some(path) => {
490 if let Some(parent) = path.parent().filter(|_| create_parent) {
491 std::fs::create_dir_all(parent).ok();
492 }
493 open_file(path)?
494 }
495 None => {
496 if config.wal_ceiling_configured_bytes != 0 || config.wal_ceiling_bytes != 0 {
497 return Err(khive_db::SqliteError::InvalidConfig(
498 "nonzero wal_ceiling_bytes requires a file-backed SQLite backend".into(),
499 )
500 .into());
501 }
502 StorageBackend::memory()?
503 }
504 };
505 // Writable backends migrate before handlers touch the DB. A detected
506 // read-only snapshot is validated at the current schema version without
507 // attempting migration DDL.
508 let schema_version = backend.prepare_core_schema()?;
509 if schema_version < khive_db::migrations::ATTACHMENT_CUTOVER_VERSION {
510 return Err(khive_db::SqliteError::InvalidData(
511 "database requires the host application-assisted V21 attachment cutover; \
512 start through khive-mcp/kkernel boot instead of constructing KhiveRuntime \
513 directly"
514 .into(),
515 )
516 .into());
517 }
518 if !backend.is_read_only() {
519 register_configured_embedding_models(&backend, &config)?;
520 }
521 Ok(Self::assemble_from_backend(
522 Arc::new(backend),
523 config,
524 false,
525 ))
526 }
527
528 /// Open a runtime for read-only inspection (no model registration, no DB creation).
529 ///
530 /// File-backed databases are opened with SQLite read-only/query-only flags
531 /// and must already be at this build's current schema version. No migrations
532 /// or configured-model registration writes are attempted. A `None` path
533 /// retains the historical ephemeral in-memory behavior for tests.
534 pub fn new_readonly(mut config: RuntimeConfig) -> RuntimeResult<Self> {
535 let wal_ceiling = config.resolve_wal_ceiling_policy(true)?;
536 Self::new_readonly_with_file_backend(config, |path| {
537 StorageBackend::sqlite_read_only_with_max_readers_and_wal_ceiling(
538 path,
539 None,
540 wal_ceiling,
541 )
542 })
543 }
544
545 /// Construct a read-only fixture runtime with a small reader pool.
546 #[cfg(any(test, feature = "test-internals"))]
547 pub fn new_readonly_for_test(mut config: RuntimeConfig) -> RuntimeResult<Self> {
548 let wal_ceiling = config.resolve_wal_ceiling_policy(true)?;
549 Self::new_readonly_with_file_backend(config, |path| {
550 StorageBackend::sqlite_read_only_with_max_readers_and_wal_ceiling(
551 path,
552 Some(2),
553 wal_ceiling,
554 )
555 })
556 }
557
558 fn new_readonly_with_file_backend(
559 mut config: RuntimeConfig,
560 open_file: impl FnOnce(&std::path::Path) -> Result<StorageBackend, khive_db::SqliteError>,
561 ) -> RuntimeResult<Self> {
562 config.prepare_engines()?;
563 #[cfg(unix)]
564 crate::events_split::socket_path::validate_configured_events_socket(&config)?;
565 #[cfg(all(test, target_os = "macos"))]
566 ensure_in_process_test_nofile_limit();
567 let backend = match &config.db_path {
568 Some(path) => open_file(path)?,
569 None => {
570 if config.wal_ceiling_configured_bytes != 0 || config.wal_ceiling_bytes != 0 {
571 return Err(khive_db::SqliteError::InvalidConfig(
572 "nonzero wal_ceiling_bytes requires a file-backed SQLite backend".into(),
573 )
574 .into());
575 }
576 StorageBackend::memory()?
577 }
578 };
579 backend.prepare_core_schema()?;
580 Ok(Self::assemble_from_backend(
581 Arc::new(backend),
582 config,
583 false,
584 ))
585 }
586
587 /// Construct a runtime from an already-opened backend.
588 ///
589 /// This is a low-level, infallible assembly seam for already-prepared
590 /// multi-backend deployments. It does not migrate or require completion of
591 /// the V21 attachment cutover, and it does not read the backend: receipt
592 /// admission is checked by the first receipt operation and cached once it
593 /// succeeds. Production hosts must first run the async kkernel/khive-mcp
594 /// coordinator and must not expose a server over a pending or incomplete
595 /// backend. Prefer [`Self::from_prepared_backend`] when constructing one
596 /// fallible host runtime.
597 ///
598 /// All storage access is through the provided `backend`; the config's
599 /// ordered embedding peers are prepared before assembly.
600 ///
601 /// # Panics
602 ///
603 /// Panics if the engine configuration is invalid. Use
604 /// [`Self::try_from_backend`] to handle configuration errors at startup.
605 pub fn from_backend(backend: Arc<StorageBackend>, config: RuntimeConfig) -> Self {
606 Self::try_from_backend(backend, config)
607 .expect("from_backend requires a valid embedding engine configuration")
608 }
609
610 /// Assemble an already-opened backend after validating its engine config.
611 ///
612 /// This retains [`Self::from_backend`]'s warning-only storage-model
613 /// registration policy; use [`Self::from_prepared_backend`] when those
614 /// registration failures must also abort startup.
615 pub fn try_from_backend(
616 backend: Arc<StorageBackend>,
617 mut config: RuntimeConfig,
618 ) -> RuntimeResult<Self> {
619 config.prepare_engines()?;
620 if !backend.is_read_only() {
621 if let Err(err) = register_configured_embedding_models(&backend, &config) {
622 tracing::warn!(error = %err, "failed to register configured embedding models");
623 }
624 }
625 Ok(Self::assemble_from_backend(backend, config, false))
626 }
627
628 /// Construct a single-backend runtime after a host boot coordinator has
629 /// completed schema preparation and any application-assisted cutover.
630 ///
631 /// Unlike [`Self::from_backend`], configured embedding-model registration
632 /// is fallible here, preserving [`Self::new`]'s single-backend startup
633 /// semantics. This method never runs migrations itself.
634 pub fn from_prepared_backend(
635 backend: Arc<StorageBackend>,
636 mut config: RuntimeConfig,
637 ) -> RuntimeResult<Self> {
638 config.prepare_engines()?;
639 if backend.attachment_cutover_status()?
640 != khive_db::migrations::AttachmentCutoverStatus::Complete
641 {
642 return Err(khive_db::SqliteError::InvalidData(
643 "from_prepared_backend requires a complete V21 attachment cutover".into(),
644 )
645 .into());
646 }
647 backend.validate_memory_visibility_cutover()?;
648 if !backend.is_read_only() {
649 register_configured_embedding_models(&backend, &config)?;
650 }
651 Ok(Self::assemble_from_backend(backend, config, true))
652 }
653
654 fn assemble_from_backend(
655 backend: Arc<StorageBackend>,
656 config: RuntimeConfig,
657 cutover_validated: bool,
658 ) -> Self {
659 if config.backend_id.as_str() == BackendId::MAIN {
660 backend.pool().main_pool_generation();
661 }
662 let ann_fresh_tail_enabled = crate::config::ann_fresh_tail_enabled_from_env();
663 let (registry, default_embedder_name) = build_embedder_registry(&config);
664 let visibility_receipts = Arc::new(
665 crate::visibility_receipts::ReceiptCapability::from_config(&config),
666 );
667 let visibility_cutover = Arc::new(crate::visibility_receipts::ReceiptCutover::new(
668 backend.clone(),
669 cutover_validated,
670 ));
671 Self {
672 visibility_receipts,
673 core_visibility_cutover: visibility_cutover.clone(),
674 visibility_cutover,
675 backend,
676 named_vector_stores: Arc::new(RwLock::new(HashMap::new())),
677 core_named_vector_stores: None,
678 core_backend: None,
679 config,
680 outbound_email_policy: Default::default(),
681 declared_backend_db_paths: Vec::new().into(),
682 diagnostic_backends: Vec::new().into(),
683 late_diagnostic_backends: Arc::new(Mutex::new(Vec::new())),
684 ann_fresh_tail_enabled,
685 embedder_registry: Arc::new(std::sync::RwLock::new(registry)),
686 default_embedder_name,
687 core_embedders: None,
688 edge_rules: Arc::new(RwLock::new(Vec::new())),
689 valid_entity_kinds: Arc::new(RwLock::new(Vec::new())),
690 valid_note_kinds: Arc::new(RwLock::new(Vec::new())),
691 entity_type_validator: Arc::new(RwLock::new(None)),
692 note_mutation_hook: Arc::new(RwLock::new(None)),
693 note_search_ann_provider: Arc::new(RwLock::new(None)),
694 note_write_validator: Arc::new(RwLock::new(None)),
695 entity_kind_hooks: Arc::new(RwLock::new(Vec::new())),
696 pack_owned_note_kinds: Arc::new(RwLock::new(Vec::new())),
697 blob_hydrator: Arc::new(OnceLock::new()),
698 fusion_executors: Arc::new(RwLock::new(HashMap::new())),
699 }
700 }
701
702 /// Wire this runtime as a secondary-backend runtime pointing at `core`.
703 ///
704 /// After this call, `self.core()` returns a handle to `core` rather than
705 /// cloning `self`. The caller (the boot path, not pack code) is responsible
706 /// for passing the correct main backend.
707 /// Binding a different core clears its named-vector cache and prior main
708 /// embedder wiring. Call [`Self::with_core_embedders_from`] with the new main
709 /// runtime after rebinding when core-routed writes require its embedders.
710 ///
711 /// Panics in debug builds if `self.config.backend_id == BackendId::MAIN`,
712 /// because the main runtime does not need a core pointer.
713 pub fn with_core_backend(mut self, core: Arc<StorageBackend>) -> Self {
714 debug_assert_ne!(
715 self.config.backend_id.as_str(),
716 BackendId::MAIN,
717 "with_core_backend must not be called on the main runtime"
718 );
719 core.pool().main_pool_generation();
720 if self.visibility_cutover.is_bound_to(&core) {
721 self.core_visibility_cutover = self.visibility_cutover.clone();
722 } else if !self.core_visibility_cutover.is_bound_to(&core) {
723 self.core_visibility_cutover = Arc::new(
724 crate::visibility_receipts::ReceiptCutover::new(core.clone(), false),
725 );
726 }
727 if self
728 .core_backend
729 .as_ref()
730 .is_some_and(|previous| !Arc::ptr_eq(previous, &core))
731 {
732 self.core_named_vector_stores = None;
733 self.core_embedders = None;
734 }
735 if self.core_named_vector_stores.is_none() {
736 self.core_named_vector_stores = Some(Arc::new(RwLock::new(HashMap::new())));
737 }
738 self.core_backend = Some(core);
739 self
740 }
741
742 /// Carry the main runtime's embedder wiring for `core()`-routed writes.
743 ///
744 /// Boot-path companion to [`with_core_backend`](Self::with_core_backend).
745 /// Without it, `core()` shares this pack runtime's own embedder registry —
746 /// which under `[packs.<name>] no_embed = true` is empty, so core-routed
747 /// concept writes would silently skip embedding on the shared graph.
748 pub fn with_core_embedders_from(mut self, main: &KhiveRuntime) -> Self {
749 debug_assert!(
750 main.core_backend.is_none(),
751 "with_core_embedders_from takes the MAIN runtime"
752 );
753 self.core_embedders = Some(CoreEmbedderState {
754 registry: main.embedder_registry.clone(),
755 default_embedder_name: main.default_embedder_name.clone(),
756 engines: main.config.engines.clone(),
757 embedding_model: main.config.embedding_model,
758 additional_embedding_models: main.config.additional_embedding_models.clone(),
759 });
760 self.core_named_vector_stores = Some(main.named_vector_stores.clone());
761 if Arc::ptr_eq(&self.backend, &main.backend) {
762 self.named_vector_stores = main.named_vector_stores.clone();
763 }
764 self
765 }
766
767 /// Return a runtime handle bound to the main (shared-graph) backend.
768 ///
769 /// When `self` is already the main runtime (`core_backend` is `None`),
770 /// this returns a clone of `self` — no new backend reference is acquired.
771 ///
772 /// When `self` is a secondary-backend runtime (`core_backend` is `Some`),
773 /// this returns a new `KhiveRuntime` backed by the main
774 /// `Arc<StorageBackend>` and sharing all registry state (`embedder_registry`,
775 /// `edge_rules`, `valid_entity_kinds`, `valid_note_kinds`,
776 /// `entity_type_validator`, `note_mutation_hook`, `entity_kind_hooks`) with `self`.
777 /// No database I/O occurs; no embedding models are reloaded.
778 ///
779 /// Use `core()` for notes and entities that must reside in the shared graph
780 /// so that `memory.recall`, cross-pack search, and `annotates` edges work.
781 /// Use `self` (or `self.sql()`) for pack-auxiliary bulk tables.
782 ///
783 /// Handlers that call `core()` more than once per request or loop should bind
784 /// `let core = self.core();` once and reuse it, since each call clones
785 /// `RuntimeConfig` (a heap-allocated struct containing `Vec<String>` fields).
786 pub fn core(&self) -> KhiveRuntime {
787 match &self.core_backend {
788 // A main-assigned pack runtime has no core pointer, but may still
789 // carry main's embedder wiring: with `no_embed` its OWN registry
790 // is empty, and core-routed concept writes must embed regardless
791 // of which backend the pack was assigned to.
792 None => match &self.core_embedders {
793 None => self.clone(),
794 Some(core_embedders) => {
795 let mut core = self.clone();
796 core.config.engines = core_embedders.engines.clone();
797 core.config.embedding_model = core_embedders.embedding_model;
798 core.config.additional_embedding_models =
799 core_embedders.additional_embedding_models.clone();
800 core.embedder_registry = core_embedders.registry.clone();
801 core.default_embedder_name = core_embedders.default_embedder_name.clone();
802 core.core_embedders = None;
803 core
804 }
805 },
806 Some(main_arc) => {
807 let mut core_config = self.config.clone();
808 core_config.backend_id = BackendId::main();
809 // Core-routed writes embed with the MAIN runtime's wiring when
810 // the boot path supplied it (see `with_core_embedders_from`);
811 // both the registry handle and the config model fields must
812 // come from main, since default-model resolution reads
813 // `config.embedding_model` (`resolve_embedding_model`).
814 let (embedder_registry, default_embedder_name) = match &self.core_embedders {
815 Some(core_embedders) => {
816 core_config.engines = core_embedders.engines.clone();
817 core_config.embedding_model = core_embedders.embedding_model;
818 core_config.additional_embedding_models =
819 core_embedders.additional_embedding_models.clone();
820 (
821 core_embedders.registry.clone(),
822 core_embedders.default_embedder_name.clone(),
823 )
824 }
825 None => (
826 self.embedder_registry.clone(),
827 self.default_embedder_name.clone(),
828 ),
829 };
830 KhiveRuntime {
831 visibility_receipts: self.visibility_receipts.clone(),
832 visibility_cutover: self.core_visibility_cutover.clone(),
833 core_visibility_cutover: self.core_visibility_cutover.clone(),
834 backend: main_arc.clone(),
835 named_vector_stores: self
836 .core_named_vector_stores
837 .clone()
838 .unwrap_or_else(|| Arc::new(RwLock::new(HashMap::new()))),
839 core_named_vector_stores: None,
840 core_backend: None,
841 config: core_config,
842 outbound_email_policy: self.outbound_email_policy.clone(),
843 declared_backend_db_paths: self.declared_backend_db_paths.clone(),
844 diagnostic_backends: self.diagnostic_backends.clone(),
845 late_diagnostic_backends: self.late_diagnostic_backends.clone(),
846 ann_fresh_tail_enabled: self.ann_fresh_tail_enabled,
847 embedder_registry,
848 default_embedder_name,
849 core_embedders: None,
850 edge_rules: self.edge_rules.clone(),
851 valid_entity_kinds: self.valid_entity_kinds.clone(),
852 valid_note_kinds: self.valid_note_kinds.clone(),
853 entity_type_validator: self.entity_type_validator.clone(),
854 note_mutation_hook: self.note_mutation_hook.clone(),
855 note_search_ann_provider: self.note_search_ann_provider.clone(),
856 note_write_validator: self.note_write_validator.clone(),
857 entity_kind_hooks: self.entity_kind_hooks.clone(),
858 pack_owned_note_kinds: self.pack_owned_note_kinds.clone(),
859 blob_hydrator: self.blob_hydrator.clone(),
860 fusion_executors: self.fusion_executors.clone(),
861 }
862 }
863 }
864 }
865
866 /// Create an in-memory runtime (for tests and ephemeral use).
867 pub fn memory() -> RuntimeResult<Self> {
868 Self::new(RuntimeConfig {
869 db_path: None,
870 packs: vec!["kg".to_string()],
871 brain_profile: None,
872 actor_id: None,
873 ..RuntimeConfig::no_embeddings()
874 })
875 }
876
877 /// Return the [`BackendId`] for this runtime's backend.
878 ///
879 /// Used by `SubstrateCoordinator` in `kkernel`
880 /// to identify which backend owns a given node, and to detect cross-backend merges.
881 pub fn backend_id(&self) -> &BackendId {
882 &self.config.backend_id
883 }
884
885 /// Whether two runtime handles share one opened physical store. The host
886 /// deduplicates same-path aliases; separately opened hard-link aliases
887 /// compare the identity pinned when SQLite opened each file.
888 pub fn shares_backend_storage_with(&self, other: &Self) -> bool {
889 if Arc::ptr_eq(&self.backend, &other.backend) {
890 return true;
891 }
892 #[cfg(any(unix, windows))]
893 {
894 matches!(
895 (
896 self.backend.pool().opened_file_identity_record(),
897 other.backend.pool().opened_file_identity_record()
898 ),
899 (Some(left), Some(right)) if left == right
900 )
901 }
902 #[cfg(not(any(unix, windows)))]
903 {
904 false
905 }
906 }
907
908 /// Install only pools the host actually opened. A bare runtime defaults
909 /// to its already-open main pool without opening or creating another file.
910 pub fn with_diagnostic_backends(mut self, backends: Arc<[OpenedDiagnosticBackend]>) -> Self {
911 self.diagnostic_backends = backends;
912 self
913 }
914
915 /// Share late-opened diagnostics with pack runtimes from the same serving
916 /// host. Independent runtimes retain independent observers.
917 pub fn with_diagnostic_observer_from(mut self, main: &KhiveRuntime) -> Self {
918 self.late_diagnostic_backends = Arc::clone(&main.late_diagnostic_backends);
919 self
920 }
921
922 fn register_late_diagnostic_pool(
923 &self,
924 backend_name: &'static str,
925 pool: &Arc<ConnectionPool>,
926 ) {
927 let mut backends = self
928 .late_diagnostic_backends
929 .lock()
930 .unwrap_or_else(std::sync::PoisonError::into_inner);
931 backends.retain(|entry| entry.pool.strong_count() > 0);
932 if let Some(existing) = backends.iter_mut().find(|entry| {
933 entry
934 .pool
935 .upgrade()
936 .is_some_and(|live| Arc::ptr_eq(&live, pool))
937 }) {
938 if !existing.backend_names.contains(&backend_name) {
939 existing.backend_names.push(backend_name);
940 }
941 return;
942 }
943 backends.push(LateOpenedDiagnosticBackend {
944 backend_names: vec![backend_name],
945 pool: Arc::downgrade(pool),
946 });
947 }
948
949 fn live_late_diagnostic_backends(&self) -> Vec<OpenedDiagnosticBackend> {
950 let mut backends = self
951 .late_diagnostic_backends
952 .lock()
953 .unwrap_or_else(std::sync::PoisonError::into_inner);
954 let mut live = Vec::new();
955 backends.retain(|entry| {
956 let Some(pool) = entry.pool.upgrade() else {
957 return false;
958 };
959 live.push(OpenedDiagnosticBackend {
960 backend_names: entry
961 .backend_names
962 .iter()
963 .map(|name| name.to_string())
964 .collect(),
965 canonical_path: pool.canonical_path().map(PathBuf::from),
966 pool,
967 });
968 true
969 });
970 live
971 }
972
973 pub fn diagnostic_backends(&self) -> Arc<[OpenedDiagnosticBackend]> {
974 let mut opened: Vec<OpenedDiagnosticBackend> = if self.diagnostic_backends.is_empty() {
975 let main = self.core().backend.pool_arc();
976 vec![OpenedDiagnosticBackend {
977 backend_names: vec![BackendId::MAIN.to_string()],
978 canonical_path: main.canonical_path().map(PathBuf::from),
979 pool: main,
980 }]
981 } else {
982 self.diagnostic_backends.iter().cloned().collect()
983 };
984 for late in self.live_late_diagnostic_backends() {
985 if let Some(existing) = opened
986 .iter_mut()
987 .find(|existing| same_diagnostic_database(existing, &late))
988 {
989 for name in late.backend_names {
990 if !existing.backend_names.contains(&name) {
991 existing.backend_names.push(name);
992 }
993 }
994 } else {
995 opened.push(late);
996 }
997 }
998 opened.into()
999 }
1000
1001 /// Whether this runtime selects the vector arm for a hybrid search —
1002 /// true exactly when a default embedding model is configured. Single
1003 /// source of truth for the policy every fan-out and single-backend
1004 /// dispatch path uses to report `arm_participation`/`vector_selected`.
1005 pub fn vector_arm_selected(&self) -> bool {
1006 self.config.embedding_model.is_some()
1007 }
1008
1009 /// Return a reference to the underlying storage backend.
1010 ///
1011 /// This is an embedder/infrastructure surface (connection pools, schema
1012 /// plans, diagnostics). Stores obtained from it are NOT wrapped by the
1013 /// message-evidence policy that [`Self::notes`] enforces: an embedder
1014 /// holding the backend already holds root-equivalent access to the
1015 /// database file, so the policy boundary sits at the typed accessors
1016 /// pack code uses, not here. Pack code must not take note stores from
1017 /// this surface.
1018 pub fn backend(&self) -> &StorageBackend {
1019 &self.backend
1020 }
1021
1022 /// Whether this runtime's bound backend is explicitly or filesystem-mode
1023 /// detected read-only.
1024 pub fn is_read_only(&self) -> bool {
1025 self.backend.is_read_only()
1026 }
1027
1028 /// Return the directory containing the backend's database file, or `None`
1029 /// for an in-memory backend.
1030 pub fn backend_data_dir(&self) -> Option<std::path::PathBuf> {
1031 self.backend.data_dir()
1032 }
1033
1034 /// Root directory for this database's ANN segment tree (`<db-file>.ann/`
1035 /// beside the file), or `None` for an in-memory backend. Scoped to the
1036 /// database file itself so two databases sharing a parent directory can
1037 /// never adopt each other's segments.
1038 pub fn backend_ann_root(&self) -> Option<std::path::PathBuf> {
1039 self.backend.ann_root()
1040 }
1041
1042 /// Writer-contention, graph-edge integrity, and WAL/checkpoint diagnostics
1043 /// (ADR-091/ADR-135 operator surface): pooled writer and audit-failure
1044 /// counters, build identity, duplicate edge-ID and list-ledger counts,
1045 /// checkpoint counters, a PASSIVE checkpoint probe, WAL file size, and
1046 /// explicitly qualified WAL-pin census. Not write-free: the
1047 /// PASSIVE probe may backfill WAL frames into the database (normal
1048 /// checkpoint I/O). It never changes logical state, escalates to TRUNCATE,
1049 /// creates a missing database file, or deletes sidecar evidence — see
1050 /// `khive_db::diagnostics` for the narrowings that make those claims hold.
1051 ///
1052 /// Always targets the *main* backend via [`Self::core`], regardless of
1053 /// which backend this runtime handle is bound to, so a report never
1054 /// describes a database this handle is not the canonical owner of.
1055 pub async fn db_diagnostics(&self) -> RuntimeResult<khive_db::diagnostics::DbDiagnostics> {
1056 // No `VerbRegistry` handle is reachable from a bare `KhiveRuntime`
1057 // (the audit-batch seam is owned by whichever registry was built
1058 // over this runtime's `EventStore`, not by the runtime itself), so
1059 // the batch-health fields report unavailable with a reason here.
1060 // Callers that hold the registry — e.g. the `db_diagnostics` verb
1061 // handler — use `Self::db_diagnostics_with_audit_metrics` with
1062 // `VerbRegistry::audit_batch_metrics()` instead.
1063 self.db_diagnostics_with_audit_metrics(None).await
1064 }
1065
1066 /// As [`Self::db_diagnostics`], but with the caller supplying the
1067 /// ADR-133 audit-batch health counters from whichever `VerbRegistry`
1068 /// owns the seam over this runtime's `EventStore` (typically
1069 /// `VerbRegistry::audit_batch_metrics()`). `None` behaves identically to
1070 /// [`Self::db_diagnostics`].
1071 pub async fn db_diagnostics_with_audit_metrics(
1072 &self,
1073 runtime_audit_batch_metrics: Option<khive_db::diagnostics::RuntimeAuditBatchMetrics>,
1074 ) -> RuntimeResult<khive_db::diagnostics::DbDiagnostics> {
1075 let pool = self.core().backend.pool_arc();
1076 // Match housekeeping's compiled legacy-record fallback (ADR-091
1077 // Amendment 6), independent of checkpoint or local sweep overrides.
1078 let legacy_sweep_interval = khive_db::SessionSweepConfig::default().interval;
1079 let build_hash = crate::build_info::BUILD_INFO
1080 .is_stamped()
1081 .then_some(crate::build_info::BUILD_INFO.source_revision);
1082 let build = khive_db::diagnostics::BuildIdentity::from_env(
1083 crate::build_info::PACKAGE_VERSION,
1084 build_hash,
1085 );
1086
1087 let mut report = khive_db::diagnostics::collect_with_runtime_audit_metrics_interruptibly(
1088 pool,
1089 build,
1090 legacy_sweep_interval,
1091 crate::pack::audit_append_failure_count(),
1092 runtime_audit_batch_metrics,
1093 )
1094 .await
1095 .map_err(RuntimeError::from)?;
1096 report.writer_contention.audit_obligation_append_failures =
1097 Some(crate::pack::audit_obligation_append_failure_count());
1098 report
1099 .writer_contention
1100 .audit_obligation_append_failures_unavailable_reason = None;
1101 let (ann_routes, fallback_routes) = crate::note_search_ann::route_totals();
1102 report.note_search_ann_route_total = ann_routes;
1103 report.note_search_fallback_route_total = fallback_routes;
1104 Ok(report)
1105 }
1106
1107 /// Collect the same per-file report as the primary diagnostic surface for
1108 /// one opened backend. The process identity comes from main; invoking
1109 /// `ProcessIdentity::current` on a secondary would miscount main generations.
1110 /// `collect_size_composition = false` skips the `dbstat` walk (see
1111 /// `khive_db::diagnostics::collect_with_runtime_audit_metrics_for_process_interruptibly`).
1112 pub async fn db_diagnostics_for_opened_backend_with_audit_metrics(
1113 &self,
1114 backend: &OpenedDiagnosticBackend,
1115 runtime_audit_batch_metrics: Option<khive_db::diagnostics::RuntimeAuditBatchMetrics>,
1116 collect_size_composition: bool,
1117 ) -> RuntimeResult<khive_db::diagnostics::DbDiagnostics> {
1118 let main_pool = self.core().backend.pool_arc();
1119 let process = khive_db::diagnostics::ProcessIdentity::current(&main_pool);
1120 let build_hash = crate::build_info::BUILD_INFO
1121 .is_stamped()
1122 .then_some(crate::build_info::BUILD_INFO.source_revision);
1123 let build = khive_db::diagnostics::BuildIdentity::from_env(
1124 crate::build_info::PACKAGE_VERSION,
1125 build_hash,
1126 );
1127 let mut report =
1128 khive_db::diagnostics::collect_with_runtime_audit_metrics_for_process_interruptibly(
1129 Arc::clone(&backend.pool),
1130 build,
1131 process,
1132 khive_db::SessionSweepConfig::default().interval,
1133 crate::pack::audit_append_failure_count(),
1134 runtime_audit_batch_metrics,
1135 collect_size_composition,
1136 )
1137 .await
1138 .map_err(RuntimeError::from)?;
1139 report.writer_contention.audit_obligation_append_failures =
1140 Some(crate::pack::audit_obligation_append_failure_count());
1141 report
1142 .writer_contention
1143 .audit_obligation_append_failures_unavailable_reason = None;
1144 let (ann_routes, fallback_routes) = crate::note_search_ann::route_totals();
1145 report.note_search_ann_route_total = ann_routes;
1146 report.note_search_fallback_route_total = fallback_routes;
1147 Ok(report)
1148 }
1149
1150 // ---- Store accessors (token-scoped) ----
1151
1152 /// Get an EntityStore scoped to the token's namespace.
1153 pub fn entities(&self, token: &NamespaceToken) -> RuntimeResult<Arc<dyn EntityStore>> {
1154 Ok(self
1155 .backend
1156 .entities_for_namespace(token.namespace().as_str())?)
1157 }
1158
1159 /// Get a GraphStore scoped to the token's namespace.
1160 pub fn graph(&self, token: &NamespaceToken) -> RuntimeResult<Arc<dyn GraphStore>> {
1161 Ok(self
1162 .backend
1163 .graph_for_namespace(token.namespace().as_str())?)
1164 }
1165
1166 /// Get a NoteStore scoped to the token's namespace.
1167 ///
1168 /// Wrapped in `note_store_guard::PolicyEnforcingNoteStore`, which
1169 /// refuses any insert/upsert of a `kind = "message"` note carrying
1170 /// `quarantined` / `channel_kind` / `channel_slug` — the transport-owned
1171 /// evidence `comm.health` trusts at face value — and refuses patching
1172 /// those keys through the property-mutation seams on any note kind, so
1173 /// the guard cannot be sidestepped by inserting a clean message note and
1174 /// patching the evidence onto it afterward. Full-row writes also preserve
1175 /// existing channel-health coordinates while allowing heartbeat metadata to
1176 /// change. The trusted channel-ingest path does not go through this accessor; see
1177 /// `Self::raw_notes` and [`Self::try_create_note_as_trusted_ingest`].
1178 pub fn notes(&self, token: &NamespaceToken) -> RuntimeResult<Arc<dyn NoteStore>> {
1179 Ok(crate::note_store_guard::PolicyEnforcingNoteStore::wrap(
1180 self.raw_notes(token)?,
1181 ))
1182 }
1183
1184 /// Get the unwrapped, policy-free NoteStore scoped to the token's namespace.
1185 ///
1186 /// Bypasses `note_store_guard::PolicyEnforcingNoteStore`. Callers
1187 /// within this crate that have already enforced the reserved-transport-
1188 /// property policy themselves (namely `try_create_note_impl`, which
1189 /// applies it conditionally based on whether the caller presented a
1190 /// [`crate::pack::ChannelIngestCapability`]) use this to reach storage
1191 /// directly rather than run a redundant, less-informed check. Not exposed
1192 /// outside this crate — every other caller must use [`Self::notes`].
1193 pub(crate) fn raw_notes(&self, token: &NamespaceToken) -> RuntimeResult<Arc<dyn NoteStore>> {
1194 Ok(self
1195 .backend
1196 .notes_for_namespace(token.namespace().as_str())?)
1197 }
1198
1199 /// Return the role-keyed attachment substrate on the canonical main backend.
1200 ///
1201 /// Attachment rows are the process-shared BlobStore's sole SQL liveness
1202 /// authority. A runtime bound directly to a secondary pack backend must call
1203 /// [`Self::core`] first; accepting a secondary mutation here would create a
1204 /// reference that the main-database GC sweep cannot see or fence.
1205 pub fn attachments(&self) -> RuntimeResult<Arc<dyn AttachmentStore>> {
1206 if self.config.backend_id.as_str() != BackendId::MAIN {
1207 return Err(RuntimeError::InvalidInput(format!(
1208 "attachments are owned by the canonical main backend; runtime backend {:?} must route through KhiveRuntime::core()",
1209 self.config.backend_id.as_str()
1210 )));
1211 }
1212 Ok(self.backend.attachments()?)
1213 }
1214
1215 /// Get an EventStore scoped to the token's namespace.
1216 ///
1217 /// When the events-daemon split (ADR-170) is configured, the store routes
1218 /// by append class: the ADR-133 idempotent audit-batch lane — the
1219 /// measured bulk of event write volume — persists to the events database
1220 /// (forwarded over the events daemon socket in daemon deployments, or
1221 /// opened directly in embedded/one-shot contexts), while plain appends
1222 /// stay on this runtime's backend, keeping every raw-SQL consumer of the
1223 /// legacy `events` table (schedule provenance, kg projection guards,
1224 /// GraphQuery's substrate union) correct by construction. Reads merge
1225 /// both stores. Unconfigured runtimes (tests, in-memory) keep the legacy
1226 /// main-store behavior. Every returned store is decorated at this typed
1227 /// accessor boundary so append callers cannot override the namespace or
1228 /// actor resolved into the sealed authorization token.
1229 pub fn events(&self, token: &NamespaceToken) -> RuntimeResult<Arc<dyn EventStore>> {
1230 Ok(crate::event_store_guard::AttributedEventStore::wrap(
1231 self.raw_events_for_namespace(token.namespace().as_str())?,
1232 token,
1233 ))
1234 }
1235
1236 /// The event sidecar inherits the already-open MAIN pool's WAL policy.
1237 /// A secondary pack's config or pool does not govern this shared file.
1238 fn events_wal_ceiling_policy(&self) -> khive_db::WalCeilingPolicy {
1239 self.core_backend
1240 .as_ref()
1241 .unwrap_or(&self.backend)
1242 .pool()
1243 .config()
1244 .wal_ceiling
1245 }
1246
1247 /// Build the undecorated event store used only by the registry's trusted
1248 /// audit composer, which stamps from each resolved `GateRequest` before
1249 /// enqueueing. Pack/runtime call sites must use [`Self::events`] instead.
1250 pub(crate) fn raw_events_for_namespace(
1251 &self,
1252 namespace: &str,
1253 ) -> RuntimeResult<Arc<dyn EventStore>> {
1254 let legacy = self.backend.events_for_namespace(namespace)?;
1255 match &self.config.events_split {
1256 None => Ok(legacy),
1257 Some(split) => {
1258 // Read-only is decided before the transport question: a
1259 // read-only runtime must neither create nor schema-initialize
1260 // an events database, and it must not forward writes to the
1261 // events daemon either — a socket in the config describes the
1262 // deployment, not this process's authority. Serve merged
1263 // reads from a read-only open of the sidecar when it exists
1264 // (the storage-layer read-only binding refuses any write that
1265 // slips through), and the legacy store alone otherwise: no
1266 // sidecar on disk means no lane rows exist, so minting the
1267 // file just to read nothing from it would be a write in
1268 // disguise.
1269 if self.backend.is_read_only() {
1270 if !split.db_path.exists() {
1271 return Ok(legacy);
1272 }
1273 let lane_backend = crate::events_split::direct_backend_with_policies(
1274 &split.db_path,
1275 true,
1276 Some(self.backend.pool().config().max_readers),
1277 self.events_wal_ceiling_policy(),
1278 None,
1279 self.events_volume_lock_dir(),
1280 )?;
1281 self.register_late_diagnostic_pool("events", &lane_backend.pool_arc());
1282 let lane = lane_backend.events_for_namespace(namespace)?;
1283 return Ok(Arc::new(crate::events_split::SplitEventStore::new(
1284 legacy, lane,
1285 )));
1286 }
1287 let lane: Arc<dyn EventStore> = match &split.socket_path {
1288 #[cfg(unix)]
1289 Some(socket) => {
1290 let client = crate::events_split::client_for(socket)?;
1291 Arc::new(crate::events_split::ForwardingEventStore::new(
1292 namespace, client,
1293 ))
1294 }
1295 #[cfg(not(unix))]
1296 Some(_socket) => {
1297 return Err(RuntimeError::InvalidInput(
1298 "events-daemon socket forwarding requires a Unix platform; \
1299 configure the events split in direct mode here"
1300 .to_string(),
1301 ));
1302 }
1303 None => {
1304 let lane_backend = crate::events_split::direct_backend_with_policies(
1305 &split.db_path,
1306 false,
1307 Some(self.backend.pool().config().max_readers),
1308 self.events_wal_ceiling_policy(),
1309 Some(self.events_disk_guard_policy()?),
1310 self.events_volume_lock_dir(),
1311 )?;
1312 self.register_late_diagnostic_pool("events", &lane_backend.pool_arc());
1313 lane_backend.events_for_namespace(namespace)?
1314 }
1315 };
1316 Ok(Arc::new(crate::events_split::SplitEventStore::new(
1317 legacy, lane,
1318 )))
1319 }
1320 }
1321 }
1322
1323 /// Get the raw SQL access capability (for ad-hoc queries).
1324 pub fn sql(&self) -> Arc<dyn SqlAccess> {
1325 self.backend.sql()
1326 }
1327
1328 /// SQL access to the events-split sidecar database for read purposes,
1329 /// when the split (ADR-170) is configured and the sidecar exists on
1330 /// disk. `None` means every event row lives in the legacy `events`
1331 /// table, so a raw-SQL consumer needs no second lookup. Consumers that
1332 /// resolve an event by id or hex prefix against `self.sql()` must also
1333 /// consult this store on a miss: the audit-batch lane's rows live only
1334 /// in the sidecar.
1335 ///
1336 /// A writable runtime opens the sidecar through the ordinary writable
1337 /// binding: WAL supports one-writer-many-readers, and the read-only
1338 /// binding's frozen-snapshot guard refuses any sidecar with a live
1339 /// writer's `-shm` beside it — exactly the live-deployment case these
1340 /// reads exist for. A read-only runtime keeps the read-only open (it
1341 /// must neither create nor schema-initialize a sidecar), which serves
1342 /// genuinely frozen snapshots and refuses live ones, matching the
1343 /// read-only arm of `events()`. Never creates a sidecar as a side
1344 /// effect of a read.
1345 pub fn events_sidecar_sql_read_only(&self) -> RuntimeResult<Option<Arc<dyn SqlAccess>>> {
1346 match &self.config.events_split {
1347 None => Ok(None),
1348 Some(split) => {
1349 if !split.db_path.exists() {
1350 return Ok(None);
1351 }
1352 let backend = if self.backend.is_read_only() {
1353 crate::events_split::direct_backend_with_policies(
1354 &split.db_path,
1355 true,
1356 Some(self.backend.pool().config().max_readers),
1357 self.events_wal_ceiling_policy(),
1358 None,
1359 self.events_volume_lock_dir(),
1360 )?
1361 } else {
1362 crate::events_split::direct_backend_with_policies(
1363 &split.db_path,
1364 false,
1365 Some(self.backend.pool().config().max_readers),
1366 self.events_wal_ceiling_policy(),
1367 Some(self.events_disk_guard_policy()?),
1368 self.events_volume_lock_dir(),
1369 )?
1370 };
1371 self.register_late_diagnostic_pool("events", &backend.pool_arc());
1372 Ok(Some(backend.sql()))
1373 }
1374 }
1375 }
1376
1377 /// Get a VectorStore for the configured embedding model, scoped to the token's namespace.
1378 ///
1379 /// Returns `Unconfigured("embedding_model")` if no model is set.
1380 pub fn vectors(
1381 &self,
1382 token: &NamespaceToken,
1383 ) -> RuntimeResult<Arc<dyn khive_storage::VectorStore>> {
1384 let model = self.resolve_embedding_model(None)?;
1385 self.vectors_for_embedding_model(token, model)
1386 }
1387
1388 /// Get a VectorStore for a specific named embedding model, scoped to the token's namespace.
1389 ///
1390 /// Accepts both built-in lattice model names/aliases and custom provider names
1391 /// registered via [`register_embedder`](Self::register_embedder). Lattice names
1392 /// are routed through the enum-backed path; custom provider names use the
1393 /// provider's declared `dimensions()` directly so that the vector store key
1394 /// is consistent with how vectors were written during `remember`/`recall`.
1395 pub fn vectors_for_model(
1396 &self,
1397 token: &NamespaceToken,
1398 model_name: &str,
1399 ) -> RuntimeResult<Arc<dyn khive_storage::VectorStore>> {
1400 let (model_name, dims) = self.vector_model_metadata(model_name)?;
1401 Ok(self.backend.vectors_for_namespace(
1402 &sanitize_key(&model_name),
1403 &model_name,
1404 dims,
1405 token.namespace().as_str(),
1406 )?)
1407 }
1408
1409 /// Resolve the storage identity and declared dimensions together so guarded
1410 /// SQL publication agrees with VectorStore, including built-in aliases.
1411 pub(crate) fn vector_model_metadata(&self, model_name: &str) -> RuntimeResult<(String, usize)> {
1412 if request_excludes_embedder(model_name) {
1413 return Err(crate::RuntimeError::UnknownModel(model_name.to_string()));
1414 }
1415 let registry = self
1416 .embedder_registry
1417 .read()
1418 .map_err(|_| crate::RuntimeError::Internal("embedder registry lock poisoned".into()))?;
1419 if let Some(model) = parse_embedding_model_alias(model_name) {
1420 // Only proceed via the lattice path if this model is actually in the
1421 // registry; otherwise fall through to the custom-provider path.
1422 let key = model.to_string();
1423 if registry.contains(&key) {
1424 return Ok((key, model.dimensions()));
1425 }
1426 }
1427 registry
1428 .get_provider(model_name)
1429 .map(|provider| (model_name.to_owned(), provider.dimensions()))
1430 .ok_or_else(|| crate::RuntimeError::UnknownModel(model_name.to_string()))
1431 }
1432
1433 /// Get a namespace-scoped vector store for a pack-owned immutable identity.
1434 ///
1435 /// The table key is syntactically validated by [`NamedVectorIdentity`]. This
1436 /// accessor additionally verifies the table's actual sqlite-vec dimension
1437 /// declaration and every persisted `embedding_model` value before returning
1438 /// the store, so reusing one key for incompatible descriptor geometry or
1439 /// semantics fails before a caller can replace rows.
1440 pub async fn vectors_for_named_identity(
1441 &self,
1442 token: &NamespaceToken,
1443 identity: &NamedVectorIdentity,
1444 ) -> RuntimeResult<Arc<dyn VectorStore>> {
1445 let namespace = token.namespace().as_str();
1446 {
1447 let cached = self.named_vector_stores.read().map_err(|_| {
1448 RuntimeError::Internal("named vector store cache lock poisoned".into())
1449 })?;
1450 if let Some(entry) = cached.get(identity.model_key()) {
1451 check_cached_named_vector_identity(&entry.identity, identity)?;
1452 if let Some(store) = entry.by_namespace.get(namespace) {
1453 return Ok(Arc::clone(store));
1454 }
1455 }
1456 }
1457 let store = self.backend.vectors_for_namespace(
1458 identity.model_key(),
1459 identity.model_name(),
1460 identity.dimensions(),
1461 namespace,
1462 )?;
1463
1464 let table = format!("vec_{}", identity.model_key());
1465 let mut reader = self.sql().reader().await?;
1466 let dimension_row = reader
1467 .query_row(SqlStatement {
1468 sql: "SELECT sql FROM sqlite_schema WHERE type = 'table' AND name = ?1".to_string(),
1469 params: vec![SqlValue::Text(table.clone())],
1470 label: Some("runtime_named_vector_dimension".to_string()),
1471 })
1472 .await?
1473 .ok_or_else(|| {
1474 RuntimeError::Internal(format!(
1475 "named vector table {table} has no sqlite_schema declaration"
1476 ))
1477 })?;
1478 let table_ddl = match dimension_row.get("sql") {
1479 Some(SqlValue::Text(value)) => value,
1480 other => {
1481 return Err(RuntimeError::Internal(format!(
1482 "named vector table {table} returned invalid schema metadata: {other:?}"
1483 )))
1484 }
1485 };
1486 let declared_dimensions = vector_dimensions_from_ddl(table_ddl).ok_or_else(|| {
1487 RuntimeError::Internal(format!(
1488 "named vector table {table} has no parseable embedding dimension"
1489 ))
1490 })?;
1491 if declared_dimensions != identity.dimensions() {
1492 return Err(RuntimeError::InvalidInput(format!(
1493 "named vector model_key {:?} is already bound to {declared_dimensions} dimensions, expected {}",
1494 identity.model_key(),
1495 identity.dimensions()
1496 )));
1497 }
1498
1499 let stored_models = reader
1500 .query_all(SqlStatement {
1501 sql: format!(
1502 "SELECT DISTINCT embedding_model FROM {table} ORDER BY embedding_model LIMIT 2"
1503 ),
1504 params: vec![],
1505 label: Some("runtime_named_vector_model_identity".to_string()),
1506 })
1507 .await?;
1508 for row in stored_models {
1509 let stored = match row.get("embedding_model") {
1510 Some(SqlValue::Text(value)) => value,
1511 other => {
1512 return Err(RuntimeError::Internal(format!(
1513 "named vector table {table} returned invalid model identity metadata: {other:?}"
1514 )))
1515 }
1516 };
1517 if stored != identity.model_name() {
1518 return Err(RuntimeError::InvalidInput(format!(
1519 "named vector model_key {:?} already contains model {stored:?}, cannot bind it to {:?}",
1520 identity.model_key(),
1521 identity.model_name()
1522 )));
1523 }
1524 }
1525
1526 self.backend
1527 .register_embedding_model(
1528 identity.model_key(),
1529 identity.model_name(),
1530 identity.model_key(),
1531 identity.dimensions() as u32,
1532 )
1533 .map_err(|error| {
1534 if matches!(
1535 &error,
1536 khive_db::SqliteError::Rusqlite(rusqlite::Error::SqliteFailure(code, _))
1537 | khive_db::SqliteError::Write { source: rusqlite::Error::SqliteFailure(code, _), .. }
1538 if code.code == rusqlite::ErrorCode::ConstraintViolation
1539 ) {
1540 RuntimeError::InvalidInput(format!(
1541 "named vector model_key {:?} is already bound to a different active model identity",
1542 identity.model_key()
1543 ))
1544 } else {
1545 RuntimeError::from(error)
1546 }
1547 })?;
1548
1549 let mut cached = self
1550 .named_vector_stores
1551 .write()
1552 .map_err(|_| RuntimeError::Internal("named vector store cache lock poisoned".into()))?;
1553 let entry = cached
1554 .entry(identity.model_key().to_owned())
1555 .or_insert_with(|| CachedNamedVectorStores {
1556 identity: identity.clone(),
1557 by_namespace: HashMap::new(),
1558 });
1559 check_cached_named_vector_identity(&entry.identity, identity)?;
1560 entry
1561 .by_namespace
1562 .insert(namespace.to_owned(), Arc::clone(&store));
1563 Ok(store)
1564 }
1565
1566 /// Output dimensions for a named embedding model, resolved from the
1567 /// embedder registry alone — no storage access. Mirrors
1568 /// [`vectors_for_model`](Self::vectors_for_model)'s resolution order:
1569 /// lattice aliases route through the enum when registered, otherwise the
1570 /// custom provider's declared `dimensions()`. `None` when no such model
1571 /// is registered.
1572 pub fn embedder_dimensions(&self, model_name: &str) -> Option<usize> {
1573 if request_excludes_embedder(model_name) {
1574 return None;
1575 }
1576 if let Some(model) = parse_embedding_model_alias(model_name) {
1577 let key = model.to_string();
1578 let in_registry = self
1579 .embedder_registry
1580 .read()
1581 .map(|reg| reg.contains(&key))
1582 .unwrap_or(false);
1583 if in_registry {
1584 return Some(model.dimensions());
1585 }
1586 }
1587 self.embedder_registry
1588 .read()
1589 .ok()?
1590 .get_provider(model_name)
1591 .map(|p| p.dimensions())
1592 }
1593
1594 fn vectors_for_embedding_model(
1595 &self,
1596 token: &NamespaceToken,
1597 model: EmbeddingModel,
1598 ) -> RuntimeResult<Arc<dyn khive_storage::VectorStore>> {
1599 Ok(self.backend.vectors_for_namespace(
1600 &vec_model_key(model),
1601 &model.to_string(),
1602 model.dimensions(),
1603 token.namespace().as_str(),
1604 )?)
1605 }
1606
1607 /// Get a TextSearch index for the entity corpus (single shared table).
1608 pub fn text(
1609 &self,
1610 token: &NamespaceToken,
1611 ) -> RuntimeResult<Arc<dyn khive_storage::TextSearch>> {
1612 let _ = token;
1613 Ok(self.backend.text("entities")?)
1614 }
1615
1616 /// Get a TextSearch index for the notes corpus (single shared table).
1617 pub fn text_for_notes(
1618 &self,
1619 token: &NamespaceToken,
1620 ) -> RuntimeResult<Arc<dyn khive_storage::TextSearch>> {
1621 let _ = token;
1622 Ok(self.backend.text("notes")?)
1623 }
1624
1625 /// Mint an authorization token for the given namespace.
1626 ///
1627 /// Consults the configured [`crate::Gate`] before minting. With the default
1628 /// `AllowAllGate` this always succeeds. When a real policy-backed gate is
1629 /// installed, this method enforces it and returns `PermissionDenied` on
1630 /// denial.
1631 ///
1632 /// The returned token's read visibility set defaults to `[ns]` — identical
1633 /// to the pre-visibility-set behaviour. Use [`Self::authorize_with_visibility`]
1634 /// to mint a token that can read additional namespaces.
1635 ///
1636 /// When `actor_id` is configured in `RuntimeConfig`, the token carries that
1637 /// actor label so that `comm.inbox` filters by `to_actor`. When
1638 /// unconfigured, the token carries `ActorRef::anonymous()` and inbox falls
1639 /// back to party-line behavior.
1640 pub fn authorize(&self, ns: Namespace) -> RuntimeResult<NamespaceToken> {
1641 let actor = crate::actor_identity::resolve_actor(self.config.actor_id.as_deref());
1642 let req = GateRequest::new(
1643 actor.clone(),
1644 ns.clone(),
1645 "authorize",
1646 serde_json::Value::Null,
1647 );
1648 match self.config.gate.check(&req) {
1649 Ok(ref decision) if decision.is_allow() => {
1650 if let khive_gate::GateDecision::Allow { ref obligations } = decision {
1651 if !obligations.is_empty() {
1652 tracing::debug!(
1653 namespace = %ns.as_str(),
1654 "authorize: obligations={:?}",
1655 obligations
1656 );
1657 }
1658 }
1659 Ok(NamespaceToken::mint_authorized(ns, actor))
1660 }
1661 Ok(khive_gate::GateDecision::Deny { reason }) => {
1662 Err(crate::RuntimeError::permission_denied("authorize", reason))
1663 }
1664 Ok(_) => Err(crate::RuntimeError::permission_denied(
1665 "authorize",
1666 "gate denied",
1667 )),
1668 Err(e) => {
1669 tracing::warn!(
1670 namespace = %ns.as_str(),
1671 error = %crate::secret_gate::bounded_masked_log_text(&e.to_string()),
1672 "authorize: gate check failed (fail-closed)"
1673 );
1674 Err(crate::RuntimeError::Internal(format!(
1675 "gate error: {}",
1676 e.wire_reason()
1677 )))
1678 }
1679 }
1680 }
1681
1682 /// Mint an authorization token with an explicit read-visibility set.
1683 ///
1684 /// `primary` is the **write namespace** — all records created via the
1685 /// returned token land there. `extra_visible` lists additional namespaces
1686 /// the token may read. The primary is always included in the visible set
1687 /// regardless of `extra_visible`.
1688 ///
1689 /// Usage (lambda:leo reading both leo and khive namespaces):
1690 /// ```rust,ignore
1691 /// let tok = rt.authorize_with_visibility(
1692 /// Namespace::parse("lambda:leo").unwrap(),
1693 /// vec![Namespace::parse("lambda:khive").unwrap()],
1694 /// )?;
1695 /// ```
1696 pub fn authorize_with_visibility(
1697 &self,
1698 primary: Namespace,
1699 extra_visible: Vec<Namespace>,
1700 ) -> RuntimeResult<NamespaceToken> {
1701 let actor = crate::actor_identity::resolve_actor(self.config.actor_id.as_deref());
1702 let req = GateRequest::new(
1703 actor.clone(),
1704 primary.clone(),
1705 "authorize",
1706 serde_json::Value::Null,
1707 );
1708 match self.config.gate.check(&req) {
1709 Ok(ref decision) if decision.is_allow() => {
1710 if let khive_gate::GateDecision::Allow { ref obligations } = decision {
1711 if !obligations.is_empty() {
1712 tracing::debug!(
1713 namespace = %primary.as_str(),
1714 "authorize_with_visibility: obligations={:?}",
1715 obligations
1716 );
1717 }
1718 }
1719 // The primary check authorizes writes to `primary` only. Each
1720 // extra namespace grants read visibility, so each one takes
1721 // its own Read-classified gate check before it may enter the
1722 // minted set — a token must never carry visibility the gate
1723 // was not asked about. Any deny or gate error refuses the
1724 // whole mint, naming the offending namespace (fail-closed).
1725 for extra in &extra_visible {
1726 let extra_req = GateRequest::new(
1727 actor.clone(),
1728 extra.clone(),
1729 "authorize.visible",
1730 serde_json::Value::Null,
1731 );
1732 match self.config.gate.check(&extra_req) {
1733 Ok(ref extra_decision) if extra_decision.is_allow() => {}
1734 Ok(khive_gate::GateDecision::Deny { reason }) => {
1735 return Err(crate::RuntimeError::permission_denied(
1736 "authorize",
1737 format!(
1738 "visibility namespace {:?} denied: {reason}",
1739 extra.as_str()
1740 ),
1741 ));
1742 }
1743 Ok(_) => {
1744 return Err(crate::RuntimeError::permission_denied(
1745 "authorize",
1746 format!("visibility namespace {:?} denied by gate", extra.as_str()),
1747 ));
1748 }
1749 Err(e) => {
1750 tracing::warn!(
1751 namespace = %extra.as_str(),
1752 error = %crate::secret_gate::bounded_masked_log_text(&e.to_string()),
1753 "authorize_with_visibility: extra-namespace gate check failed (fail-closed)"
1754 );
1755 return Err(crate::RuntimeError::Internal(format!(
1756 "gate error: {}",
1757 e.wire_reason()
1758 )));
1759 }
1760 }
1761 }
1762 Ok(NamespaceToken::mint_with_visibility(
1763 primary,
1764 extra_visible,
1765 actor,
1766 ))
1767 }
1768 Ok(khive_gate::GateDecision::Deny { reason }) => {
1769 Err(crate::RuntimeError::permission_denied("authorize", reason))
1770 }
1771 Ok(_) => Err(crate::RuntimeError::permission_denied(
1772 "authorize",
1773 "gate denied",
1774 )),
1775 Err(e) => {
1776 tracing::warn!(
1777 namespace = %primary.as_str(),
1778 error = %crate::secret_gate::bounded_masked_log_text(&e.to_string()),
1779 "authorize_with_visibility: gate check failed (fail-closed)"
1780 );
1781 Err(crate::RuntimeError::Internal(format!(
1782 "gate error: {}",
1783 e.wire_reason()
1784 )))
1785 }
1786 }
1787 }
1788
1789 /// Install the pack-aggregated edge endpoint rules.
1790 ///
1791 /// Called by the transport layer after the `VerbRegistry` is built so
1792 /// that runtime-layer edge validation can consult pack rules. Idempotent:
1793 /// later calls overwrite the previous rule set.
1794 pub fn install_edge_rules(&self, rules: Vec<EdgeEndpointRule>) {
1795 if let Ok(mut guard) = self.edge_rules.write() {
1796 *guard = rules;
1797 }
1798 }
1799
1800 /// Install an already-paired blob hydrator into this runtime.
1801 ///
1802 /// Reinstalling the exact same `Arc` is idempotent. A different pair is
1803 /// rejected: replacing it would split or reset the aggregate admission
1804 /// budget while requests may still hold leases.
1805 pub fn install_blob_hydrator(
1806 &self,
1807 hydrator: Arc<crate::blob::BlobHydrator>,
1808 ) -> RuntimeResult<()> {
1809 if hydrator.budget_bytes() != self.config.blob_hydration_bytes {
1810 return Err(RuntimeError::InvalidInput(format!(
1811 "blob hydrator budget {} does not match this runtime's resolved budget {}",
1812 hydrator.budget_bytes(),
1813 self.config.blob_hydration_bytes
1814 )));
1815 }
1816 // The mode gate must sit on THIS seam, not only on `install_blob_store`:
1817 // `BlobHydrator::new` is public, so without it a caller pairs a writable
1818 // store, installs it here, and `blob_store()` hands mutating pack paths
1819 // a writable store on a runtime whose declared mode is read-only.
1820 // Boot paths installing one hydrator across handles of MIXED modes —
1821 // where the blob pack's own backend mode, not each receiving
1822 // handle's, governs mutability — go through
1823 // [`Self::install_shared_blob_hydrator`] instead.
1824 if self.is_read_only() && !hydrator.enforces_read_only() {
1825 return Err(RuntimeError::InvalidInput(
1826 "this runtime is read-only: install the raw store with install_blob_store, \
1827 which wraps it so every physical mutator refuses"
1828 .to_string(),
1829 ));
1830 }
1831 self.install_blob_hydrator_slot(hydrator)
1832 }
1833
1834 /// Install a boot-shared hydrator whose mutability is governed by the
1835 /// blob runtime's own mode, not this handle's domain-store mode.
1836 ///
1837 /// ADR-160 D3 installs one hydrator `Arc` on every runtime handle a boot
1838 /// produces, and the documented multi-backend matrix includes a writable
1839 /// blob secondary beside a read-only main: there the shared hydrator is
1840 /// legitimately writable on a read-only domain handle. The mode decision
1841 /// must therefore already be encoded in the hydrator, and it must have
1842 /// been DERIVED, not declared: only hydrators built through
1843 /// [`crate::BlobHydrator::resolve_for_governing_backend`] — whose mode
1844 /// comes from the governing backend's own access mode — are accepted
1845 /// here. A hand-paired hydrator (`BlobHydrator::new` / `for_mode`) is
1846 /// refused so a safe downstream caller cannot use this seam to put a
1847 /// writable store on a read-only runtime; such callers use
1848 /// [`Self::install_blob_hydrator`], which holds hydrator mode against
1849 /// this runtime's own.
1850 ///
1851 /// This gate is a wrong-wiring guard, not an in-process sandbox: which
1852 /// backend governs is the boot host's topology assertion, and a caller
1853 /// who deliberately selects an unrelated writable backend as governing
1854 /// is outside what any runtime seam can enforce (see the trust-model
1855 /// note on [`crate::BlobHydrator::resolve_for_governing_backend`]).
1856 pub fn install_shared_blob_hydrator(
1857 &self,
1858 hydrator: Arc<crate::blob::BlobHydrator>,
1859 ) -> RuntimeResult<()> {
1860 if !hydrator.is_governed() {
1861 return Err(RuntimeError::InvalidInput(
1862 "the shared install seam accepts only hydrators whose mode was derived from a \
1863 governing backend (BlobHydrator::resolve_for_governing_backend); use \
1864 install_blob_hydrator for a hand-paired hydrator"
1865 .to_string(),
1866 ));
1867 }
1868 if hydrator.budget_bytes() != self.config.blob_hydration_bytes {
1869 return Err(RuntimeError::InvalidInput(format!(
1870 "blob hydrator budget {} does not match this runtime's resolved budget {}",
1871 hydrator.budget_bytes(),
1872 self.config.blob_hydration_bytes
1873 )));
1874 }
1875 self.install_blob_hydrator_slot(hydrator)
1876 }
1877
1878 /// Whether `candidate` is the same pairing as the installed `current`:
1879 /// the exact `Arc`, or a distinct hydrator allocation over the same raw
1880 /// store with the same budget and mode. The latter arises when two
1881 /// concurrent first installs each construct a hydrator from one raw
1882 /// store — the `OnceLock` loser must read as idempotent, not as a
1883 /// conflicting install.
1884 fn is_same_blob_pairing(
1885 current: &Arc<crate::blob::BlobHydrator>,
1886 candidate: &Arc<crate::blob::BlobHydrator>,
1887 ) -> bool {
1888 Arc::ptr_eq(current, candidate)
1889 || (Arc::ptr_eq(¤t.raw_store(), &candidate.raw_store())
1890 && current.budget_bytes() == candidate.budget_bytes()
1891 && current.enforces_read_only() == candidate.enforces_read_only())
1892 }
1893
1894 /// One-shot slot semantics shared by both install seams: first install
1895 /// wins, an equivalent pairing is idempotent, a different pairing is
1896 /// refused (replacing it would split or reset the aggregate admission
1897 /// budget while requests may still hold leases).
1898 fn install_blob_hydrator_slot(
1899 &self,
1900 hydrator: Arc<crate::blob::BlobHydrator>,
1901 ) -> RuntimeResult<()> {
1902 if let Some(current) = self.blob_hydrator.get() {
1903 return if Self::is_same_blob_pairing(current, &hydrator) {
1904 Ok(())
1905 } else {
1906 Err(RuntimeError::InvalidInput(
1907 "a different blob hydrator is already installed".to_string(),
1908 ))
1909 };
1910 }
1911
1912 match self.blob_hydrator.set(hydrator) {
1913 Ok(()) => Ok(()),
1914 Err(candidate) => {
1915 let current = self.blob_hydrator.get().ok_or_else(|| {
1916 RuntimeError::Internal(
1917 "blob hydrator install raced without a visible winner".to_string(),
1918 )
1919 })?;
1920 if Self::is_same_blob_pairing(current, &candidate) {
1921 Ok(())
1922 } else {
1923 Err(RuntimeError::InvalidInput(
1924 "a different blob hydrator is already installed".to_string(),
1925 ))
1926 }
1927 }
1928 }
1929 }
1930
1931 /// Pair and install a store using this runtime's resolved hydration budget.
1932 ///
1933 /// Boot paths that own multiple runtimes should instead construct one
1934 /// [`crate::BlobHydrator`] and call [`Self::install_blob_hydrator`] with
1935 /// the same `Arc` on every handle.
1936 pub fn install_blob_store(
1937 &self,
1938 store: Arc<dyn khive_storage::BlobStore>,
1939 ) -> RuntimeResult<()> {
1940 if let Some(current) = self.blob_hydrator.get() {
1941 // Reinstalling the same raw store is idempotent in BOTH modes:
1942 // on a read-only runtime the installed hydrator wraps the raw
1943 // store, so identity is checked against the raw handle the
1944 // hydrator remembers, not only the (possibly wrapped) paired one.
1945 let current_store = current.store();
1946 if Arc::ptr_eq(¤t_store, &store) || Arc::ptr_eq(¤t.raw_store(), &store) {
1947 return Ok(());
1948 }
1949 }
1950 // A read-only runtime holds its mode at this seam, not only during
1951 // boot resolution: an arbitrary store installed after launch is
1952 // wrapped so every physical mutator refuses while the bounded read
1953 // surface stays available. Without this, post-boot installation is a
1954 // writable bypass of the runtime's declared mode.
1955 let hydrator = if self.is_read_only() {
1956 crate::blob::BlobHydrator::new_read_only(store, self.config.blob_hydration_bytes)?
1957 } else {
1958 crate::blob::BlobHydrator::new(store, self.config.blob_hydration_bytes)?
1959 };
1960 self.install_blob_hydrator(Arc::new(hydrator))
1961 }
1962
1963 /// Return the installed shared blob hydrator, if boot configured one.
1964 pub fn blob_hydrator(&self) -> Option<Arc<crate::blob::BlobHydrator>> {
1965 self.blob_hydrator.get().cloned()
1966 }
1967
1968 /// Return the installed `BlobStore`, if the boot path resolved and
1969 /// installed one. `None` when no `[storage.blob]` selection was ever
1970 /// installed — e.g. a bare/test runtime constructed without going
1971 /// through the `khive-mcp` boot path.
1972 pub fn blob_store(&self) -> Option<Arc<dyn khive_storage::BlobStore>> {
1973 self.blob_hydrator.get().map(|hydrator| hydrator.store())
1974 }
1975
1976 /// Return the installed `BlobStore`, or `RuntimeError::Unconfigured` with
1977 /// the operator-facing message when none is installed.
1978 ///
1979 /// Packs share this conversion so an operator sees one error wherever the
1980 /// missing `[storage.blob]` configuration is first hit.
1981 pub fn require_blob_store(&self) -> RuntimeResult<Arc<dyn khive_storage::BlobStore>> {
1982 self.blob_store().ok_or_else(Self::no_blob_store)
1983 }
1984
1985 /// Return the installed shared blob hydrator, or `RuntimeError::Unconfigured`
1986 /// with the same message as [`Self::require_blob_store`] when none is
1987 /// installed. The store is read from the hydrator, so both accessors are
1988 /// unset under the same condition.
1989 pub fn require_blob_hydrator(&self) -> RuntimeResult<Arc<crate::blob::BlobHydrator>> {
1990 self.blob_hydrator().ok_or_else(Self::no_blob_store)
1991 }
1992
1993 fn no_blob_store() -> RuntimeError {
1994 RuntimeError::Unconfigured(
1995 "no BlobStore installed on this server (configure [storage.blob] in khive.toml, or \
1996 KHIVE_BLOB_ROOT)"
1997 .to_string(),
1998 )
1999 }
2000
2001 /// Install the pack-aggregated valid entity and note kinds.
2002 ///
2003 /// Called by the transport layer after the `VerbRegistry` is built so that
2004 /// runtime-layer entity/note creation and import validate kind strings against
2005 /// the merged pack vocabulary. Idempotent: later calls overwrite previous sets.
2006 ///
2007 /// When no kinds are installed (empty lists), kind validation is skipped at
2008 /// the runtime layer. The pack handler layer remains the primary enforcement
2009 /// point; this provides defense-in-depth for direct Rust callers and import.
2010 pub fn install_kind_registry(&self, entity_kinds: Vec<String>, note_kinds: Vec<String>) {
2011 if let Ok(mut guard) = self.valid_entity_kinds.write() {
2012 *guard = entity_kinds;
2013 }
2014 if let Ok(mut guard) = self.valid_note_kinds.write() {
2015 let prior = std::mem::take(&mut *guard);
2016 *guard = note_kinds
2017 .into_iter()
2018 .map(|name| NoteKindEntry {
2019 embedding_policy: prior
2020 .iter()
2021 .find(|entry| entry.name == name)
2022 .map(|entry| entry.embedding_policy)
2023 .unwrap_or_default(),
2024 name,
2025 registered: true,
2026 })
2027 .collect();
2028 }
2029 }
2030
2031 /// Install pack-declared embedding policy on the note-kind registry.
2032 /// The transport calls this for every runtime after pack registration.
2033 pub fn install_note_embedding_policies(&self, policies: &[crate::NoteEmbeddingPolicySpec]) {
2034 if let Ok(mut guard) = self.valid_note_kinds.write() {
2035 for spec in policies {
2036 if let Some(entry) = guard.iter_mut().find(|entry| entry.name == spec.kind) {
2037 entry.embedding_policy = spec.policy;
2038 } else {
2039 guard.push(NoteKindEntry {
2040 name: spec.kind.to_owned(),
2041 embedding_policy: spec.policy,
2042 registered: false,
2043 });
2044 }
2045 }
2046 }
2047 }
2048
2049 /// Registered models selected by the installed embedding policy for a note kind.
2050 /// Unknown kinds retain the all-models default.
2051 pub fn embedding_models_for_note_kind(&self, kind: &str) -> Vec<String> {
2052 let policy = self
2053 .valid_note_kinds
2054 .read()
2055 .ok()
2056 .and_then(|guard| {
2057 guard
2058 .iter()
2059 .find(|entry| entry.name == kind)
2060 .map(|entry| entry.embedding_policy)
2061 })
2062 .unwrap_or_default();
2063 let models = self.registered_embedding_model_names();
2064 match policy {
2065 crate::NoteEmbeddingPolicy::AllModels => models,
2066 crate::NoteEmbeddingPolicy::DefaultModel => {
2067 let default = self.default_embedder_name();
2068 models
2069 .into_iter()
2070 .filter(|name| name.as_str() == default)
2071 .collect()
2072 }
2073 }
2074 }
2075
2076 /// Install the pack-owned note kinds aggregated from the pack registry.
2077 ///
2078 /// Called by the transport after the `VerbRegistry` is built, same timing
2079 /// as [`install_kind_registry`](Self::install_kind_registry).
2080 pub fn install_pack_owned_note_kinds(&self, kinds: Vec<String>) {
2081 if let Ok(mut guard) = self.pack_owned_note_kinds.write() {
2082 *guard = kinds;
2083 }
2084 }
2085
2086 /// Whether `kind` is a note kind owned by a pack (see
2087 /// [`install_pack_owned_note_kinds`](Self::install_pack_owned_note_kinds)).
2088 ///
2089 /// Always `false` before the transport installs the list — a bare runtime
2090 /// has no packs, so no kind is pack-owned there.
2091 pub fn is_pack_owned_note_kind(&self, kind: &str) -> bool {
2092 self.pack_owned_note_kinds
2093 .read()
2094 .map(|g| g.iter().any(|k| k == kind))
2095 .unwrap_or(false)
2096 }
2097
2098 /// Validate that `kind` is a pack-registered entity kind.
2099 ///
2100 /// Returns `Ok(())` when no kinds are installed (bare runtime without packs).
2101 /// Returns `InvalidInput` when kinds are installed and `kind` is not among them.
2102 pub(crate) fn validate_entity_kind(&self, kind: &str) -> crate::RuntimeResult<()> {
2103 let guard = self.valid_entity_kinds.read().map_err(|_| {
2104 crate::RuntimeError::Internal("entity kind registry lock poisoned".into())
2105 })?;
2106 if guard.is_empty() {
2107 return Ok(());
2108 }
2109 if guard.iter().any(|k| k == kind) {
2110 Ok(())
2111 } else {
2112 Err(crate::RuntimeError::InvalidInput(format!(
2113 "unknown entity kind {kind:?}; valid: {}",
2114 guard.join(", ")
2115 )))
2116 }
2117 }
2118
2119 /// Validate that `kind` is a pack-registered note kind.
2120 ///
2121 /// Returns `Ok(())` when no kinds are installed (bare runtime without packs).
2122 /// Returns `InvalidInput` when kinds are installed and `kind` is not among them.
2123 pub(crate) fn validate_note_kind(&self, kind: &str) -> crate::RuntimeResult<()> {
2124 let guard = self.valid_note_kinds.read().map_err(|_| {
2125 crate::RuntimeError::Internal("note kind registry lock poisoned".into())
2126 })?;
2127 if !guard.iter().any(|entry| entry.registered) {
2128 return Ok(());
2129 }
2130 if guard
2131 .iter()
2132 .any(|entry| entry.registered && entry.name == kind)
2133 {
2134 Ok(())
2135 } else {
2136 let valid = guard
2137 .iter()
2138 .filter(|entry| entry.registered)
2139 .map(|entry| entry.name.as_str())
2140 .collect::<Vec<_>>()
2141 .join(", ");
2142 Err(crate::RuntimeError::InvalidInput(format!(
2143 "unknown note kind {kind:?}; valid: {}",
2144 valid
2145 )))
2146 }
2147 }
2148
2149 /// Install a pack-supplied entity-type validator.
2150 ///
2151 /// Called by the `KgPack` during registration so that `create_many` can validate
2152 /// `entity_type` values at the runtime layer, closing the hole where direct Rust
2153 /// callers bypass the handler-layer `validate_entity_type` check.
2154 ///
2155 /// The callback receives `(kind, entity_type)` and returns the normalised type
2156 /// string, or `RuntimeError::InvalidInput` if the type is not registered for that
2157 /// kind. Passing `entity_type = None` must return `Ok(None)`.
2158 pub fn install_entity_type_validator(&self, f: EntityTypeValidatorFn) {
2159 if let Ok(mut guard) = self.entity_type_validator.write() {
2160 *guard = Some(f);
2161 }
2162 }
2163
2164 /// Validate and normalise `entity_type` through the pack-installed validator.
2165 ///
2166 /// Returns `Ok(entity_type)` when no validator is installed (bare runtime).
2167 /// Returns `InvalidInput` when a validator is installed and rejects the type.
2168 pub(crate) fn validate_entity_type_for_kind(
2169 &self,
2170 kind: &str,
2171 entity_type: Option<&str>,
2172 ) -> crate::RuntimeResult<Option<String>> {
2173 let guard = self.entity_type_validator.read().map_err(|_| {
2174 crate::RuntimeError::Internal("entity type validator lock poisoned".into())
2175 })?;
2176 match guard.as_ref() {
2177 None => Ok(entity_type.map(str::to_string)),
2178 Some(validate) => validate(kind, entity_type),
2179 }
2180 }
2181
2182 /// Install a pack-owned note-mutation hook.
2183 ///
2184 /// Overwrites any previously-installed hook, same single-slot semantics
2185 /// as [`install_entity_type_validator`](Self::install_entity_type_validator).
2186 /// In practice only one pack (`khive-pack-memory`) installs one today;
2187 /// if a second pack ever needs this, the slot should be widened to a
2188 /// `Vec` at that point rather than silently overwritten.
2189 pub fn install_note_mutation_hook(&self, f: NoteMutationHookFn) {
2190 if let Ok(mut guard) = self.note_mutation_hook.write() {
2191 *guard = Some(f);
2192 }
2193 }
2194
2195 /// Clone read-side backend and embedder handles without retaining this
2196 /// runtime's pack callback slots. Providers stored in one of those slots
2197 /// must not hold a clone that points back to their own installation Arc,
2198 /// including indirectly through another hook's captured runtime.
2199 pub fn detached_for_note_search_ann_provider(&self) -> Self {
2200 let mut detached = self.clone();
2201 detached.note_search_ann_provider = Arc::new(RwLock::new(None));
2202 detached.note_mutation_hook = Arc::new(RwLock::new(None));
2203 detached.entity_type_validator = Arc::new(RwLock::new(None));
2204 detached.note_write_validator = Arc::new(RwLock::new(None));
2205 detached.entity_kind_hooks = Arc::new(RwLock::new(Vec::new()));
2206 detached.fusion_executors = Arc::new(RwLock::new(HashMap::new()));
2207 detached
2208 }
2209
2210 /// Install the memory pack's note-search provider only on its own opened
2211 /// backend. A same-named but separate store keeps the exact search route.
2212 pub fn install_note_search_ann_provider(&self, provider: Arc<dyn NoteSearchAnnProvider>) {
2213 if !provider.serves_backend(self) {
2214 return;
2215 }
2216 if let Ok(mut guard) = self.note_search_ann_provider.write() {
2217 *guard = Some(provider);
2218 }
2219 }
2220
2221 pub(crate) fn note_search_ann_provider(
2222 &self,
2223 ) -> RuntimeResult<Option<Arc<dyn NoteSearchAnnProvider>>> {
2224 self.note_search_ann_provider
2225 .read()
2226 .map(|guard| {
2227 guard
2228 .as_ref()
2229 .filter(|provider| provider.serves_backend(self))
2230 .cloned()
2231 })
2232 .map_err(|_| RuntimeError::Internal("note-search ANN provider lock poisoned".into()))
2233 }
2234
2235 /// Install the pack-aggregated entity-kind update hooks (issue #2943).
2236 ///
2237 /// Called by the transport after the `VerbRegistry` is built, same
2238 /// timing as [`install_kind_registry`](Self::install_kind_registry) —
2239 /// pass `registry.entity_kind_hooks()`. Idempotent: a later call
2240 /// replaces the set.
2241 pub fn install_entity_kind_hooks(&self, hooks: EntityKindHooks) {
2242 if let Ok(mut guard) = self.entity_kind_hooks.write() {
2243 *guard = hooks;
2244 }
2245 }
2246
2247 /// The installed `KindHook` for entity `kind`, if its owning pack
2248 /// registered one via [`install_entity_kind_hooks`](Self::install_entity_kind_hooks).
2249 ///
2250 /// `None` before the transport installs the aggregate (bare runtime) or
2251 /// when no pack registered a hook for this entity kind — the caller
2252 /// treats this the same as a hook whose `validate_entity_update`
2253 /// inherited the trait's `Ok(())` default.
2254 pub(crate) fn entity_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
2255 self.entity_kind_hooks.read().ok().and_then(|guard| {
2256 guard
2257 .iter()
2258 .find(|(k, _)| k == kind)
2259 .map(|(_, hook)| hook.clone())
2260 })
2261 }
2262
2263 /// Install a pack-owned note-write validator.
2264 ///
2265 /// Called during pack registration (`PackRuntime::register_note_write_validator`)
2266 /// so that the covered note-write sites carrying caller-supplied
2267 /// `properties` derive the owning pack's identity properties from the
2268 /// authorization token, closing the gap where a direct Rust caller, the
2269 /// generic `create` verb, or the proposal-apply path (which dispatches no
2270 /// pack hooks) writes them unchecked. Single-slot semantics, same as
2271 /// [`install_note_mutation_hook`](Self::install_note_mutation_hook): a
2272 /// second installing pack overwrites the first, so a validator must return
2273 /// kinds it does not own unchanged.
2274 ///
2275 /// Covered sites — each calls `derive_note_write_properties`
2276 /// before the write: `create_note_inner` (`operations.rs`, the generic
2277 /// `create` verb funnel and every other public `create_note*` variant),
2278 /// `atomic_prepare::prepare_add_note` (the proposal-apply add-note path),
2279 /// and `atomic_message::create_notes_atomic_with_report` (the atomic
2280 /// multi-note writer).
2281 ///
2282 /// NOT covered by this validator: `try_create_note` (`operations.rs`).
2283 /// `try_create_note` is deliberately excluded — its only caller path is
2284 /// `comm.ingest`, where `properties.from_actor` is the external transport
2285 /// sender named by the `from` parameter, not the authenticated caller,
2286 /// and where transport-owned quarantine/channel properties are
2287 /// legitimately established. Running the generic validator there would
2288 /// stamp every inbound message as the ingesting daemon and reject the
2289 /// evidence the trusted ingest handler just derived. `try_create_note`
2290 /// instead runs its own narrower reserved-transport-property check
2291 /// inline (`operations.rs`'s `try_create_note_impl`), which allows the
2292 /// `message`-kind transport properties only when called through
2293 /// [`Self::try_create_note_as_trusted_ingest`] with a
2294 /// [`crate::pack::ChannelIngestCapability`].
2295 ///
2296 /// The `NoteStore` returned by [`notes`](Self::notes) is covered by a
2297 /// different, narrower mechanism: it is wrapped in
2298 /// `note_store_guard::PolicyEnforcingNoteStore`, which refuses
2299 /// `upsert_note` / `upsert_notes` / `try_insert_note` /
2300 /// `replace_note_if_unchanged` calls that would write a `kind = "message"`
2301 /// note carrying `quarantined` / `channel_kind` / `channel_slug`, and
2302 /// refuses `set_note_property` / `try_patch_note_property` /
2303 /// `patch_note_property_atomic` / `update_note_properties` calls that
2304 /// would patch any of those keys onto any note — unconditionally, since
2305 /// that public accessor has no way to see a trust decision. `try_create_note_impl` itself reaches storage through
2306 /// `Self::raw_notes`, the unwrapped accessor, so its own inline check
2307 /// (which can legitimately allow those properties for trusted ingest)
2308 /// is not double-enforced or contradicted by the wrapper.
2309 /// Register a pack-defined custom fusion strategy under `name` (ADR-012).
2310 ///
2311 /// Unlike `install_entity_type_validator`/`install_note_mutation_hook`,
2312 /// this slot is keyed rather than single-occupancy: multiple packs each
2313 /// register their own named strategy, and a second registration under an
2314 /// already-used `name` replaces the first. Looked up by
2315 /// `FusionStrategy::Custom { name, .. }` at the hybrid-search dispatch
2316 /// boundary in `crate::fusion`; an unregistered name fails closed with
2317 /// `RuntimeError::UnknownFusionStrategy` rather than silently falling
2318 /// back to RRF.
2319 pub fn register_fusion_strategy(
2320 &self,
2321 name: impl Into<String>,
2322 executor: Arc<dyn crate::fusion::FusionExecutor>,
2323 ) {
2324 if let Ok(mut guard) = self.fusion_executors.write() {
2325 guard.insert(name.into(), executor);
2326 }
2327 }
2328
2329 /// Resolve a registered custom fusion executor by name.
2330 ///
2331 /// Returns `RuntimeError::UnknownFusionStrategy` when no pack has
2332 /// registered `name` — callers must invoke this before any
2333 /// empty-input/zero-limit short circuit so a misconfigured name errors
2334 /// on every call, including zero-result ones.
2335 pub(crate) fn fusion_executor(
2336 &self,
2337 name: &str,
2338 ) -> RuntimeResult<Arc<dyn crate::fusion::FusionExecutor>> {
2339 let guard = self
2340 .fusion_executors
2341 .read()
2342 .map_err(|_| RuntimeError::Internal("fusion executor registry lock poisoned".into()))?;
2343 guard
2344 .get(name)
2345 .cloned()
2346 .ok_or_else(|| RuntimeError::UnknownFusionStrategy(name.to_string()))
2347 }
2348
2349 pub fn install_note_write_validator(&self, f: NoteWriteValidatorFn) {
2350 if let Ok(mut guard) = self.note_write_validator.write() {
2351 *guard = Some(f);
2352 }
2353 }
2354
2355 /// Whether a note-write validator is installed on this runtime.
2356 ///
2357 /// Exists so a transport's own tests can assert, per boot path, that the
2358 /// documented startup sequence actually filled the slot. A missing install
2359 /// fails open and silently — an empty slot passes caller-supplied
2360 /// properties straight through, which no write site can distinguish from a
2361 /// validator that approved them — so occupancy is asserted, never assumed.
2362 pub fn has_note_write_validator(&self) -> bool {
2363 self.note_write_validator
2364 .read()
2365 .map(|g| g.is_some())
2366 .unwrap_or(false)
2367 }
2368
2369 /// Run caller-supplied note `properties` through the installed note-write
2370 /// validator, returning the properties to store.
2371 ///
2372 /// Returns them unchanged when no validator is installed (bare runtime).
2373 pub(crate) fn derive_note_write_properties(
2374 &self,
2375 kind: &str,
2376 token: &NamespaceToken,
2377 properties: Option<serde_json::Value>,
2378 ) -> RuntimeResult<Option<serde_json::Value>> {
2379 let validator = self
2380 .note_write_validator
2381 .read()
2382 .map_err(|_| RuntimeError::Internal("note write validator lock poisoned".into()))?
2383 .clone();
2384 match validator {
2385 None => Ok(properties),
2386 Some(validate) => validate(kind, &token.actor().id, properties),
2387 }
2388 }
2389
2390 /// Invoke the pack-installed note-mutation hook, if any.
2391 ///
2392 /// `kind` is the note's `kind` string (e.g. `"memory"`); `id` is the
2393 /// note's UUID. No-op when no hook is installed (bare runtime, or no
2394 /// pack cares). Errors inside the hook are the hook's own concern to
2395 /// handle/log — this call site cannot propagate a failure without
2396 /// changing `update_note`/`delete_note`'s already-committed success
2397 /// return value.
2398 pub(crate) async fn fire_note_mutation_hook(&self, kind: &str, id: uuid::Uuid) {
2399 let hook = self
2400 .note_mutation_hook
2401 .read()
2402 .ok()
2403 .and_then(|guard| guard.clone());
2404 if let Some(hook) = hook {
2405 hook(kind.to_string(), id).await;
2406 }
2407 }
2408
2409 /// Snapshot of currently-installed pack edge rules.
2410 ///
2411 /// This is the same composed rule set `validate_edge_relation_endpoints`
2412 /// consults via `pack_rule_allows` when accepting/rejecting an edge. Public
2413 /// so pack-layer error-hint code (e.g. `khive-pack-kg`'s
2414 /// `valid_relations_for_entity_pair`) can derive hints from the exact
2415 /// source the validator uses, rather than maintaining a separate
2416 /// hand-authored table that can drift out of sync.
2417 pub fn pack_edge_rules(&self) -> Vec<EdgeEndpointRule> {
2418 self.edge_rules
2419 .read()
2420 .map(|g| g.clone())
2421 .unwrap_or_default()
2422 }
2423
2424 /// Borrow the installed pack edge rules for a synchronous calculation.
2425 pub(crate) fn with_pack_edge_rules<T>(&self, f: impl FnOnce(&[EdgeEndpointRule]) -> T) -> T {
2426 match self.edge_rules.read() {
2427 Ok(rules) => f(&rules),
2428 Err(_) => f(&[]),
2429 }
2430 }
2431
2432 /// Return the name of the default embedding model (empty string if none configured).
2433 pub fn default_embedder_name(&self) -> &str {
2434 self.default_embedder_name.as_ref()
2435 }
2436
2437 /// Resolve a model name (or `None` for the default) to an `EmbeddingModel`.
2438 ///
2439 /// Returns `UnknownModel` if the name is not in the registry, or
2440 /// `Unconfigured` if `None` is passed and no default model is set.
2441 pub fn resolve_embedding_model(&self, name: Option<&str>) -> RuntimeResult<EmbeddingModel> {
2442 let model = match name {
2443 Some(raw) => parse_embedding_model_alias(raw)
2444 .ok_or_else(|| crate::RuntimeError::UnknownModel(raw.to_string()))?,
2445 None => self
2446 .config
2447 .embedding_model
2448 .ok_or_else(|| crate::RuntimeError::Unconfigured("embedding_model".into()))?,
2449 };
2450 let key = model.to_string();
2451 if request_excludes_embedder(&key) {
2452 return Err(crate::RuntimeError::UnknownModel(
2453 name.unwrap_or_else(|| self.default_embedder_name())
2454 .to_string(),
2455 ));
2456 }
2457 let contains = self
2458 .embedder_registry
2459 .read()
2460 .map(|reg| reg.contains(&key))
2461 .unwrap_or(false);
2462 if contains {
2463 Ok(model)
2464 } else {
2465 Err(crate::RuntimeError::UnknownModel(
2466 name.unwrap_or_else(|| self.default_embedder_name())
2467 .to_string(),
2468 ))
2469 }
2470 }
2471
2472 /// Names of all registered embedding models in this runtime.
2473 ///
2474 /// Includes both built-in lattice models and any custom embedders
2475 /// registered by packs via [`register_embedder`](Self::register_embedder).
2476 /// Useful for operations that must touch every model's storage (e.g.,
2477 /// scoped vector deletion on note delete). The default model is included.
2478 pub fn registered_embedding_model_names(&self) -> Vec<String> {
2479 self.embedder_registry
2480 .read()
2481 .map(|reg| {
2482 reg.names()
2483 .into_iter()
2484 .filter(|name| !request_excludes_embedder(name))
2485 .collect()
2486 })
2487 .unwrap_or_default()
2488 }
2489
2490 /// Get the lazily-initialized embedding service for the named model.
2491 ///
2492 /// Accepts both built-in lattice model names (e.g. `"all-minilm-l6-v2"`,
2493 /// `"paraphrase"`) and custom provider names registered via
2494 /// [`register_embedder`](Self::register_embedder).
2495 ///
2496 /// For lattice model names, aliases (e.g. `"paraphrase"`) are resolved to
2497 /// their canonical key before looking up the registry. For custom providers
2498 /// the name must match exactly as supplied during registration.
2499 ///
2500 /// First call for any name loads the underlying service (cold start cost);
2501 /// subsequent calls are cheap (registry caches the `Arc`).
2502 pub async fn embedder(&self, name: &str) -> RuntimeResult<Arc<dyn EmbeddingService>> {
2503 Ok(self.embedder_inner(name, None).await?.0)
2504 }
2505
2506 pub(crate) async fn embedder_with_token(
2507 &self,
2508 token: &NamespaceToken,
2509 name: &str,
2510 ) -> RuntimeResult<Arc<dyn EmbeddingService>> {
2511 Ok(self.embedder_inner(name, Some(token)).await?.0)
2512 }
2513
2514 /// Resolve the service and its document-preparation attestation from the
2515 /// same registry entry. Cloning it also avoids holding a registry lock
2516 /// across the asynchronous service resolution.
2517 pub(crate) async fn embedder_with_input_attestation(
2518 &self,
2519 name: &str,
2520 token: Option<&NamespaceToken>,
2521 ) -> RuntimeResult<(Arc<dyn EmbeddingService>, bool)> {
2522 self.embedder_inner(name, token).await
2523 }
2524
2525 /// Register a custom embedding provider with this runtime.
2526 ///
2527 /// The provider is added to the shared [`EmbedderRegistry`] so all clones
2528 /// of this runtime see the new provider immediately. Setup may replace a
2529 /// provider before it is selected for resolution. Later duplicates are
2530 /// refused and logged; use [`try_register_embedder`](Self::try_register_embedder)
2531 /// when the caller needs to handle registration errors.
2532 ///
2533 /// Packs should call this from [`crate::PackRuntime::register_embedders`] (the
2534 /// hook is invoked by the transport during pack initialisation, before the
2535 /// first verb dispatch).
2536 ///
2537 /// [`EmbedderRegistry`]: crate::embedder_registry::EmbedderRegistry
2538 pub fn register_embedder(
2539 &self,
2540 provider: impl crate::embedder_registry::EmbedderProvider + 'static,
2541 ) {
2542 if let Err(error) = self.try_register_embedder(provider) {
2543 tracing::warn!(%error, "embedder registration refused");
2544 }
2545 }
2546
2547 /// Register a custom embedding provider and return serving-duplicate or lock errors.
2548 ///
2549 /// Unlike [`register_embedder`](Self::register_embedder), this method lets
2550 /// callers fail initialization when registration cannot be completed.
2551 pub fn try_register_embedder(
2552 &self,
2553 provider: impl crate::embedder_registry::EmbedderProvider + 'static,
2554 ) -> RuntimeResult<()> {
2555 if let Some(engine) = self
2556 .config
2557 .engines
2558 .as_ref()
2559 .and_then(|engines| engines.iter().find(|engine| engine.name == provider.name()))
2560 {
2561 engine
2562 .check_dimensions(provider.dimensions())
2563 .map_err(|error| RuntimeError::InvalidInput(error.to_string()))?;
2564 }
2565 let mut registry = self
2566 .embedder_registry
2567 .write()
2568 .map_err(|_| RuntimeError::Internal("embedder registry lock poisoned".into()))?;
2569 registry.register(provider)
2570 }
2571
2572 /// Install a deterministic backend for exact-input provenance tests.
2573 /// The test adapter, not the supplied backend, owns lattice passage
2574 /// prefixing; this API is absent unless `test-internals` is enabled.
2575 #[cfg(feature = "test-internals")]
2576 pub fn register_test_audited_embedder(
2577 &self,
2578 model: EmbeddingModel,
2579 provider: impl crate::embedder_registry::EmbedderProvider + 'static,
2580 ) {
2581 self.embedder_registry
2582 .write()
2583 .expect("test embedder registry lock")
2584 .register_test_audited(model, provider);
2585 }
2586
2587 /// List registered embedding models via `SqlAccess`, routing through the
2588 /// existing connection pool rather than opening a fresh `Connection` per call.
2589 ///
2590 /// Optionally filter by `engine_name`. Returns an empty vec when the
2591 /// `_embedding_models` table does not yet exist (e.g. no migrations have run
2592 /// or no models have been registered). All other SQL errors are propagated.
2593 /// Malformed required fields, unknown statuses and invalid non-NULL optional
2594 /// dimensions/UUIDs are warned and skipped. The legacy optional timestamps
2595 /// retain their non-integer-to-None compatibility behavior.
2596 pub async fn list_embedding_models(
2597 &self,
2598 engine_filter: Option<&str>,
2599 ) -> RuntimeResult<Vec<crate::EmbeddingModelRecord>> {
2600 use khive_storage::{SqlStatement, SqlValue};
2601
2602 let (sql_text, params) = if let Some(engine) = engine_filter {
2603 (
2604 "SELECT id, engine_name, model_id, key_version, dim, output_dim, status, \
2605 activated_at, superseded_at, superseded_by, canonical_key, created_at \
2606 FROM _embedding_models WHERE engine_name = ?1 \
2607 ORDER BY engine_name, activated_at IS NULL, activated_at"
2608 .to_string(),
2609 vec![SqlValue::Text(engine.to_string())],
2610 )
2611 } else {
2612 (
2613 "SELECT id, engine_name, model_id, key_version, dim, output_dim, status, \
2614 activated_at, superseded_at, superseded_by, canonical_key, created_at \
2615 FROM _embedding_models \
2616 ORDER BY engine_name, activated_at IS NULL, activated_at"
2617 .to_string(),
2618 vec![],
2619 )
2620 };
2621
2622 let stmt = SqlStatement {
2623 sql: sql_text,
2624 params,
2625 label: Some("list_embedding_models".into()),
2626 };
2627
2628 let mut reader = self
2629 .sql()
2630 .reader()
2631 .await
2632 .map_err(crate::RuntimeError::Storage)?;
2633
2634 let rows = match reader.query_all(stmt).await {
2635 Ok(rows) => rows,
2636 Err(e) if e.to_string().contains("no such table: _embedding_models") => {
2637 return Ok(Vec::new())
2638 }
2639 Err(e) => return Err(crate::RuntimeError::Storage(e)),
2640 };
2641
2642 Ok(rows
2643 .iter()
2644 .filter_map(crate::EmbeddingModelRecord::from_sql_row)
2645 .collect())
2646 }
2647}
2648
2649fn vector_dimensions_from_ddl(ddl: &str) -> Option<usize> {
2650 let lower = ddl.to_ascii_lowercase();
2651 let suffix = lower.split_once("embedding float[")?.1;
2652 let dimension = suffix.split_once(']')?.0;
2653 if dimension.is_empty() || !dimension.bytes().all(|byte| byte.is_ascii_digit()) {
2654 return None;
2655 }
2656 dimension.parse().ok()
2657}
2658
2659// IN-CRATE TEST JUSTIFICATION: tests here cover KhiveRuntime construction helpers
2660// (in-memory backend wiring, NamespaceToken::for_namespace) that are
2661// pub(crate)-only and cannot be called from the integration test crate.
2662#[cfg(test)]
2663#[path = "runtime_tests.rs"]
2664mod tests;