convergio-knowledge 0.1.1

Vector knowledge store — semantic memory for agents via embeddings
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//! Pruning helpers for the vector store — list_all_metadata + delete_batch.
//!
//! Extracted from store.rs to stay under the 300-line limit.

use std::collections::HashSet;

use arrow_array::StringArray;
use futures::StreamExt;
use lancedb::query::ExecutableQuery;

use crate::store::LanceVectorStore;
use crate::types::KnowledgeEntry;

impl LanceVectorStore {
    /// Collect all source_ids in the store (used by seed dedup).
    pub async fn all_source_ids(&self) -> Result<HashSet<String>, String> {
        let conn = self.conn.lock().await;
        let tbl = conn
            .open_table(&self.table_name)
            .execute()
            .await
            .map_err(|e| format!("open: {e}"))?;
        if tbl.count_rows(None).await.unwrap_or(0) == 0 {
            return Ok(HashSet::new());
        }
        let mut stream = tbl
            .query()
            .execute()
            .await
            .map_err(|e| format!("query source_ids: {e}"))?;
        let mut ids = HashSet::new();
        while let Some(Ok(batch)) = stream.next().await {
            if let Some(col) = batch
                .column_by_name("source_id")
                .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            {
                for i in 0..batch.num_rows() {
                    ids.insert(col.value(i).to_string());
                }
            }
        }
        Ok(ids)
    }

    /// List all entries (lightweight metadata only — no vectors).
    /// Used by the pruning system to scan for stale/duplicate entries.
    pub async fn list_all_metadata(&self) -> Result<Vec<KnowledgeEntry>, String> {
        let conn = self.conn.lock().await;
        let tbl = conn
            .open_table(&self.table_name)
            .execute()
            .await
            .map_err(|e| format!("open: {e}"))?;
        if tbl.count_rows(None).await.unwrap_or(0) == 0 {
            return Ok(vec![]);
        }
        let mut stream = tbl
            .query()
            .execute()
            .await
            .map_err(|e| format!("query all: {e}"))?;

        let mut out = Vec::new();
        while let Some(Ok(batch)) = stream.next().await {
            let col = |name: &str| -> Option<&StringArray> {
                batch.column_by_name(name)?.as_any().downcast_ref()
            };
            let (Some(ids), Some(contents), Some(src_types), Some(src_ids)) = (
                col("id"),
                col("content"),
                col("source_type"),
                col("source_id"),
            ) else {
                continue;
            };
            let org_ids = col("org_id");
            let agent_ids = col("agent_id");
            let project_ids = col("project_id");
            let visibilities = col("visibility");
            let created = col("created_at");

            for i in 0..batch.num_rows() {
                out.push(KnowledgeEntry {
                    id: ids.value(i).to_string(),
                    content: contents.value(i).to_string(),
                    source_type: src_types.value(i).to_string(),
                    source_id: src_ids.value(i).to_string(),
                    org_id: org_ids.map(|a| a.value(i).to_string()),
                    agent_id: agent_ids.map(|a| a.value(i).to_string()),
                    project_id: project_ids.map(|a| a.value(i).to_string()),
                    visibility: visibilities
                        .map(|a| a.value(i).to_string())
                        .unwrap_or_else(|| "org".to_string()),
                    created_at: created.map(|a| a.value(i).to_string()).unwrap_or_default(),
                });
            }
        }
        Ok(out)
    }

    /// Delete multiple entries by ID in a single filter expression.
    pub async fn delete_batch(&self, ids: &[String]) -> Result<usize, String> {
        if ids.is_empty() {
            return Ok(0);
        }
        let conn = self.conn.lock().await;
        let table = conn
            .open_table(&self.table_name)
            .execute()
            .await
            .map_err(|e| format!("open: {e}"))?;
        let escaped: Vec<String> = ids
            .iter()
            .map(|id| format!("'{}'", id.replace('\'', "''")))
            .collect();
        let filter = format!("id IN ({})", escaped.join(", "));
        table
            .delete(&filter)
            .await
            .map_err(|e| format!("delete_batch: {e}"))?;
        Ok(ids.len())
    }
}