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