talos_evolution/
store.rs

1//! KnowledgeStore — SQLite persistence for observations and patterns.
2
3use chrono::{DateTime, Utc};
4use rusqlite::{Connection, params};
5
6use crate::EvolutionResult as Result;
7use crate::{Observation, Pattern, SignalType};
8
9/// SQLite-backed store for evolution data.
10pub struct KnowledgeStore {
11    conn: Connection,
12}
13
14impl KnowledgeStore {
15    /// Open or create a knowledge store at the given path.
16    pub fn open(path: &str) -> Result<Self> {
17        let conn = Connection::open(path)?;
18        let store = Self { conn };
19        store.migrate()?;
20        Ok(store)
21    }
22
23    /// Open an in-memory knowledge store for testing.
24    pub fn open_memory() -> Result<Self> {
25        let conn = Connection::open_in_memory()?;
26        let store = Self { conn };
27        store.migrate()?;
28        Ok(store)
29    }
30
31    fn migrate(&self) -> Result<()> {
32        self.conn.execute_batch(
33            "CREATE TABLE IF NOT EXISTS observations (
34                id TEXT PRIMARY KEY,
35                signal_type TEXT NOT NULL,
36                intensity REAL NOT NULL,
37                context TEXT NOT NULL,
38                timestamp TEXT NOT NULL,
39                session_id TEXT,
40                turn_number INTEGER
41            );
42
43            CREATE TABLE IF NOT EXISTS patterns (
44                id TEXT PRIMARY KEY,
45                description TEXT NOT NULL,
46                instruction TEXT NOT NULL,
47                confidence REAL NOT NULL,
48                evidence_count INTEGER NOT NULL,
49                first_observed TEXT NOT NULL,
50                last_updated TEXT NOT NULL,
51                category TEXT NOT NULL,
52                active INTEGER NOT NULL DEFAULT 1,
53                content_hash TEXT NOT NULL DEFAULT ''
54            );
55
56            CREATE TABLE IF NOT EXISTS conflicts (
57                id TEXT PRIMARY KEY,
58                pattern_a_id TEXT NOT NULL,
59                pattern_b_id TEXT NOT NULL,
60                description TEXT NOT NULL,
61                detected_at TEXT NOT NULL,
62                resolved INTEGER NOT NULL DEFAULT 0,
63                winner_id TEXT
64            );
65
66            -- I021-S1: MenteDB-aligned tables
67            CREATE TABLE IF NOT EXISTS signals (
68                id TEXT PRIMARY KEY,
69                kind TEXT NOT NULL,
70                intensity REAL NOT NULL,
71                context TEXT NOT NULL,
72                tool_name TEXT,
73                turn_observation_id TEXT REFERENCES turn_observations(id),
74                timestamp TEXT NOT NULL
75            );
76
77            CREATE TABLE IF NOT EXISTS turn_observations (
78                id TEXT PRIMARY KEY,
79                session_id TEXT NOT NULL,
80                turn_number INTEGER NOT NULL,
81                outcome TEXT NOT NULL,
82                duration_ms INTEGER NOT NULL,
83                timestamp TEXT NOT NULL
84            );
85
86            CREATE TABLE IF NOT EXISTS schema_version (
87                version INTEGER NOT NULL
88            );",
89        )?;
90
91        // I021-S4: Detect v1 schema (missing pattern.key) and hard-reset if needed.
92        let has_key_column: i64 = self
93            .conn
94            .prepare("SELECT COUNT(*) FROM pragma_table_info('patterns') WHERE name = 'key'")?
95            .query_row([], |row| row.get(0))?;
96
97        if has_key_column == 0 {
98            let obs_count: i64 = self
99                .conn
100                .query_row("SELECT COUNT(*) FROM observations", [], |row| row.get(0))
101                .unwrap_or(0);
102            let pat_count: i64 = self
103                .conn
104                .query_row("SELECT COUNT(*) FROM patterns", [], |row| row.get(0))
105                .unwrap_or(0);
106
107            if obs_count > 0 || pat_count > 0 {
108                let _ = self.conn.execute("DELETE FROM observations", []);
109                let _ = self.conn.execute("DELETE FROM patterns", []);
110                let _ = self.conn.execute("DELETE FROM conflicts", []);
111                let _ = self.conn.execute("DELETE FROM signals", []);
112                let _ = self.conn.execute("DELETE FROM turn_observations", []);
113                tracing::warn!(
114                    observations = obs_count,
115                    patterns = pat_count,
116                    "knowledge.db schema migration: hard-reset, removed observations and patterns"
117                );
118            }
119        }
120
121        // Add content_hash column to existing databases (SQLite ALTER doesn't support IF NOT EXISTS).
122        // We catch the "duplicate column" error to make this idempotent.
123        let has_column = self
124            .conn
125            .prepare(
126                "SELECT COUNT(*) FROM pragma_table_info('patterns') WHERE name = 'content_hash'",
127            )?
128            .query_row([], |row| row.get::<_, i64>(0))?;
129        if has_column == 0 {
130            let _ = self.conn.execute(
131                "ALTER TABLE patterns ADD COLUMN content_hash TEXT NOT NULL DEFAULT ''",
132                [],
133            );
134        }
135
136        // I021-S3: Add MenteDB-aligned columns to patterns table.
137        for (col, default) in [
138            ("key", "''"),
139            ("value", "'null'"),
140            ("contradicting_count", "0"),
141            ("last_reinforced", "''"),
142            ("source_sessions", "'[]'"),
143        ] {
144            let has_col: i64 = self
145                .conn
146                .prepare(&format!(
147                    "SELECT COUNT(*) FROM pragma_table_info('patterns') WHERE name = '{col}'"
148                ))?
149                .query_row([], |row| row.get(0))?;
150            if has_col == 0 {
151                let _ = self.conn.execute(
152                    &format!(
153                        "ALTER TABLE patterns ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}"
154                    ),
155                    [],
156                );
157            }
158        }
159
160        // Initialize schema_version if empty (new database).
161        let version_count: i64 =
162            self.conn
163                .query_row("SELECT COUNT(*) FROM schema_version", [], |row| row.get(0))?;
164        if version_count == 0 {
165            let _ = self
166                .conn
167                .execute("INSERT INTO schema_version (version) VALUES (2)", []);
168        }
169
170        Ok(())
171    }
172
173    /// Insert an observation.
174    pub fn insert_observation(&self, obs: &Observation) -> Result<()> {
175        self.conn.execute(
176            "INSERT INTO observations (id, signal_type, intensity, context, timestamp, session_id, turn_number)
177             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
178            params![
179                obs.id,
180                format!("{:?}", obs.signal_type),
181                obs.intensity,
182                obs.context,
183                obs.timestamp.to_rfc3339(),
184                obs.session_id,
185                obs.turn_number,
186            ],
187        )?;
188        Ok(())
189    }
190
191    /// Get all observations.
192    pub fn get_observations(&self) -> Result<Vec<Observation>> {
193        let mut stmt = self.conn.prepare(
194            "SELECT id, signal_type, intensity, context, timestamp, session_id, turn_number
195             FROM observations ORDER BY timestamp DESC",
196        )?;
197
198        let observations = stmt
199            .query_map([], |row| {
200                let signal_type_str: String = row.get(1)?;
201                let signal_type = match signal_type_str.as_str() {
202                    "Correction" => SignalType::Correction,
203                    "Error" => SignalType::Error,
204                    "Satisfaction" => SignalType::Satisfaction,
205                    "Inefficiency" => SignalType::Inefficiency,
206                    _ => SignalType::Correction,
207                };
208
209                let timestamp_str: String = row.get(4)?;
210                let timestamp = DateTime::parse_from_rfc3339(&timestamp_str)
211                    .map(|dt| dt.with_timezone(&Utc))
212                    .unwrap_or_else(|_| Utc::now());
213
214                Ok(Observation {
215                    id: row.get(0)?,
216                    signal_type,
217                    intensity: row.get(2)?,
218                    context: row.get(3)?,
219                    timestamp,
220                    session_id: row.get(5)?,
221                    turn_number: row.get(6)?,
222                })
223            })?
224            .collect::<std::result::Result<Vec<_>, _>>()?;
225
226        Ok(observations)
227    }
228
229    /// Insert a pattern.
230    pub fn insert_pattern(&self, pattern: &Pattern) -> Result<()> {
231        let value_json = serde_json::to_string(&pattern.value).unwrap_or_else(|_| "null".into());
232        let sessions_json =
233            serde_json::to_string(&pattern.source_sessions).unwrap_or_else(|_| "[]".into());
234        self.conn.execute(
235            "INSERT INTO patterns (id, description, instruction, confidence, evidence_count, first_observed, last_updated, category, active, content_hash, key, value, contradicting_count, last_reinforced, source_sessions)
236             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
237            params![
238                pattern.id,
239                pattern.description,
240                pattern.instruction,
241                pattern.confidence,
242                pattern.evidence_count,
243                pattern.first_observed.to_rfc3339(),
244                pattern.last_updated.to_rfc3339(),
245                pattern.category,
246                pattern.active as i32,
247                pattern.content_hash,
248                pattern.key,
249                value_json,
250                pattern.contradicting_count as i32,
251                pattern.last_reinforced.to_rfc3339(),
252                sessions_json,
253            ],
254        )?;
255        Ok(())
256    }
257
258    /// Get active patterns with confidence above threshold.
259    pub fn get_active_patterns(&self, min_confidence: f64) -> Result<Vec<Pattern>> {
260        let mut stmt = self.conn.prepare(
261            "SELECT id, description, instruction, confidence, evidence_count, first_observed, last_updated, category, active, content_hash, key, value, contradicting_count, last_reinforced, source_sessions
262             FROM patterns WHERE active = 1 AND confidence >= ?1 ORDER BY confidence DESC",
263        )?;
264
265        let patterns = stmt
266            .query_map(params![min_confidence], |row| {
267                let first_observed_str: String = row.get(5)?;
268                let first_observed = DateTime::parse_from_rfc3339(&first_observed_str)
269                    .map(|dt| dt.with_timezone(&Utc))
270                    .unwrap_or_else(|_| Utc::now());
271
272                let last_updated_str: String = row.get(6)?;
273                let last_updated = DateTime::parse_from_rfc3339(&last_updated_str)
274                    .map(|dt| dt.with_timezone(&Utc))
275                    .unwrap_or_else(|_| Utc::now());
276
277                let key: String = row.get(10)?;
278                let value_str: String = row.get(11)?;
279                let value: serde_json::Value =
280                    serde_json::from_str(&value_str).unwrap_or(serde_json::Value::Null);
281                let contradicting_raw: String = row.get(12)?;
282                let contradicting_count: u32 = contradicting_raw.parse().unwrap_or(0);
283
284                let last_reinforced_str: String = row.get(13)?;
285                let last_reinforced = if last_reinforced_str.is_empty() {
286                    first_observed
287                } else {
288                    DateTime::parse_from_rfc3339(&last_reinforced_str)
289                        .map(|dt| dt.with_timezone(&Utc))
290                        .unwrap_or(first_observed)
291                };
292
293                let sessions_str: String = row.get(14)?;
294                let source_sessions: Vec<uuid::Uuid> =
295                    serde_json::from_str(&sessions_str).unwrap_or_default();
296
297                Ok(Pattern {
298                    id: row.get(0)?,
299                    description: row.get(1)?,
300                    instruction: row.get(2)?,
301                    confidence: row.get(3)?,
302                    evidence_count: row.get(4)?,
303                    first_observed,
304                    last_updated,
305                    category: row.get(7)?,
306                    active: row.get::<_, i32>(8)? == 1,
307                    content_hash: row.get(9)?,
308                    key,
309                    value,
310                    contradicting_count,
311                    last_reinforced,
312                    source_sessions,
313                })
314            })?
315            .collect::<std::result::Result<Vec<_>, _>>()?;
316
317        Ok(patterns)
318    }
319
320    /// Update pattern confidence and evidence count.
321    pub fn update_pattern(&self, pattern: &Pattern) -> Result<()> {
322        self.conn.execute(
323            "UPDATE patterns SET confidence = ?1, evidence_count = ?2, last_updated = ?3
324             WHERE id = ?4",
325            params![
326                pattern.confidence,
327                pattern.evidence_count,
328                pattern.last_updated.to_rfc3339(),
329                pattern.id,
330            ],
331        )?;
332        Ok(())
333    }
334
335    /// Deactivate a pattern.
336    pub fn deactivate_pattern(&self, pattern_id: &str) -> Result<()> {
337        self.conn.execute(
338            "UPDATE patterns SET active = 0 WHERE id = ?1",
339            params![pattern_id],
340        )?;
341        Ok(())
342    }
343
344    /// Deactivate patterns whose instruction exceeds `max_bytes`. Returns count.
345    pub fn delete_oversized_patterns(&self, max_bytes: usize) -> Result<usize> {
346        let changes = self.conn.execute(
347            "UPDATE patterns SET active = 0 WHERE length(instruction) > ?1",
348            params![max_bytes as i64],
349        )?;
350        Ok(changes)
351    }
352
353    /// Get all patterns (including inactive).
354    pub fn get_all_patterns(&self) -> Result<Vec<Pattern>> {
355        let mut stmt = self.conn.prepare(
356            "SELECT id, description, instruction, confidence, evidence_count, first_observed, last_updated, category, active, content_hash, key, value, contradicting_count, last_reinforced, source_sessions
357             FROM patterns ORDER BY confidence DESC",
358        )?;
359
360        let patterns = stmt
361            .query_map([], |row| {
362                let first_observed_str: String = row.get(5)?;
363                let first_observed = DateTime::parse_from_rfc3339(&first_observed_str)
364                    .map(|dt| dt.with_timezone(&Utc))
365                    .unwrap_or_else(|_| Utc::now());
366
367                let last_updated_str: String = row.get(6)?;
368                let last_updated = DateTime::parse_from_rfc3339(&last_updated_str)
369                    .map(|dt| dt.with_timezone(&Utc))
370                    .unwrap_or_else(|_| Utc::now());
371
372                let key: String = row.get(10)?;
373                let value_str: String = row.get(11)?;
374                let value: serde_json::Value =
375                    serde_json::from_str(&value_str).unwrap_or(serde_json::Value::Null);
376                let contradicting_raw: String = row.get(12)?;
377                let contradicting_count: u32 = contradicting_raw.parse().unwrap_or(0);
378
379                let last_reinforced_str: String = row.get(13)?;
380                let last_reinforced = if last_reinforced_str.is_empty() {
381                    first_observed
382                } else {
383                    DateTime::parse_from_rfc3339(&last_reinforced_str)
384                        .map(|dt| dt.with_timezone(&Utc))
385                        .unwrap_or(first_observed)
386                };
387
388                let sessions_str: String = row.get(14)?;
389                let source_sessions: Vec<uuid::Uuid> =
390                    serde_json::from_str(&sessions_str).unwrap_or_default();
391
392                Ok(Pattern {
393                    id: row.get(0)?,
394                    description: row.get(1)?,
395                    instruction: row.get(2)?,
396                    confidence: row.get(3)?,
397                    evidence_count: row.get(4)?,
398                    first_observed,
399                    last_updated,
400                    category: row.get(7)?,
401                    active: row.get::<_, i32>(8)? == 1,
402                    content_hash: row.get(9)?,
403                    key,
404                    value,
405                    contradicting_count,
406                    last_reinforced,
407                    source_sessions,
408                })
409            })?
410            .collect::<std::result::Result<Vec<_>, _>>()?;
411
412        Ok(patterns)
413    }
414}
415
416#[cfg(test)]
417#[allow(warnings)]
418mod tests {
419    use super::*;
420
421    #[test]
422    fn test_store_operations() {
423        let store = KnowledgeStore::open_memory().expect("operation should succeed");
424
425        let obs = Observation::new(
426            SignalType::Correction,
427            0.8,
428            "Use functional style".to_string(),
429            Some("session-1".to_string()),
430            Some(5),
431        );
432        store
433            .insert_observation(&obs)
434            .expect("operation should succeed");
435
436        let observations = store.get_observations().expect("operation should succeed");
437        assert_eq!(observations.len(), 1);
438        assert_eq!(observations[0].signal_type, SignalType::Correction);
439
440        let mut pattern = Pattern::new(
441            "Prefer functional style".to_string(),
442            "Use functional programming patterns".to_string(),
443            "preference".to_string(),
444        );
445        pattern.confidence = 0.8;
446        pattern.evidence_count = 3;
447        store
448            .insert_pattern(&pattern)
449            .expect("operation should succeed");
450
451        let patterns = store
452            .get_active_patterns(0.7)
453            .expect("operation should succeed");
454        assert_eq!(patterns.len(), 1);
455        assert_eq!(patterns[0].confidence, 0.8);
456    }
457
458    #[test]
459    fn test_update_pattern() {
460        let store = KnowledgeStore::open_memory().expect("operation should succeed");
461
462        let mut pattern = Pattern::new(
463            "Test pattern".to_string(),
464            "Test instruction".to_string(),
465            "test".to_string(),
466        );
467        pattern.confidence = 0.5;
468        store
469            .insert_pattern(&pattern)
470            .expect("operation should succeed");
471
472        pattern.confidence = 0.9;
473        pattern.evidence_count = 10;
474        store
475            .update_pattern(&pattern)
476            .expect("operation should succeed");
477
478        let patterns = store.get_all_patterns().expect("operation should succeed");
479        assert_eq!(patterns[0].confidence, 0.9);
480        assert_eq!(patterns[0].evidence_count, 10);
481    }
482
483    #[test]
484    fn test_delete_oversized_patterns_deactivates_but_keeps_row() {
485        let store = KnowledgeStore::open_memory().expect("operation should succeed");
486
487        let mut pattern = Pattern::new(
488            "Big pattern".to_string(),
489            "x".repeat(10_000),
490            "test".to_string(),
491        );
492        pattern.confidence = 0.9;
493        store
494            .insert_pattern(&pattern)
495            .expect("operation should succeed");
496
497        let count = store
498            .delete_oversized_patterns(4096)
499            .expect("operation should succeed");
500        assert_eq!(count, 1);
501
502        let all = store.get_all_patterns().expect("operation should succeed");
503        assert_eq!(all.len(), 1);
504        assert!(!all[0].active, "pattern should be deactivated, not deleted");
505    }
506
507    #[test]
508    fn test_delete_oversized_patterns_returns_count() {
509        let store = KnowledgeStore::open_memory().expect("operation should succeed");
510
511        for i in 0..3 {
512            let mut pattern = Pattern::new(
513                format!("pattern {i}"),
514                "x".repeat(5000 + i * 1000),
515                "test".to_string(),
516            );
517            pattern.confidence = 0.9;
518            store
519                .insert_pattern(&pattern)
520                .expect("operation should succeed");
521        }
522
523        let count = store
524            .delete_oversized_patterns(5500)
525            .expect("operation should succeed");
526        assert_eq!(count, 2);
527    }
528
529    #[test]
530    fn test_pattern_roundtrip_preserves_content_hash() {
531        let store = KnowledgeStore::open_memory().expect("operation should succeed");
532
533        let mut pattern = Pattern::new(
534            "Test".to_string(),
535            "Test instruction content".to_string(),
536            "test".to_string(),
537        );
538        pattern.confidence = 0.8;
539        let original_hash = pattern.content_hash.clone();
540        store
541            .insert_pattern(&pattern)
542            .expect("operation should succeed");
543
544        let patterns = store
545            .get_active_patterns(0.0)
546            .expect("operation should succeed");
547        assert_eq!(patterns.len(), 1);
548        assert_eq!(patterns[0].content_hash, original_hash);
549    }
550
551    #[test]
552    fn test_hard_reset_on_v1_schema_db() {
553        let dir = tempfile::tempdir().expect("tempdir");
554        let db_path = dir.path().join("knowledge.db");
555
556        // Create a v1-schema database (without the new MenteDB columns).
557        {
558            let conn = Connection::open(&db_path).expect("operation should succeed");
559            conn.execute_batch(
560                "CREATE TABLE observations (
561                    id TEXT PRIMARY KEY, signal_type TEXT NOT NULL, intensity REAL NOT NULL,
562                    context TEXT NOT NULL, timestamp TEXT NOT NULL, session_id TEXT, turn_number INTEGER
563                );
564                CREATE TABLE patterns (
565                    id TEXT PRIMARY KEY, description TEXT NOT NULL, instruction TEXT NOT NULL,
566                    confidence REAL NOT NULL, evidence_count INTEGER NOT NULL,
567                    first_observed TEXT NOT NULL, last_updated TEXT NOT NULL,
568                    category TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1,
569                    content_hash TEXT NOT NULL DEFAULT ''
570                );
571                CREATE TABLE conflicts (
572                    id TEXT PRIMARY KEY, pattern_a_id TEXT NOT NULL, pattern_b_id TEXT NOT NULL,
573                    description TEXT NOT NULL, detected_at TEXT NOT NULL,
574                    resolved INTEGER NOT NULL DEFAULT 0, winner_id TEXT
575                );",
576            )
577            .expect("operation should succeed");
578
579            // Insert v1 data.
580            conn.execute(
581                "INSERT INTO observations (id, signal_type, intensity, context, timestamp, session_id, turn_number)
582                 VALUES ('v1-obs', 'Correction', 0.8, 'v1 context', '2026-01-01T00:00:00Z', 'sess-1', 1)",
583                [],
584            )
585            .expect("operation should succeed");
586            conn.execute(
587                "INSERT INTO patterns (id, description, instruction, confidence, evidence_count, first_observed, last_updated, category, active, content_hash)
588                 VALUES ('v1-pat', 'v1 pattern', 'v1 instruction', 0.9, 5, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', 'preference', 1, 'hash123')",
589                [],
590            )
591            .expect("operation should succeed");
592        }
593
594        // Open with new code — should detect v1 schema and hard-reset.
595        let store = KnowledgeStore::open(db_path.to_str().expect("operation should succeed"))
596            .expect("open v1 db");
597
598        // Data should be wiped.
599        let observations = store.get_observations().expect("operation should succeed");
600        assert!(
601            observations.is_empty(),
602            "v1 observations should be wiped, got {:?}",
603            observations
604        );
605        let patterns = store.get_all_patterns().expect("operation should succeed");
606        assert!(
607            patterns.is_empty(),
608            "v1 patterns should be wiped, got {:?}",
609            patterns
610        );
611
612        // Second open should be idempotent (no re-reset, schema is now v2).
613        drop(store);
614        let store2 = KnowledgeStore::open(db_path.to_str().expect("operation should succeed"))
615            .expect("reopen v2 db");
616        let observations2 = store2.get_observations().expect("operation should succeed");
617        assert!(observations2.is_empty(), "second open should not re-reset");
618    }
619}
talos_evolution/store.rs

talos_evolution/
store.rs
