Skip to main content

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(&current.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(&current_store, &store) || Arc::ptr_eq(&current.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;