khive_pack_knowledge/
pack.rs

1//! `KnowledgePack` struct, factory, and `PackRuntime` impl.
2
3use std::sync::Mutex;
4
5use async_trait::async_trait;
6use serde_json::Value;
7use uuid::Uuid;
8
9use khive_brain_core::SectionPosteriorState;
10use khive_runtime::pack::{PackByIdResolver, PackRuntime};
11use khive_runtime::{KhiveRuntime, NamespaceToken, Resolved, RuntimeError, VerbRegistry};
12use khive_storage::types::{SqlStatement, SqlValue};
13use khive_storage::{PhaseCancelledPayload, PhaseCompletedPayload, PhaseStartedPayload};
14use khive_types::{EventKind, HandlerDef, Pack};
15
16use crate::knowledge::util::{atom_from_row, atom_to_json, domain_from_row, domain_to_json};
17use crate::knowledge::vamana;
18use crate::knowledge::KnowledgeHandlers;
19use crate::vocab::KNOWLEDGE_HANDLERS;
20
21/// Knowledge corpus pack — atoms, domains, TF-IDF search, fold, import, and KG concept verbs.
22pub struct KnowledgePack {
23    pub(crate) runtime: KhiveRuntime,
24    pub(crate) ann: vamana::SharedAnn,
25    pub(crate) section_posteriors: Mutex<SectionPosteriorState>,
26    /// Explicit brain profile ID from config (ADR-035 §Brain profile configuration).
27    ///
28    /// Tier-1 of the 3-tier feedback resolution: when set, `knowledge.feedback` directs
29    /// feedback to this profile via `brain.feedback`. When absent, tier-2
30    /// (namespace-bound profile) and tier-3 (global prior) are tried in order.
31    pub(crate) brain_profile: Option<String>,
32}
33
34impl Pack for KnowledgePack {
35    const NAME: &'static str = "knowledge";
36    const NOTE_KINDS: &'static [&'static str] = &[];
37    const ENTITY_KINDS: &'static [&'static str] = &[];
38    const HANDLERS: &'static [HandlerDef] = &KNOWLEDGE_HANDLERS;
39    const REQUIRES: &'static [&'static str] = &["kg"];
40}
41
42impl KnowledgePack {
43    /// Create a new pack bound to the given runtime, initializing a shared ANN index.
44    pub fn new(runtime: KhiveRuntime) -> Self {
45        let brain_profile = runtime.config().brain_profile.clone();
46        Self {
47            runtime,
48            ann: vamana::new_shared(),
49            section_posteriors: Mutex::new(SectionPosteriorState::new()),
50            brain_profile,
51        }
52    }
53}
54
55struct KnowledgePackFactory;
56
57impl khive_runtime::PackFactory for KnowledgePackFactory {
58    fn name(&self) -> &'static str {
59        "knowledge"
60    }
61
62    fn requires(&self) -> &'static [&'static str] {
63        &["kg"]
64    }
65
66    fn create(&self, runtime: KhiveRuntime) -> Box<dyn khive_runtime::PackRuntime> {
67        Box::new(KnowledgePack::new(runtime))
68    }
69
70    fn create_resolver(
71        &self,
72        runtime: KhiveRuntime,
73    ) -> Option<Box<dyn khive_runtime::PackByIdResolver>> {
74        Some(Box::new(KnowledgePack::new(runtime)))
75    }
76}
77
78inventory::submit! { khive_runtime::PackRegistration(&KnowledgePackFactory) }
79
80#[async_trait]
81impl PackRuntime for KnowledgePack {
82    fn name(&self) -> &str {
83        <KnowledgePack as Pack>::NAME
84    }
85
86    fn note_kinds(&self) -> &'static [&'static str] {
87        <KnowledgePack as Pack>::NOTE_KINDS
88    }
89
90    fn entity_kinds(&self) -> &'static [&'static str] {
91        <KnowledgePack as Pack>::ENTITY_KINDS
92    }
93
94    fn handlers(&self) -> &'static [HandlerDef] {
95        &KNOWLEDGE_HANDLERS
96    }
97
98    fn requires(&self) -> &'static [&'static str] {
99        <KnowledgePack as Pack>::REQUIRES
100    }
101
102    async fn warm(&self) {
103        crate::knowledge::vamana::warm_known_snapshots(&self.runtime, &self.ann).await;
104        if !self.runtime.default_embedder_name().is_empty() {
105            let runtime = self.runtime.clone();
106            // ADR-103 Amendment 1 Part 2: the phase span brackets ONLY this
107            // configured-embedder branch's spawned embed call, not the
108            // unconditional `warm_known_snapshots` above -- an unconditional
109            // span would record a phase for an embedder warmup that never
110            // ran whenever no embedder is configured. Minted here (before
111            // spawn) rather than inside the task: `authorize` is cheap and
112            // synchronous, and minting outside keeps the same shape as
113            // KgPack::warm's unconditional case.
114            let token = match runtime.authorize(khive_runtime::Namespace::local()) {
115                Ok(token) => Some(token),
116                Err(e) => {
117                    tracing::warn!(
118                        error = %e,
119                        "knowledge_embedder_warm: failed to mint ADR-103 phase-span \
120                         attribution token; warmup proceeds without telemetry"
121                    );
122                    None
123                }
124            };
125            tokio::spawn(async move {
126                if let Some(token) = &token {
127                    khive_runtime::emit_phase_event(
128                        &runtime,
129                        token,
130                        "knowledge.embedder_warm",
131                        EventKind::PhaseStarted,
132                        PhaseStartedPayload {
133                            work_class: "warm".into(),
134                            phase: "knowledge_embedder_warm".into(),
135                            corpus_size: None,
136                        },
137                    )
138                    .await;
139                }
140                let phase_start = std::time::Instant::now();
141                let cpu_start = khive_runtime::process_resource_usage();
142                let result = runtime.embed("__khive_knowledge_warm__").await;
143                if let Some(token) = &token {
144                    let wall_us = phase_start.elapsed().as_micros() as i64;
145                    let cpu_us = khive_runtime::cpu_delta_us(
146                        cpu_start,
147                        khive_runtime::process_resource_usage(),
148                    );
149                    match &result {
150                        Err(e) if khive_runtime::is_benign_shutdown_cancellation(e) => {
151                            khive_runtime::emit_phase_event(
152                                &runtime,
153                                token,
154                                "knowledge.embedder_warm",
155                                EventKind::PhaseCancelled,
156                                PhaseCancelledPayload {
157                                    work_class: "warm".into(),
158                                    phase: "knowledge_embedder_warm".into(),
159                                    wall_us,
160                                    cpu_us,
161                                },
162                            )
163                            .await;
164                        }
165                        _ => {
166                            khive_runtime::emit_phase_event(
167                                &runtime,
168                                token,
169                                "knowledge.embedder_warm",
170                                EventKind::PhaseCompleted,
171                                PhaseCompletedPayload {
172                                    work_class: "warm".into(),
173                                    phase: "knowledge_embedder_warm".into(),
174                                    wall_us,
175                                    cpu_us,
176                                },
177                            )
178                            .await;
179                        }
180                    }
181                }
182                let _ = result;
183            });
184        }
185    }
186
187    async fn dispatch(
188        &self,
189        verb: &str,
190        params: Value,
191        registry: &VerbRegistry,
192        token: &NamespaceToken,
193    ) -> Result<Value, RuntimeError> {
194        match verb {
195            "knowledge.upsert_atoms" => {
196                KnowledgeHandlers::upsert_atoms(&self.runtime, token, params).await
197            }
198            "knowledge.upsert_domains" => {
199                KnowledgeHandlers::upsert_domains(&self.runtime, token, params).await
200            }
201            "knowledge.get" => KnowledgeHandlers::get(&self.runtime, token, params).await,
202            "knowledge.list" => KnowledgeHandlers::list(&self.runtime, token, params).await,
203            "knowledge.delete_atoms" => {
204                KnowledgeHandlers::delete_atoms(&self.runtime, token, params).await
205            }
206            "knowledge.stats" => KnowledgeHandlers::stats(&self.runtime, token, params).await,
207            "knowledge.index" => {
208                KnowledgeHandlers::index(&self.runtime, token, params, &self.ann, None).await
209            }
210            "knowledge.fold" => KnowledgeHandlers::fold(&self.runtime, token, params).await,
211            "knowledge.search" => {
212                KnowledgeHandlers::search(&self.runtime, token, params, &self.ann).await
213            }
214            "knowledge.suggest" => {
215                KnowledgeHandlers::suggest(&self.runtime, token, params, &self.ann).await
216            }
217            "knowledge.compose" => {
218                let type_weights = self.resolve_compose_type_weights(registry, token).await;
219                KnowledgeHandlers::compose(&self.runtime, token, params, &self.ann, type_weights)
220                    .await
221            }
222            "knowledge.edit" => {
223                KnowledgeHandlers::edit(&self.runtime, token, params, &self.ann).await
224            }
225            "knowledge.import" => {
226                KnowledgeHandlers::import(&self.runtime, token, params, &self.ann).await
227            }
228            "knowledge.challenge" => {
229                KnowledgeHandlers::challenge(&self.runtime, token, params).await
230            }
231            "knowledge.adjudicate" => {
232                KnowledgeHandlers::adjudicate(&self.runtime, token, params).await
233            }
234            "knowledge.learn" => self.handle_learn(token, params).await,
235            "knowledge.cite" => self.handle_cite(token, params).await,
236            "knowledge.topic" => self.handle_topic(token, params).await,
237            "knowledge.feedback" => self.handle_feedback(token, params, registry).await,
238            _ => Err(RuntimeError::InvalidInput(format!(
239                "knowledge pack does not handle verb {verb:?}"
240            ))),
241        }
242    }
243}
244
245#[async_trait]
246impl PackByIdResolver for KnowledgePack {
247    /// Resolve a live knowledge record by UUID.
248    ///
249    /// Queries `knowledge_domains` first (canonical record), then
250    /// `knowledge_atoms`. The domain mirror atom shares the domain's UUID
251    /// (crud.rs:279/300) so domains must win over the mirror.
252    async fn resolve_by_id(&self, id: Uuid) -> Result<Option<Resolved>, RuntimeError> {
253        let sql = self.runtime.sql();
254        let id_str = id.to_string();
255
256        let mut reader = sql
257            .reader()
258            .await
259            .map_err(|e| RuntimeError::Internal(format!("knowledge resolve_by_id reader: {e}")))?;
260
261        // 1. Check knowledge_domains first (canonical over the mirror atom).
262        let domain_row = reader
263            .query_row(SqlStatement {
264                sql: "SELECT id, namespace, slug, name, description, tags, members, \
265                      created_at, updated_at, deleted_at \
266                      FROM knowledge_domains WHERE id = ?1 AND deleted_at IS NULL LIMIT 1"
267                    .into(),
268                params: vec![SqlValue::Text(id_str.clone())],
269                label: Some("knowledge.resolve_by_id.domain".into()),
270            })
271            .await
272            .map_err(|e| {
273                RuntimeError::Internal(format!("knowledge resolve_by_id domain query: {e}"))
274            })?;
275
276        if let Some(row) = domain_row {
277            if let Some(domain) = domain_from_row(&row) {
278                return Ok(Some(Resolved::PackRecord {
279                    pack: "knowledge".into(),
280                    kind: "domain".into(),
281                    data: domain_to_json(&domain),
282                }));
283            }
284        }
285
286        // 2. Check knowledge_atoms.
287        let atom_row = reader
288            .query_row(SqlStatement {
289                sql: "SELECT id, namespace, slug, name, content, tags, properties, \
290                      status, finalized, source_uri, source_type, created_at, updated_at, deleted_at \
291                      FROM knowledge_atoms WHERE id = ?1 AND deleted_at IS NULL LIMIT 1"
292                    .into(),
293                params: vec![SqlValue::Text(id_str)],
294                label: Some("knowledge.resolve_by_id.atom".into()),
295            })
296            .await
297            .map_err(|e| {
298                RuntimeError::Internal(format!("knowledge resolve_by_id atom query: {e}"))
299            })?;
300
301        if let Some(row) = atom_row {
302            if let Some(atom) = atom_from_row(&row) {
303                return Ok(Some(Resolved::PackRecord {
304                    pack: "knowledge".into(),
305                    kind: "atom".into(),
306                    data: atom_to_json(&atom),
307                }));
308            }
309        }
310
311        Ok(None)
312    }
313
314    /// Resolve a knowledge record including already-soft-deleted records.
315    ///
316    /// Used by the hard-delete path to locate tombstoned records.
317    async fn resolve_by_id_including_deleted(
318        &self,
319        id: Uuid,
320    ) -> Result<Option<Resolved>, RuntimeError> {
321        let sql = self.runtime.sql();
322        let id_str = id.to_string();
323
324        let mut reader = sql.reader().await.map_err(|e| {
325            RuntimeError::Internal(format!(
326                "knowledge resolve_by_id_including_deleted reader: {e}"
327            ))
328        })?;
329
330        let domain_row = reader
331            .query_row(SqlStatement {
332                sql: "SELECT id, namespace, slug, name, description, tags, members, \
333                      created_at, updated_at, deleted_at \
334                      FROM knowledge_domains WHERE id = ?1 LIMIT 1"
335                    .into(),
336                params: vec![SqlValue::Text(id_str.clone())],
337                label: Some("knowledge.resolve_by_id_incl_deleted.domain".into()),
338            })
339            .await
340            .map_err(|e| {
341                RuntimeError::Internal(format!(
342                    "knowledge resolve_by_id_including_deleted domain query: {e}"
343                ))
344            })?;
345
346        if let Some(row) = domain_row {
347            if let Some(domain) = domain_from_row(&row) {
348                return Ok(Some(Resolved::PackRecord {
349                    pack: "knowledge".into(),
350                    kind: "domain".into(),
351                    data: domain_to_json(&domain),
352                }));
353            }
354        }
355
356        let atom_row = reader
357            .query_row(SqlStatement {
358                sql: "SELECT id, namespace, slug, name, content, tags, properties, \
359                      status, finalized, source_uri, source_type, created_at, updated_at, deleted_at \
360                      FROM knowledge_atoms WHERE id = ?1 LIMIT 1"
361                    .into(),
362                params: vec![SqlValue::Text(id_str)],
363                label: Some("knowledge.resolve_by_id_incl_deleted.atom".into()),
364            })
365            .await
366            .map_err(|e| {
367                RuntimeError::Internal(format!(
368                    "knowledge resolve_by_id_including_deleted atom query: {e}"
369                ))
370            })?;
371
372        if let Some(row) = atom_row {
373            if let Some(atom) = atom_from_row(&row) {
374                return Ok(Some(Resolved::PackRecord {
375                    pack: "knowledge".into(),
376                    kind: "atom".into(),
377                    data: atom_to_json(&atom),
378                }));
379            }
380        }
381
382        Ok(None)
383    }
384
385    /// Delete a knowledge domain or atom by UUID.
386    ///
387    /// Soft-delete by default (`hard=false`): sets `deleted_at = now()`.
388    /// For domains, also tombstones the mirror atom in `knowledge_atoms`.
389    /// Hard-delete (`hard=true`): permanently removes rows from the table(s).
390    async fn delete_by_id(&self, id: Uuid, hard: bool) -> Result<serde_json::Value, RuntimeError> {
391        let sql = self.runtime.sql();
392        let id_str = id.to_string();
393
394        // First determine the kind (including deleted rows so hard-delete can find tombstones).
395        let kind = match self.resolve_by_id_including_deleted(id).await? {
396            Some(Resolved::PackRecord { kind, .. }) => kind,
397            _ => {
398                return Err(RuntimeError::NotFound(format!(
399                    "knowledge record not found: {id}"
400                )));
401            }
402        };
403
404        let now_us = std::time::SystemTime::now()
405            .duration_since(std::time::UNIX_EPOCH)
406            .unwrap_or_default()
407            .as_micros() as i64;
408
409        let mut writer = sql
410            .writer()
411            .await
412            .map_err(|e| RuntimeError::Internal(format!("knowledge delete_by_id writer: {e}")))?;
413
414        match kind.as_str() {
415            "domain" => {
416                if hard {
417                    // Hard-delete: sections → mirror atom → domain (all in one transaction).
418                    // knowledge_sections has a FK to knowledge_atoms(id) without ON DELETE
419                    // CASCADE, so sections must go before the atom row.
420                    writer
421                        .execute_batch(vec![
422                            SqlStatement {
423                                sql: "DELETE FROM knowledge_sections WHERE atom_id = ?1".into(),
424                                params: vec![SqlValue::Text(id_str.clone())],
425                                label: Some(
426                                    "knowledge.delete_by_id.domain_mirror_sections.hard".into(),
427                                ),
428                            },
429                            SqlStatement {
430                                sql: "DELETE FROM knowledge_atoms WHERE id = ?1".into(),
431                                params: vec![SqlValue::Text(id_str.clone())],
432                                label: Some("knowledge.delete_by_id.domain_mirror.hard".into()),
433                            },
434                            SqlStatement {
435                                sql: "DELETE FROM knowledge_domains WHERE id = ?1".into(),
436                                params: vec![SqlValue::Text(id_str.clone())],
437                                label: Some("knowledge.delete_by_id.domain.hard".into()),
438                            },
439                        ])
440                        .await
441                        .map_err(|e| {
442                            RuntimeError::Internal(format!(
443                                "knowledge delete_by_id domain hard: {e}"
444                            ))
445                        })?;
446                } else {
447                    writer
448                        .execute(SqlStatement {
449                            sql: "UPDATE knowledge_domains SET deleted_at = ?1 \
450                                  WHERE id = ?2 AND deleted_at IS NULL"
451                                .into(),
452                            params: vec![SqlValue::Integer(now_us), SqlValue::Text(id_str.clone())],
453                            label: Some("knowledge.delete_by_id.domain.soft".into()),
454                        })
455                        .await
456                        .map_err(|e| {
457                            RuntimeError::Internal(format!(
458                                "knowledge delete_by_id domain soft: {e}"
459                            ))
460                        })?;
461                    // Tombstone the mirror atom so FTS no longer surfaces it.
462                    writer
463                        .execute(SqlStatement {
464                            sql: "UPDATE knowledge_atoms SET deleted_at = ?1 \
465                                  WHERE id = ?2 AND deleted_at IS NULL"
466                                .into(),
467                            params: vec![SqlValue::Integer(now_us), SqlValue::Text(id_str.clone())],
468                            label: Some("knowledge.delete_by_id.domain_mirror.soft".into()),
469                        })
470                        .await
471                        .map_err(|e| {
472                            RuntimeError::Internal(format!(
473                                "knowledge delete_by_id domain mirror soft: {e}"
474                            ))
475                        })?;
476                }
477            }
478            "atom" => {
479                if hard {
480                    // Hard-delete: sections first (FK constraint), then the atom row.
481                    writer
482                        .execute_batch(vec![
483                            SqlStatement {
484                                sql: "DELETE FROM knowledge_sections WHERE atom_id = ?1".into(),
485                                params: vec![SqlValue::Text(id_str.clone())],
486                                label: Some("knowledge.delete_by_id.atom_sections.hard".into()),
487                            },
488                            SqlStatement {
489                                sql: "DELETE FROM knowledge_atoms WHERE id = ?1".into(),
490                                params: vec![SqlValue::Text(id_str.clone())],
491                                label: Some("knowledge.delete_by_id.atom.hard".into()),
492                            },
493                        ])
494                        .await
495                        .map_err(|e| {
496                            RuntimeError::Internal(format!("knowledge delete_by_id atom hard: {e}"))
497                        })?;
498                } else {
499                    writer
500                        .execute(SqlStatement {
501                            sql: "UPDATE knowledge_atoms SET deleted_at = ?1 \
502                                  WHERE id = ?2 AND deleted_at IS NULL"
503                                .into(),
504                            params: vec![SqlValue::Integer(now_us), SqlValue::Text(id_str.clone())],
505                            label: Some("knowledge.delete_by_id.atom.soft".into()),
506                        })
507                        .await
508                        .map_err(|e| {
509                            RuntimeError::Internal(format!("knowledge delete_by_id atom soft: {e}"))
510                        })?;
511                }
512            }
513            other => {
514                return Err(RuntimeError::Internal(format!(
515                    "knowledge delete_by_id: unexpected kind {other:?}"
516                )));
517            }
518        }
519
520        Ok(serde_json::json!({
521            "deleted": true,
522            "id": id_str,
523            "kind": kind,
524            "hard": hard,
525        }))
526    }
527}
528
529#[cfg(test)]
530mod tests {
531    use khive_runtime::pack::PackRuntime;
532    use khive_runtime::{KhiveRuntime, Namespace};
533
534    use super::KnowledgePack;
535
536    // ADR-103 Amendment 1 Part 2: knowledge_embedder_warm phase-span events.
537    //
538    // `KhiveRuntime::memory()` builds on `RuntimeConfig::no_embeddings()`
539    // (runtime.rs), so `default_embedder_name()` is empty and the
540    // configured-embedder branch in `KnowledgePack::warm()` never runs. This
541    // pins that gating: the unconditional `vamana::warm_known_snapshots` call
542    // above the branch must never itself emit a phase event, and the
543    // configured-only branch must not fire when unconfigured.
544    #[tokio::test]
545    async fn knowledge_warm_emits_no_phase_events_when_no_embedder_configured() {
546        let rt = KhiveRuntime::memory().expect("in-memory runtime");
547        let token = rt.authorize(Namespace::local()).expect("authorize local");
548        let event_store = rt.events(&token).expect("event store must be available");
549
550        let pack = KnowledgePack::new(rt.clone());
551        pack.warm().await;
552
553        let page = event_store
554            .query_events(
555                khive_storage::EventFilter::default(),
556                khive_storage::PageRequest {
557                    limit: 10,
558                    offset: 0,
559                },
560            )
561            .await
562            .expect("query events");
563
564        assert!(
565            page.items.is_empty(),
566            "no embedder configured -> KnowledgePack::warm() must emit zero phase events \
567             (configured-only branch must not fire, and warm_known_snapshots must not \
568             independently emit): {:?}",
569            page.items
570        );
571    }
572}
khive_pack_knowledge/pack.rs

khive_pack_knowledge/
pack.rs
