convergio_knowledge/
ext.rs

1//! Extension trait implementation for convergio-knowledge.
2
3use std::path::PathBuf;
4use std::sync::{Arc, OnceLock};
5
6use convergio_db::pool::ConnPool;
7use convergio_types::events::{DomainEvent, EventFilter, EventKind};
8use convergio_types::extension::{
9    AppContext, Extension, Health, McpToolDef, Metric, Migration, ScheduledTask,
10};
11use convergio_types::manifest::{Capability, Manifest, ModuleKind};
12
13use crate::routes::{knowledge_routes, KnowledgeState};
14use crate::store::LanceVectorStore;
15
16pub struct KnowledgeExtension {
17    pool: ConnPool,
18    lance_path: PathBuf,
19    store: OnceLock<Option<Arc<LanceVectorStore>>>,
20}
21
22impl KnowledgeExtension {
23    pub fn new(pool: ConnPool) -> Self {
24        let lance_path = dirs::home_dir()
25            .unwrap_or_else(|| PathBuf::from("."))
26            .join(".convergio/data/knowledge.lance");
27        Self {
28            pool,
29            lance_path,
30            store: OnceLock::new(),
31        }
32    }
33
34    fn get_store(&self) -> Option<Arc<LanceVectorStore>> {
35        self.store
36            .get_or_init(|| {
37                let path = self.lance_path.clone();
38                // Spawn a separate thread to init LanceDB — avoids tokio runtime nesting issues
39                let result = std::thread::spawn(move || {
40                    let rt = tokio::runtime::Builder::new_current_thread()
41                        .enable_all()
42                        .build()
43                        .ok()?;
44                    rt.block_on(LanceVectorStore::open(&path)).ok()
45                })
46                .join()
47                .ok()
48                .flatten();
49                match result {
50                    Some(s) => Some(Arc::new(s)),
51                    None => {
52                        tracing::warn!("lance store unavailable — knowledge search disabled");
53                        None
54                    }
55                }
56            })
57            .clone()
58    }
59
60    fn state(&self) -> Arc<KnowledgeState> {
61        Arc::new(KnowledgeState {
62            pool: self.pool.clone(),
63            store: self.get_store(),
64        })
65    }
66}
67
68impl Extension for KnowledgeExtension {
69    fn manifest(&self) -> Manifest {
70        Manifest {
71            id: "convergio-knowledge".to_string(),
72            description: "Vector knowledge store — semantic memory for agents via embeddings"
73                .to_string(),
74            version: env!("CARGO_PKG_VERSION").to_string(),
75            kind: ModuleKind::Platform,
76            provides: vec![
77                Capability {
78                    name: "knowledge-vectors".to_string(),
79                    version: "1.0.0".to_string(),
80                    description: "Embedding-based semantic search over agent knowledge".to_string(),
81                },
82                Capability {
83                    name: "knowledge-api".to_string(),
84                    version: "1.0.0".to_string(),
85                    description: "REST API for knowledge search and write".to_string(),
86                },
87            ],
88            requires: vec![],
89            agent_tools: vec![],
90            required_roles: vec!["orchestrator".into(), "all".into()],
91        }
92    }
93
94    fn migrations(&self) -> Vec<Migration> {
95        vec![]
96    }
97
98    fn routes(&self, _ctx: &AppContext) -> Option<axum::Router> {
99        Some(knowledge_routes(self.state()))
100    }
101
102    fn health(&self) -> Health {
103        let Some(store) = self.get_store() else {
104            return Health::Ok;
105        };
106        let ok = std::thread::spawn(move || {
107            let rt = tokio::runtime::Builder::new_current_thread()
108                .enable_all()
109                .build()
110                .ok()?;
111            rt.block_on(store.stats()).ok()
112        })
113        .join()
114        .ok()
115        .flatten();
116        if ok.is_some() {
117            Health::Ok
118        } else {
119            Health::Degraded {
120                reason: "lance stats failed".into(),
121            }
122        }
123    }
124
125    fn metrics(&self) -> Vec<Metric> {
126        let Some(store) = self.get_store() else {
127            return vec![];
128        };
129        let count = std::thread::spawn(move || {
130            let rt = tokio::runtime::Builder::new_current_thread()
131                .enable_all()
132                .build()
133                .ok()?;
134            rt.block_on(store.stats())
135                .ok()
136                .map(|s| s.total_entries as f64)
137        })
138        .join()
139        .ok()
140        .flatten()
141        .unwrap_or(0.0);
142        vec![Metric {
143            name: "knowledge_entries_total".to_string(),
144            value: count,
145            labels: vec![],
146        }]
147    }
148
149    fn subscriptions(&self) -> Vec<EventFilter> {
150        vec![EventFilter {
151            kind_prefix: Some("Task".to_string()),
152            org: None,
153            actor: None,
154        }]
155    }
156
157    fn on_event(&self, event: &DomainEvent) {
158        if let EventKind::TaskCompleted { task_id } = &event.kind {
159            if let Some(store) = self.get_store() {
160                crate::hooks::on_task_completed(&self.pool, store, *task_id)
161            };
162        }
163    }
164
165    fn scheduled_tasks(&self) -> Vec<ScheduledTask> {
166        vec![
167            ScheduledTask {
168                name: "knowledge-commit-sync",
169                cron: "*/15 * * * *",
170            },
171            ScheduledTask {
172                name: "knowledge-auto-prune",
173                cron: "0 3 * * *", // daily at 3 AM
174            },
175        ]
176    }
177
178    fn on_scheduled_task(&self, task_name: &str) {
179        if task_name == "knowledge-commit-sync" {
180            if let Some(store) = self.get_store() {
181                crate::hooks::sync_recent_commits(store)
182            };
183        }
184        if task_name == "knowledge-auto-prune" {
185            if let Some(store) = self.get_store() {
186                let store = store.clone();
187                tokio::spawn(async move {
188                    let req = crate::pruning::PruneRequest {
189                        max_age_days: None,
190                        source_type: None,
191                        dry_run: false,
192                    };
193                    let report = crate::pruning::prune(&store, &req).await;
194                    tracing::info!(
195                        stale = report.stale_removed,
196                        dedup = report.duplicates_removed,
197                        source_dedup = report.source_dedup_removed,
198                        total = report.total_removed,
199                        "auto-prune completed"
200                    );
201                });
202            };
203        }
204    }
205
206    fn mcp_tools(&self) -> Vec<McpToolDef> {
207        crate::mcp_defs::knowledge_tools()
208    }
209}
convergio_knowledge/ext.rs

convergio_knowledge/
ext.rs
