1use crate::embeddings::{BoxedEmbeddingProvider, EmbeddingSpaceAccess, HashEmbedding};
28use crate::error::{Result, RuvectorError};
29use crate::types::*;
30use crate::vector_db::VectorDB;
31use parking_lot::RwLock;
32use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
33use serde::{Deserialize, Serialize};
34use std::collections::HashMap;
35use std::sync::Arc;
36
37const REFLEXION_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("reflexion_episodes");
39const SKILLS_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("skills_library");
40const CAUSAL_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("causal_edges");
41const LEARNING_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("learning_sessions");
42const EMBEDDING_SPACE_TABLE: TableDefinition<&str, &[u8]> =
44 TableDefinition::new("embedding_space_identity");
45const EMBEDDING_SPACE_KEY: &str = "__ruvector_embedding_space__";
48
49#[derive(Debug, Clone, Serialize, Deserialize)]
52pub struct ReflexionEpisode {
53 pub id: String,
54 pub task: String,
55 pub actions: Vec<String>,
56 pub observations: Vec<String>,
57 pub critique: String,
58 pub embedding: Vec<f32>,
59 pub timestamp: i64,
60 pub metadata: Option<HashMap<String, serde_json::Value>>,
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
65pub struct Skill {
66 pub id: String,
67 pub name: String,
68 pub description: String,
69 pub parameters: HashMap<String, String>,
70 pub examples: Vec<String>,
71 pub embedding: Vec<f32>,
72 pub usage_count: usize,
73 pub success_rate: f64,
74 pub created_at: i64,
75 pub updated_at: i64,
76}
77
78#[derive(Debug, Clone, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
80pub struct CausalEdge {
81 pub id: String,
82 pub causes: Vec<String>, pub effects: Vec<String>, pub confidence: f64,
85 pub context: String,
86 pub embedding: Vec<f32>,
87 pub observations: usize,
88 pub timestamp: i64,
89}
90
91#[derive(Debug, Clone, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
93pub struct LearningSession {
94 pub id: String,
95 pub algorithm: String, pub state_dim: usize,
97 pub action_dim: usize,
98 pub experiences: Vec<Experience>,
99 pub model_params: Option<Vec<u8>>, pub created_at: i64,
101 pub updated_at: i64,
102}
103
104#[derive(Debug, Clone, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
106pub struct Experience {
107 pub state: Vec<f32>,
108 pub action: Vec<f32>,
109 pub reward: f64,
110 pub next_state: Vec<f32>,
111 pub done: bool,
112 pub timestamp: i64,
113}
114
115#[derive(Debug, Clone, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
117pub struct Prediction {
118 pub action: Vec<f32>,
119 pub confidence_lower: f64,
120 pub confidence_upper: f64,
121 pub mean_confidence: f64,
122}
123
124#[derive(Debug, Clone, Serialize, Deserialize)]
126pub struct UtilitySearchResult {
127 pub result: SearchResult,
128 pub utility_score: f64,
129 pub similarity_score: f64,
130 pub causal_uplift: f64,
131 pub latency_penalty: f64,
132}
133
134pub struct AgenticDB {
136 vector_db: Arc<VectorDB>,
137 db: Arc<Database>,
138 _dimensions: usize,
139 embedding_provider: BoxedEmbeddingProvider,
140}
141
142impl AgenticDB {
143 pub fn new(options: DbOptions) -> Result<Self> {
145 let embedding_provider = Arc::new(HashEmbedding::new(options.dimensions));
146 Self::with_embedding_provider(options, embedding_provider)
147 }
148
149 pub fn with_embedding_provider(
188 options: DbOptions,
189 embedding_provider: BoxedEmbeddingProvider,
190 ) -> Result<Self> {
191 if options.dimensions != embedding_provider.dimensions() {
193 return Err(RuvectorError::InvalidDimension(format!(
194 "Options dimensions ({}) do not match embedding provider dimensions ({})",
195 options.dimensions,
196 embedding_provider.dimensions()
197 )));
198 }
199
200 let vector_db = Arc::new(VectorDB::new(options.clone())?);
202
203 let agentic_path = format!("{}.agentic", options.storage_path);
205 let db = Arc::new(Database::create(&agentic_path)?);
206
207 let write_txn = db.begin_write()?;
209 {
210 let reflexion = write_txn.open_table(REFLEXION_TABLE)?;
211 let skills = write_txn.open_table(SKILLS_TABLE)?;
212 let causal = write_txn.open_table(CAUSAL_TABLE)?;
213 let _ = write_txn.open_table(LEARNING_TABLE)?;
214
215 let mut spaces = write_txn.open_table(EMBEDDING_SPACE_TABLE)?;
228 let active = embedding_provider.embedding_space();
229 let active_id = active.embedding_space_id()?;
230 let stored_json = match vector_db.load_config_value(EMBEDDING_SPACE_KEY)? {
231 Some(primary) => Some(primary),
232 None => spaces
233 .get(EMBEDDING_SPACE_KEY)?
234 .map(|value| {
235 String::from_utf8(value.value().to_vec())
238 .map_err(|e| RuvectorError::SerializationError(e.to_string()))
239 })
240 .transpose()?,
241 };
242 match stored_json {
243 Some(json) => {
244 let stored: crate::embeddings::EmbeddingSpaceIdentity =
245 serde_json::from_str(&json)
246 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
247 let access = EmbeddingSpaceAccess::for_identities(&stored, active)?;
253 if !access.text_embedding || !access.corpus_mutation {
254 return Err(RuvectorError::EmbeddingSpaceMismatch {
255 stored: stored.embedding_space_id()?,
256 active: active_id,
257 });
258 }
259 vector_db.save_config_value(EMBEDDING_SPACE_KEY, &json)?;
263 spaces.insert(EMBEDDING_SPACE_KEY, json.as_bytes())?;
264 }
265 None => {
266 let populated = !vector_db.is_empty()?
275 || !reflexion.is_empty()?
276 || !skills.is_empty()?
277 || !causal.is_empty()?;
278 if populated {
279 tracing::warn!(
280 embedding_space_id = %active_id,
281 provider = %active.provider,
282 model_id = %active.model_id,
283 vectors = vector_db.len()?,
284 "corpus holds data but has no recorded embedding-space identity \
285 (pre-ADR-281 store, or its provenance was removed); adopting the \
286 active provider's identity WITHOUT verification — vectors already \
287 in this store may come from a different embedding space"
288 );
289 }
290 let json = serde_json::to_string(active)
291 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
292 vector_db.save_config_value(EMBEDDING_SPACE_KEY, &json)?;
293 spaces.insert(EMBEDDING_SPACE_KEY, json.as_bytes())?;
294 }
295 }
296 }
297 write_txn.commit()?;
298
299 Ok(Self {
300 vector_db,
301 db,
302 _dimensions: options.dimensions,
303 embedding_provider,
304 })
305 }
306
307 pub fn with_dimensions(dimensions: usize) -> Result<Self> {
309 let options = DbOptions {
310 dimensions,
311 ..DbOptions::default()
312 };
313 Self::new(options)
314 }
315
316 pub fn embedding_provider_name(&self) -> &str {
318 self.embedding_provider.name()
319 }
320
321 pub fn insert(&self, entry: VectorEntry) -> Result<VectorId> {
325 self.vector_db.insert(entry)
326 }
327
328 pub fn insert_batch(&self, entries: Vec<VectorEntry>) -> Result<Vec<VectorId>> {
330 self.vector_db.insert_batch(entries)
331 }
332
333 pub fn search(&self, query: SearchQuery) -> Result<Vec<SearchResult>> {
335 self.vector_db.search(query)
336 }
337
338 pub fn delete(&self, id: &str) -> Result<bool> {
340 self.vector_db.delete(id)
341 }
342
343 pub fn get(&self, id: &str) -> Result<Option<VectorEntry>> {
345 self.vector_db.get(id)
346 }
347
348 pub fn store_episode(
352 &self,
353 task: String,
354 actions: Vec<String>,
355 observations: Vec<String>,
356 critique: String,
357 ) -> Result<String> {
358 let id = uuid::Uuid::new_v4().to_string();
359
360 let embedding = self.generate_passage_embedding(&critique)?;
362
363 let episode = ReflexionEpisode {
364 id: id.clone(),
365 task,
366 actions,
367 observations,
368 critique,
369 embedding: embedding.clone(),
370 timestamp: chrono::Utc::now().timestamp(),
371 metadata: None,
372 };
373
374 let write_txn = self.db.begin_write()?;
376 {
377 let mut table = write_txn.open_table(REFLEXION_TABLE)?;
378 let json = serde_json::to_vec(&episode)
380 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
381 table.insert(id.as_str(), json.as_slice())?;
382 }
383 write_txn.commit()?;
384
385 self.vector_db.insert(VectorEntry {
387 id: Some(format!("reflexion_{}", id)),
388 vector: embedding,
389 metadata: Some({
390 let mut meta = HashMap::new();
391 meta.insert("type".to_string(), serde_json::json!("reflexion"));
392 meta.insert("episode_id".to_string(), serde_json::json!(id.clone()));
393 meta
394 }),
395 })?;
396
397 Ok(id)
398 }
399
400 pub fn retrieve_similar_episodes(
402 &self,
403 query: &str,
404 k: usize,
405 ) -> Result<Vec<ReflexionEpisode>> {
406 let query_embedding = self.generate_query_embedding(query)?;
408
409 let results = self.vector_db.search(SearchQuery {
411 vector: query_embedding,
412 k,
413 filter: Some({
414 let mut filter = HashMap::new();
415 filter.insert("type".to_string(), serde_json::json!("reflexion"));
416 filter
417 }),
418 ef_search: None,
419 })?;
420
421 let mut episodes = Vec::new();
423 let read_txn = self.db.begin_read()?;
424 let table = read_txn.open_table(REFLEXION_TABLE)?;
425
426 for result in results {
427 if let Some(metadata) = result.metadata {
428 if let Some(episode_id) = metadata.get("episode_id") {
429 let id = episode_id.as_str().unwrap();
430 if let Some(data) = table.get(id)? {
431 let episode: ReflexionEpisode = serde_json::from_slice(data.value())
433 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
434 episodes.push(episode);
435 }
436 }
437 }
438 }
439
440 Ok(episodes)
441 }
442
443 pub fn create_skill(
447 &self,
448 name: String,
449 description: String,
450 parameters: HashMap<String, String>,
451 examples: Vec<String>,
452 ) -> Result<String> {
453 let id = uuid::Uuid::new_v4().to_string();
454
455 let embedding = self.generate_passage_embedding(&description)?;
457
458 let skill = Skill {
459 id: id.clone(),
460 name,
461 description,
462 parameters,
463 examples,
464 embedding: embedding.clone(),
465 usage_count: 0,
466 success_rate: 0.0,
467 created_at: chrono::Utc::now().timestamp(),
468 updated_at: chrono::Utc::now().timestamp(),
469 };
470
471 let write_txn = self.db.begin_write()?;
473 {
474 let mut table = write_txn.open_table(SKILLS_TABLE)?;
475 let data = bincode::encode_to_vec(&skill, bincode::config::standard())
476 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
477 table.insert(id.as_str(), data.as_slice())?;
478 }
479 write_txn.commit()?;
480
481 self.vector_db.insert(VectorEntry {
483 id: Some(format!("skill_{}", id)),
484 vector: embedding,
485 metadata: Some({
486 let mut meta = HashMap::new();
487 meta.insert("type".to_string(), serde_json::json!("skill"));
488 meta.insert("skill_id".to_string(), serde_json::json!(id.clone()));
489 meta
490 }),
491 })?;
492
493 Ok(id)
494 }
495
496 pub fn search_skills(&self, query_description: &str, k: usize) -> Result<Vec<Skill>> {
498 let query_embedding = self.generate_query_embedding(query_description)?;
499
500 let results = self.vector_db.search(SearchQuery {
501 vector: query_embedding,
502 k,
503 filter: Some({
504 let mut filter = HashMap::new();
505 filter.insert("type".to_string(), serde_json::json!("skill"));
506 filter
507 }),
508 ef_search: None,
509 })?;
510
511 let mut skills = Vec::new();
512 let read_txn = self.db.begin_read()?;
513 let table = read_txn.open_table(SKILLS_TABLE)?;
514
515 for result in results {
516 if let Some(metadata) = result.metadata {
517 if let Some(skill_id) = metadata.get("skill_id") {
518 let id = skill_id.as_str().unwrap();
519 if let Some(data) = table.get(id)? {
520 let (skill, _): (Skill, usize) =
521 bincode::decode_from_slice(data.value(), bincode::config::standard())
522 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
523 skills.push(skill);
524 }
525 }
526 }
527 }
528
529 Ok(skills)
530 }
531
532 pub fn auto_consolidate(
534 &self,
535 action_sequences: Vec<Vec<String>>,
536 success_threshold: usize,
537 ) -> Result<Vec<String>> {
538 let mut skill_ids = Vec::new();
539
540 for sequence in action_sequences {
542 if sequence.len() >= success_threshold {
543 let description = format!("Skill: {}", sequence.join(" -> "));
544 let skill_id = self.create_skill(
545 format!("Auto-Skill-{}", uuid::Uuid::new_v4()),
546 description,
547 HashMap::new(),
548 sequence.clone(),
549 )?;
550 skill_ids.push(skill_id);
551 }
552 }
553
554 Ok(skill_ids)
555 }
556
557 pub fn add_causal_edge(
561 &self,
562 causes: Vec<String>,
563 effects: Vec<String>,
564 confidence: f64,
565 context: String,
566 ) -> Result<String> {
567 let id = uuid::Uuid::new_v4().to_string();
568
569 let embedding = self.generate_passage_embedding(&context)?;
571
572 let edge = CausalEdge {
573 id: id.clone(),
574 causes,
575 effects,
576 confidence,
577 context,
578 embedding: embedding.clone(),
579 observations: 1,
580 timestamp: chrono::Utc::now().timestamp(),
581 };
582
583 let write_txn = self.db.begin_write()?;
585 {
586 let mut table = write_txn.open_table(CAUSAL_TABLE)?;
587 let data = bincode::encode_to_vec(&edge, bincode::config::standard())
588 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
589 table.insert(id.as_str(), data.as_slice())?;
590 }
591 write_txn.commit()?;
592
593 self.vector_db.insert(VectorEntry {
595 id: Some(format!("causal_{}", id)),
596 vector: embedding,
597 metadata: Some({
598 let mut meta = HashMap::new();
599 meta.insert("type".to_string(), serde_json::json!("causal"));
600 meta.insert("causal_id".to_string(), serde_json::json!(id.clone()));
601 meta.insert("confidence".to_string(), serde_json::json!(confidence));
602 meta
603 }),
604 })?;
605
606 Ok(id)
607 }
608
609 pub fn query_with_utility(
611 &self,
612 query: &str,
613 k: usize,
614 alpha: f64,
615 beta: f64,
616 gamma: f64,
617 ) -> Result<Vec<UtilitySearchResult>> {
618 let start_time = std::time::Instant::now();
619 let query_embedding = self.generate_query_embedding(query)?;
620
621 let results = self.vector_db.search(SearchQuery {
623 vector: query_embedding,
624 k: k * 2, filter: Some({
626 let mut filter = HashMap::new();
627 filter.insert("type".to_string(), serde_json::json!("causal"));
628 filter
629 }),
630 ef_search: None,
631 })?;
632
633 let mut utility_results = Vec::new();
634
635 for result in results {
636 let similarity_score = 1.0 / (1.0 + result.score as f64); let causal_uplift = if let Some(ref metadata) = result.metadata {
640 metadata
641 .get("confidence")
642 .and_then(|v| v.as_f64())
643 .unwrap_or(0.0)
644 } else {
645 0.0
646 };
647
648 let latency = start_time.elapsed().as_secs_f64();
649 let latency_penalty = latency * gamma;
650
651 let utility_score = alpha * similarity_score + beta * causal_uplift - latency_penalty;
653
654 utility_results.push(UtilitySearchResult {
655 result,
656 utility_score,
657 similarity_score,
658 causal_uplift,
659 latency_penalty,
660 });
661 }
662
663 utility_results.sort_by(|a, b| b.utility_score.partial_cmp(&a.utility_score).unwrap());
665 utility_results.truncate(k);
666
667 Ok(utility_results)
668 }
669
670 pub fn start_session(
674 &self,
675 algorithm: String,
676 state_dim: usize,
677 action_dim: usize,
678 ) -> Result<String> {
679 let id = uuid::Uuid::new_v4().to_string();
680
681 let session = LearningSession {
682 id: id.clone(),
683 algorithm,
684 state_dim,
685 action_dim,
686 experiences: Vec::new(),
687 model_params: None,
688 created_at: chrono::Utc::now().timestamp(),
689 updated_at: chrono::Utc::now().timestamp(),
690 };
691
692 let write_txn = self.db.begin_write()?;
693 {
694 let mut table = write_txn.open_table(LEARNING_TABLE)?;
695 let data = bincode::encode_to_vec(&session, bincode::config::standard())
696 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
697 table.insert(id.as_str(), data.as_slice())?;
698 }
699 write_txn.commit()?;
700
701 Ok(id)
702 }
703
704 pub fn add_experience(
706 &self,
707 session_id: &str,
708 state: Vec<f32>,
709 action: Vec<f32>,
710 reward: f64,
711 next_state: Vec<f32>,
712 done: bool,
713 ) -> Result<()> {
714 let read_txn = self.db.begin_read()?;
715 let table = read_txn.open_table(LEARNING_TABLE)?;
716
717 let data = table
718 .get(session_id)?
719 .ok_or_else(|| RuvectorError::VectorNotFound(session_id.to_string()))?;
720
721 let (mut session, _): (LearningSession, usize) =
722 bincode::decode_from_slice(data.value(), bincode::config::standard())
723 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
724
725 drop(table);
726 drop(read_txn);
727
728 session.experiences.push(Experience {
730 state,
731 action,
732 reward,
733 next_state,
734 done,
735 timestamp: chrono::Utc::now().timestamp(),
736 });
737 session.updated_at = chrono::Utc::now().timestamp();
738
739 let write_txn = self.db.begin_write()?;
741 {
742 let mut table = write_txn.open_table(LEARNING_TABLE)?;
743 let data = bincode::encode_to_vec(&session, bincode::config::standard())
744 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
745 table.insert(session_id, data.as_slice())?;
746 }
747 write_txn.commit()?;
748
749 Ok(())
750 }
751
752 pub fn predict_with_confidence(&self, session_id: &str, state: Vec<f32>) -> Result<Prediction> {
754 let read_txn = self.db.begin_read()?;
755 let table = read_txn.open_table(LEARNING_TABLE)?;
756
757 let data = table
758 .get(session_id)?
759 .ok_or_else(|| RuvectorError::VectorNotFound(session_id.to_string()))?;
760
761 let (session, _): (LearningSession, usize) =
762 bincode::decode_from_slice(data.value(), bincode::config::standard())
763 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
764
765 let mut similar_actions = Vec::new();
767 let mut rewards = Vec::new();
768
769 for exp in &session.experiences {
770 let distance = euclidean_distance(&state, &exp.state);
771 if distance < 1.0 {
772 similar_actions.push(exp.action.clone());
774 rewards.push(exp.reward);
775 }
776 }
777
778 if similar_actions.is_empty() {
779 return Ok(Prediction {
781 action: vec![0.0; session.action_dim],
782 confidence_lower: 0.0,
783 confidence_upper: 0.0,
784 mean_confidence: 0.0,
785 });
786 }
787
788 let total_reward: f64 = rewards.iter().sum();
790 let mut action = vec![0.0; session.action_dim];
791
792 for (act, reward) in similar_actions.iter().zip(rewards.iter()) {
793 let weight = reward / total_reward;
794 for (i, val) in act.iter().enumerate() {
795 action[i] += val * weight as f32;
796 }
797 }
798
799 let mean_reward = total_reward / rewards.len() as f64;
801 let std_dev = calculate_std_dev(&rewards, mean_reward);
802
803 Ok(Prediction {
804 action,
805 confidence_lower: mean_reward - 1.96 * std_dev,
806 confidence_upper: mean_reward + 1.96 * std_dev,
807 mean_confidence: mean_reward,
808 })
809 }
810
811 pub fn get_session(&self, session_id: &str) -> Result<Option<LearningSession>> {
813 let read_txn = self.db.begin_read()?;
814 let table = read_txn.open_table(LEARNING_TABLE)?;
815
816 if let Some(data) = table.get(session_id)? {
817 let (session, _): (LearningSession, usize) =
818 bincode::decode_from_slice(data.value(), bincode::config::standard())
819 .map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
820 Ok(Some(session))
821 } else {
822 Ok(None)
823 }
824 }
825
826 fn generate_query_embedding(&self, text: &str) -> Result<Vec<f32>> {
850 self.embedding_provider.embed_query(text)
851 }
852
853 fn generate_passage_embedding(&self, text: &str) -> Result<Vec<f32>> {
854 self.embedding_provider.embed_passage(text)
855 }
856}
857
858fn euclidean_distance(a: &[f32], b: &[f32]) -> f32 {
860 a.iter()
861 .zip(b.iter())
862 .map(|(x, y)| (x - y).powi(2))
863 .sum::<f32>()
864 .sqrt()
865}
866
867fn calculate_std_dev(values: &[f64], mean: f64) -> f64 {
868 let variance = values.iter().map(|x| (x - mean).powi(2)).sum::<f64>() / values.len() as f64;
869 variance.sqrt()
870}
871
872pub struct PolicyMemoryStore<'a> {
886 db: &'a AgenticDB,
887}
888
889#[derive(Debug, Clone, Serialize, Deserialize)]
891pub struct PolicyAction {
892 pub action: String,
894 pub reward: f64,
896 pub q_value: f64,
898 pub state_embedding: Vec<f32>,
900 pub timestamp: i64,
902}
903
904#[derive(Debug, Clone, Serialize, Deserialize)]
906pub struct PolicyEntry {
907 pub id: String,
909 pub state_id: String,
911 pub action: PolicyAction,
913 pub metadata: Option<HashMap<String, serde_json::Value>>,
915}
916
917impl<'a> PolicyMemoryStore<'a> {
918 pub fn new(db: &'a AgenticDB) -> Self {
920 Self { db }
921 }
922
923 pub fn store_policy(
925 &self,
926 state_id: &str,
927 state_embedding: Vec<f32>,
928 action: &str,
929 reward: f64,
930 q_value: f64,
931 ) -> Result<String> {
932 let id = uuid::Uuid::new_v4().to_string();
933 let timestamp = chrono::Utc::now().timestamp();
934
935 let _entry = PolicyEntry {
936 id: id.clone(),
937 state_id: state_id.to_string(),
938 action: PolicyAction {
939 action: action.to_string(),
940 reward,
941 q_value,
942 state_embedding: state_embedding.clone(),
943 timestamp,
944 },
945 metadata: None,
946 };
947
948 self.db.vector_db.insert(VectorEntry {
950 id: Some(format!("policy_{}", id)),
951 vector: state_embedding,
952 metadata: Some({
953 let mut meta = HashMap::new();
954 meta.insert("type".to_string(), serde_json::json!("policy"));
955 meta.insert("policy_id".to_string(), serde_json::json!(id.clone()));
956 meta.insert("state_id".to_string(), serde_json::json!(state_id));
957 meta.insert("action".to_string(), serde_json::json!(action));
958 meta.insert("reward".to_string(), serde_json::json!(reward));
959 meta.insert("q_value".to_string(), serde_json::json!(q_value));
960 meta
961 }),
962 })?;
963
964 Ok(id)
965 }
966
967 pub fn retrieve_similar_states(
969 &self,
970 state_embedding: &[f32],
971 k: usize,
972 ) -> Result<Vec<PolicyEntry>> {
973 let results = self.db.vector_db.search(SearchQuery {
974 vector: state_embedding.to_vec(),
975 k,
976 filter: Some({
977 let mut filter = HashMap::new();
978 filter.insert("type".to_string(), serde_json::json!("policy"));
979 filter
980 }),
981 ef_search: None,
982 })?;
983
984 let mut entries = Vec::new();
985 for result in results {
986 if let Some(metadata) = result.metadata {
987 let policy_id = metadata
988 .get("policy_id")
989 .and_then(|v| v.as_str())
990 .unwrap_or("");
991 let state_id = metadata
992 .get("state_id")
993 .and_then(|v| v.as_str())
994 .unwrap_or("");
995 let action = metadata
996 .get("action")
997 .and_then(|v| v.as_str())
998 .unwrap_or("");
999 let reward = metadata
1000 .get("reward")
1001 .and_then(|v| v.as_f64())
1002 .unwrap_or(0.0);
1003 let q_value = metadata
1004 .get("q_value")
1005 .and_then(|v| v.as_f64())
1006 .unwrap_or(0.0);
1007
1008 entries.push(PolicyEntry {
1009 id: policy_id.to_string(),
1010 state_id: state_id.to_string(),
1011 action: PolicyAction {
1012 action: action.to_string(),
1013 reward,
1014 q_value,
1015 state_embedding: result.vector.unwrap_or_default(),
1016 timestamp: 0,
1017 },
1018 metadata: None,
1019 });
1020 }
1021 }
1022
1023 Ok(entries)
1024 }
1025
1026 pub fn get_best_action(&self, state_embedding: &[f32], k: usize) -> Result<Option<String>> {
1028 let similar = self.retrieve_similar_states(state_embedding, k)?;
1029
1030 similar
1031 .into_iter()
1032 .max_by(|a, b| a.action.q_value.partial_cmp(&b.action.q_value).unwrap())
1033 .map(|entry| Ok(entry.action.action))
1034 .transpose()
1035 }
1036
1037 pub fn update_q_value(&self, policy_id: &str, new_q_value: f64) -> Result<()> {
1043 let key = format!("policy_{}", policy_id);
1044
1045 let mut entry = self
1047 .db
1048 .vector_db
1049 .get(&key)?
1050 .ok_or_else(|| crate::error::RuvectorError::VectorNotFound(key.clone()))?;
1051
1052 let mut meta = entry.metadata.take().unwrap_or_default();
1054 meta.insert("q_value".to_string(), serde_json::json!(new_q_value));
1055 entry.metadata = Some(meta);
1056
1057 let _ = self.db.vector_db.delete(&key);
1060 self.db.vector_db.insert(entry)?;
1061 Ok(())
1062 }
1063}
1064
1065pub struct SessionStateIndex<'a> {
1070 db: &'a AgenticDB,
1071 session_id: String,
1072 ttl_seconds: i64,
1073}
1074
1075#[derive(Debug, Clone, Serialize, Deserialize)]
1077pub struct SessionTurn {
1078 pub id: String,
1080 pub session_id: String,
1082 pub turn_number: usize,
1084 pub role: String,
1086 pub content: String,
1088 pub embedding: Vec<f32>,
1090 pub timestamp: i64,
1092 pub expires_at: i64,
1094}
1095
1096impl<'a> SessionStateIndex<'a> {
1097 pub fn new(db: &'a AgenticDB, session_id: &str, ttl_seconds: i64) -> Self {
1099 Self {
1100 db,
1101 session_id: session_id.to_string(),
1102 ttl_seconds,
1103 }
1104 }
1105
1106 pub fn add_turn(&self, turn_number: usize, role: &str, content: &str) -> Result<String> {
1108 let id = uuid::Uuid::new_v4().to_string();
1109 let timestamp = chrono::Utc::now().timestamp();
1110 let expires_at = timestamp + self.ttl_seconds;
1111
1112 let embedding = self.db.generate_passage_embedding(content)?;
1114
1115 self.db.vector_db.insert(VectorEntry {
1117 id: Some(format!("session_{}_{}", self.session_id, id)),
1118 vector: embedding,
1119 metadata: Some({
1120 let mut meta = HashMap::new();
1121 meta.insert("type".to_string(), serde_json::json!("session_turn"));
1122 meta.insert(
1123 "session_id".to_string(),
1124 serde_json::json!(self.session_id.clone()),
1125 );
1126 meta.insert("turn_id".to_string(), serde_json::json!(id.clone()));
1127 meta.insert("turn_number".to_string(), serde_json::json!(turn_number));
1128 meta.insert("role".to_string(), serde_json::json!(role));
1129 meta.insert("content".to_string(), serde_json::json!(content));
1130 meta.insert("timestamp".to_string(), serde_json::json!(timestamp));
1131 meta.insert("expires_at".to_string(), serde_json::json!(expires_at));
1132 meta
1133 }),
1134 })?;
1135
1136 Ok(id)
1137 }
1138
1139 pub fn find_relevant_turns(&self, query: &str, k: usize) -> Result<Vec<SessionTurn>> {
1141 let query_embedding = self.db.generate_query_embedding(query)?;
1142 let current_time = chrono::Utc::now().timestamp();
1143
1144 let results = self.db.vector_db.search(SearchQuery {
1145 vector: query_embedding,
1146 k: k * 2, filter: Some({
1148 let mut filter = HashMap::new();
1149 filter.insert("type".to_string(), serde_json::json!("session_turn"));
1150 filter.insert(
1151 "session_id".to_string(),
1152 serde_json::json!(self.session_id.clone()),
1153 );
1154 filter
1155 }),
1156 ef_search: None,
1157 })?;
1158
1159 let mut turns = Vec::new();
1160 for result in results {
1161 if let Some(metadata) = result.metadata {
1162 let expires_at = metadata
1163 .get("expires_at")
1164 .and_then(|v| v.as_i64())
1165 .unwrap_or(0);
1166
1167 if expires_at < current_time {
1169 continue;
1170 }
1171
1172 turns.push(SessionTurn {
1173 id: metadata
1174 .get("turn_id")
1175 .and_then(|v| v.as_str())
1176 .unwrap_or("")
1177 .to_string(),
1178 session_id: self.session_id.clone(),
1179 turn_number: metadata
1180 .get("turn_number")
1181 .and_then(|v| v.as_u64())
1182 .unwrap_or(0) as usize,
1183 role: metadata
1184 .get("role")
1185 .and_then(|v| v.as_str())
1186 .unwrap_or("")
1187 .to_string(),
1188 content: metadata
1189 .get("content")
1190 .and_then(|v| v.as_str())
1191 .unwrap_or("")
1192 .to_string(),
1193 embedding: result.vector.unwrap_or_default(),
1194 timestamp: metadata
1195 .get("timestamp")
1196 .and_then(|v| v.as_i64())
1197 .unwrap_or(0),
1198 expires_at,
1199 });
1200
1201 if turns.len() >= k {
1202 break;
1203 }
1204 }
1205 }
1206
1207 Ok(turns)
1208 }
1209
1210 pub fn get_session_context(&self) -> Result<Vec<SessionTurn>> {
1212 let mut turns = self.find_relevant_turns("", 1000)?;
1213 turns.sort_by_key(|t| t.turn_number);
1214 Ok(turns)
1215 }
1216
1217 pub fn cleanup_expired(&self) -> Result<usize> {
1219 let current_time = chrono::Utc::now().timestamp();
1220 let all_turns = self.find_relevant_turns("", 10000)?;
1221 let mut deleted = 0;
1222
1223 for turn in all_turns {
1224 if turn.expires_at < current_time {
1225 let _ = self
1226 .db
1227 .vector_db
1228 .delete(&format!("session_{}_{}", self.session_id, turn.id));
1229 deleted += 1;
1230 }
1231 }
1232
1233 Ok(deleted)
1234 }
1235}
1236
1237pub struct WitnessLog<'a> {
1241 db: &'a AgenticDB,
1242 last_hash: RwLock<Option<String>>,
1243}
1244
1245#[derive(Debug, Clone, Serialize, Deserialize)]
1247pub struct WitnessEntry {
1248 pub id: String,
1250 pub prev_hash: Option<String>,
1252 pub hash: String,
1254 pub agent_id: String,
1256 pub action_type: String,
1258 pub details: String,
1260 pub embedding: Vec<f32>,
1262 pub timestamp: i64,
1264 pub metadata: Option<HashMap<String, serde_json::Value>>,
1266}
1267
1268impl<'a> WitnessLog<'a> {
1269 pub fn new(db: &'a AgenticDB) -> Self {
1271 Self {
1272 db,
1273 last_hash: RwLock::new(None),
1274 }
1275 }
1276
1277 fn compute_hash(
1279 prev_hash: &Option<String>,
1280 agent_id: &str,
1281 action_type: &str,
1282 details: &str,
1283 timestamp: i64,
1284 ) -> String {
1285 use std::collections::hash_map::DefaultHasher;
1286 use std::hash::{Hash, Hasher};
1287
1288 let mut hasher = DefaultHasher::new();
1289 if let Some(prev) = prev_hash {
1290 prev.hash(&mut hasher);
1291 }
1292 agent_id.hash(&mut hasher);
1293 action_type.hash(&mut hasher);
1294 details.hash(&mut hasher);
1295 timestamp.hash(&mut hasher);
1296 format!("{:016x}", hasher.finish())
1297 }
1298
1299 pub fn append(&self, agent_id: &str, action_type: &str, details: &str) -> Result<String> {
1301 let id = uuid::Uuid::new_v4().to_string();
1302 let timestamp = chrono::Utc::now().timestamp();
1303
1304 let prev_hash = self.last_hash.read().clone();
1306
1307 let hash = Self::compute_hash(&prev_hash, agent_id, action_type, details, timestamp);
1309
1310 let embedding = self
1312 .db
1313 .generate_passage_embedding(&format!("{} {} {}", agent_id, action_type, details))?;
1314
1315 self.db.vector_db.insert(VectorEntry {
1317 id: Some(format!("witness_{}", id)),
1318 vector: embedding.clone(),
1319 metadata: Some({
1320 let mut meta = HashMap::new();
1321 meta.insert("type".to_string(), serde_json::json!("witness"));
1322 meta.insert("witness_id".to_string(), serde_json::json!(id.clone()));
1323 meta.insert("agent_id".to_string(), serde_json::json!(agent_id));
1324 meta.insert("action_type".to_string(), serde_json::json!(action_type));
1325 meta.insert("details".to_string(), serde_json::json!(details));
1326 meta.insert("timestamp".to_string(), serde_json::json!(timestamp));
1327 meta.insert("hash".to_string(), serde_json::json!(hash.clone()));
1328 if let Some(ref prev) = prev_hash {
1329 meta.insert("prev_hash".to_string(), serde_json::json!(prev));
1330 }
1331 meta
1332 }),
1333 })?;
1334
1335 *self.last_hash.write() = Some(hash.clone());
1337
1338 Ok(id)
1339 }
1340
1341 pub fn search(&self, query: &str, k: usize) -> Result<Vec<WitnessEntry>> {
1343 let query_embedding = self.db.generate_query_embedding(query)?;
1344
1345 let results = self.db.vector_db.search(SearchQuery {
1346 vector: query_embedding,
1347 k,
1348 filter: Some({
1349 let mut filter = HashMap::new();
1350 filter.insert("type".to_string(), serde_json::json!("witness"));
1351 filter
1352 }),
1353 ef_search: None,
1354 })?;
1355
1356 let mut entries = Vec::new();
1357 for result in results {
1358 if let Some(metadata) = result.metadata {
1359 entries.push(WitnessEntry {
1360 id: metadata
1361 .get("witness_id")
1362 .and_then(|v| v.as_str())
1363 .unwrap_or("")
1364 .to_string(),
1365 prev_hash: metadata
1366 .get("prev_hash")
1367 .and_then(|v| v.as_str())
1368 .map(|s| s.to_string()),
1369 hash: metadata
1370 .get("hash")
1371 .and_then(|v| v.as_str())
1372 .unwrap_or("")
1373 .to_string(),
1374 agent_id: metadata
1375 .get("agent_id")
1376 .and_then(|v| v.as_str())
1377 .unwrap_or("")
1378 .to_string(),
1379 action_type: metadata
1380 .get("action_type")
1381 .and_then(|v| v.as_str())
1382 .unwrap_or("")
1383 .to_string(),
1384 details: metadata
1385 .get("details")
1386 .and_then(|v| v.as_str())
1387 .unwrap_or("")
1388 .to_string(),
1389 embedding: result.vector.unwrap_or_default(),
1390 timestamp: metadata
1391 .get("timestamp")
1392 .and_then(|v| v.as_i64())
1393 .unwrap_or(0),
1394 metadata: None,
1395 });
1396 }
1397 }
1398
1399 Ok(entries)
1400 }
1401
1402 pub fn get_by_agent(&self, agent_id: &str, k: usize) -> Result<Vec<WitnessEntry>> {
1404 self.search(agent_id, k)
1406 }
1407
1408 pub fn verify_chain(&self) -> Result<bool> {
1410 let entries = self.search("", 10000)?;
1411
1412 let mut sorted_entries = entries;
1414 sorted_entries.sort_by_key(|e| e.timestamp);
1415
1416 for i in 1..sorted_entries.len() {
1418 let prev = &sorted_entries[i - 1];
1419 let curr = &sorted_entries[i];
1420
1421 if let Some(ref prev_hash) = curr.prev_hash {
1422 if prev_hash != &prev.hash {
1423 return Ok(false);
1424 }
1425 }
1426 }
1427
1428 Ok(true)
1429 }
1430}
1431
1432impl AgenticDB {
1433 pub fn policy_memory(&self) -> PolicyMemoryStore<'_> {
1435 PolicyMemoryStore::new(self)
1436 }
1437
1438 pub fn session_index(&self, session_id: &str, ttl_seconds: i64) -> SessionStateIndex<'_> {
1440 SessionStateIndex::new(self, session_id, ttl_seconds)
1441 }
1442
1443 pub fn witness_log(&self) -> WitnessLog<'_> {
1445 WitnessLog::new(self)
1446 }
1447}
1448
1449#[cfg(test)]
1450mod tests {
1451 use super::*;
1452 use tempfile::tempdir;
1453
1454 fn create_test_db() -> Result<AgenticDB> {
1455 let dir = tempdir().unwrap();
1456 let mut options = DbOptions::default();
1457 options.storage_path = dir.path().join("test.db").to_string_lossy().to_string();
1458 options.dimensions = 128;
1459 AgenticDB::new(options)
1460 }
1461
1462 #[test]
1463 fn test_reflexion_episode() -> Result<()> {
1464 let db = create_test_db()?;
1465
1466 let id = db.store_episode(
1467 "Solve math problem".to_string(),
1468 vec!["read problem".to_string(), "calculate".to_string()],
1469 vec!["got 42".to_string()],
1470 "Should have shown work".to_string(),
1471 )?;
1472
1473 let episodes = db.retrieve_similar_episodes("math problem solving", 5)?;
1474 assert!(!episodes.is_empty());
1475 assert_eq!(episodes[0].id, id);
1476
1477 Ok(())
1478 }
1479
1480 #[test]
1481 fn test_update_q_value_preserves_entry() -> Result<()> {
1482 let db = create_test_db()?;
1485 let pm = db.policy_memory();
1486 let emb = vec![0.1_f32; 128];
1487
1488 let id = pm.store_policy("state_1", emb.clone(), "shoot", 1.0, 0.5)?;
1489
1490 pm.update_q_value(&id, 0.9)?;
1492
1493 let hits = pm.retrieve_similar_states(&emb, 5)?;
1495 assert!(
1496 !hits.is_empty(),
1497 "policy was deleted by update_q_value (#562 regression)"
1498 );
1499 assert!(
1500 hits.iter().any(|p| (p.action.q_value - 0.9).abs() < 1e-9),
1501 "q_value not updated to 0.9; got {:?}",
1502 hits.iter().map(|p| p.action.q_value).collect::<Vec<_>>()
1503 );
1504
1505 assert!(pm.update_q_value("does_not_exist", 1.0).is_err());
1507
1508 Ok(())
1509 }
1510
1511 #[test]
1512 fn test_skill_library() -> Result<()> {
1513 let db = create_test_db()?;
1514
1515 let mut params = HashMap::new();
1516 params.insert("input".to_string(), "string".to_string());
1517
1518 let skill_id = db.create_skill(
1519 "Parse JSON".to_string(),
1520 "Parse JSON from string".to_string(),
1521 params,
1522 vec!["json.parse()".to_string()],
1523 )?;
1524
1525 let skills = db.search_skills("parse json data", 5)?;
1526 assert!(!skills.is_empty());
1527
1528 Ok(())
1529 }
1530
1531 #[test]
1532 fn test_causal_edge() -> Result<()> {
1533 let db = create_test_db()?;
1534
1535 let edge_id = db.add_causal_edge(
1536 vec!["rain".to_string()],
1537 vec!["wet ground".to_string()],
1538 0.95,
1539 "Weather observation".to_string(),
1540 )?;
1541
1542 let results = db.query_with_utility("weather patterns", 5, 0.7, 0.2, 0.1)?;
1543 assert!(!results.is_empty());
1544
1545 Ok(())
1546 }
1547
1548 #[test]
1549 fn test_learning_session() -> Result<()> {
1550 let db = create_test_db()?;
1551
1552 let session_id = db.start_session("Q-Learning".to_string(), 4, 2)?;
1553
1554 db.add_experience(
1555 &session_id,
1556 vec![1.0, 0.0, 0.0, 0.0],
1557 vec![1.0, 0.0],
1558 1.0,
1559 vec![0.0, 1.0, 0.0, 0.0],
1560 false,
1561 )?;
1562
1563 let prediction = db.predict_with_confidence(&session_id, vec![1.0, 0.0, 0.0, 0.0])?;
1564 assert_eq!(prediction.action.len(), 2);
1565
1566 Ok(())
1567 }
1568
1569 #[test]
1570 fn test_auto_consolidate() -> Result<()> {
1571 let db = create_test_db()?;
1572
1573 let sequences = vec![
1574 vec![
1575 "step1".to_string(),
1576 "step2".to_string(),
1577 "step3".to_string(),
1578 ],
1579 vec![
1580 "action1".to_string(),
1581 "action2".to_string(),
1582 "action3".to_string(),
1583 ],
1584 ];
1585
1586 let skill_ids = db.auto_consolidate(sequences, 3)?;
1587 assert_eq!(skill_ids.len(), 2);
1588
1589 Ok(())
1590 }
1591}