#![allow(clippy::all)]
use crate::types::{
ClearFileIndexResult, GlobalMemoryRecord, GlobalMemorySearchHit, GlobalMemoryWriteResult,
KnowledgeCoverageRecord, KnowledgeItemRecord, KnowledgeItemStatus, KnowledgePromotionRequest,
KnowledgePromotionResult, KnowledgeSpaceRecord, MemoryChunk, MemoryConfig, MemoryError,
MemoryResult, MemoryStats, MemoryTier, ProjectMemoryStats, DEFAULT_EMBEDDING_DIMENSION,
};
use chrono::{DateTime, Utc};
use rusqlite::{ffi::sqlite3_auto_extension, params, Connection, OptionalExtension, Row};
use sqlite_vec::sqlite3_vec_init;
use std::collections::HashSet;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
type ProjectIndexStatusRow = (
Option<String>,
Option<i64>,
Option<i64>,
Option<i64>,
Option<i64>,
Option<i64>,
);
pub struct MemoryDatabase {
conn: Arc<Mutex<Connection>>,
db_path: std::path::PathBuf,
}
include!("memory_database_impl_parts/part01.rs");
include!("memory_database_impl_parts/part02.rs");
fn row_to_chunk(row: &Row, tier: MemoryTier) -> Result<MemoryChunk, rusqlite::Error> {
let id: String = row.get(0)?;
let content: String = row.get(1)?;
let (session_id, project_id, source_idx, created_at_idx, token_count_idx, metadata_idx) =
match tier {
MemoryTier::Session => (
Some(row.get(2)?),
row.get(3)?,
4usize,
5usize,
6usize,
7usize,
),
MemoryTier::Project => (
row.get(2)?,
Some(row.get(3)?),
4usize,
5usize,
6usize,
7usize,
),
MemoryTier::Global => (None, None, 2usize, 3usize, 4usize, 5usize),
};
let source: String = row.get(source_idx)?;
let created_at_str: String = row.get(created_at_idx)?;
let token_count: i64 = row.get(token_count_idx)?;
let metadata_str: Option<String> = row.get(metadata_idx)?;
let created_at = DateTime::parse_from_rfc3339(&created_at_str)
.map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(e))
})?
.with_timezone(&Utc);
let metadata = metadata_str
.filter(|s| !s.is_empty())
.and_then(|s| serde_json::from_str(&s).ok());
let source_path = row.get::<_, Option<String>>("source_path").ok().flatten();
let source_mtime = row.get::<_, Option<i64>>("source_mtime").ok().flatten();
let source_size = row.get::<_, Option<i64>>("source_size").ok().flatten();
let source_hash = row.get::<_, Option<String>>("source_hash").ok().flatten();
Ok(MemoryChunk {
id,
content,
tier,
session_id,
project_id,
source,
source_path,
source_mtime,
source_size,
source_hash,
created_at,
token_count,
metadata,
})
}
fn require_scope_id<'a>(tier: MemoryTier, scope: Option<&'a str>) -> MemoryResult<&'a str> {
scope
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| {
crate::types::MemoryError::InvalidConfig(match tier {
MemoryTier::Session => "tier=session requires session_id".to_string(),
MemoryTier::Project => "tier=project requires project_id".to_string(),
MemoryTier::Global => "tier=global does not require a scope id".to_string(),
})
})
}
fn row_to_global_record(row: &Row) -> Result<GlobalMemoryRecord, rusqlite::Error> {
let metadata_str: Option<String> = row.get(12)?;
let provenance_str: Option<String> = row.get(13)?;
Ok(GlobalMemoryRecord {
id: row.get(0)?,
user_id: row.get(1)?,
source_type: row.get(2)?,
content: row.get(3)?,
content_hash: row.get(4)?,
run_id: row.get(5)?,
session_id: row.get(6)?,
message_id: row.get(7)?,
tool_name: row.get(8)?,
project_tag: row.get(9)?,
channel_tag: row.get(10)?,
host_tag: row.get(11)?,
metadata: metadata_str
.filter(|s| !s.is_empty())
.and_then(|s| serde_json::from_str(&s).ok()),
provenance: provenance_str
.filter(|s| !s.is_empty())
.and_then(|s| serde_json::from_str(&s).ok()),
redaction_status: row.get(14)?,
redaction_count: row.get::<_, i64>(15)? as u32,
visibility: row.get(16)?,
demoted: row.get::<_, i64>(17)? != 0,
score_boost: row.get(18)?,
created_at_ms: row.get::<_, i64>(19)? as u64,
updated_at_ms: row.get::<_, i64>(20)? as u64,
expires_at_ms: row.get::<_, Option<i64>>(21)?.map(|v| v as u64),
})
}
impl MemoryDatabase {
pub async fn get_node_by_uri(
&self,
uri: &str,
) -> MemoryResult<Option<crate::types::MemoryNode>> {
let conn = self.conn.lock().await;
let mut stmt = conn.prepare(
"SELECT id, uri, parent_uri, node_type, created_at, updated_at, metadata
FROM memory_nodes WHERE uri = ?1",
)?;
let result = stmt.query_row(params![uri], |row| {
let node_type_str: String = row.get(3)?;
let node_type = node_type_str
.parse()
.unwrap_or(crate::types::NodeType::File);
let metadata_str: Option<String> = row.get(6)?;
Ok(crate::types::MemoryNode {
id: row.get(0)?,
uri: row.get(1)?,
parent_uri: row.get(2)?,
node_type,
created_at: row.get::<_, String>(4)?.parse().unwrap_or_default(),
updated_at: row.get::<_, String>(5)?.parse().unwrap_or_default(),
metadata: metadata_str.and_then(|s| serde_json::from_str(&s).ok()),
})
});
match result {
Ok(node) => Ok(Some(node)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(MemoryError::Database(e)),
}
}
pub async fn create_node(
&self,
uri: &str,
parent_uri: Option<&str>,
node_type: crate::types::NodeType,
metadata: Option<&serde_json::Value>,
) -> MemoryResult<String> {
let id = uuid::Uuid::new_v4().to_string();
let now = Utc::now().to_rfc3339();
let metadata_str = metadata.map(|m| serde_json::to_string(m)).transpose()?;
let conn = self.conn.lock().await;
conn.execute(
"INSERT INTO memory_nodes (id, uri, parent_uri, node_type, created_at, updated_at, metadata)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![id, uri, parent_uri, node_type.to_string(), now, now, metadata_str],
)?;
Ok(id)
}
pub async fn list_directory(&self, uri: &str) -> MemoryResult<Vec<crate::types::MemoryNode>> {
let conn = self.conn.lock().await;
let mut stmt = conn.prepare(
"SELECT id, uri, parent_uri, node_type, created_at, updated_at, metadata
FROM memory_nodes WHERE parent_uri = ?1 ORDER BY node_type DESC, uri ASC",
)?;
let rows = stmt.query_map(params![uri], |row| {
let node_type_str: String = row.get(3)?;
let node_type = node_type_str
.parse()
.unwrap_or(crate::types::NodeType::File);
let metadata_str: Option<String> = row.get(6)?;
Ok(crate::types::MemoryNode {
id: row.get(0)?,
uri: row.get(1)?,
parent_uri: row.get(2)?,
node_type,
created_at: row.get::<_, String>(4)?.parse().unwrap_or_default(),
updated_at: row.get::<_, String>(5)?.parse().unwrap_or_default(),
metadata: metadata_str.and_then(|s| serde_json::from_str(&s).ok()),
})
})?;
rows.collect::<Result<Vec<_>, _>>()
.map_err(MemoryError::Database)
}
pub async fn get_layer(
&self,
node_id: &str,
layer_type: crate::types::LayerType,
) -> MemoryResult<Option<crate::types::MemoryLayer>> {
let conn = self.conn.lock().await;
let mut stmt = conn.prepare(
"SELECT id, node_id, layer_type, content, token_count, embedding_id, created_at, source_chunk_id
FROM memory_layers WHERE node_id = ?1 AND layer_type = ?2"
)?;
let result = stmt.query_row(params![node_id, layer_type.to_string()], |row| {
let layer_type_str: String = row.get(2)?;
let layer_type = layer_type_str
.parse()
.unwrap_or(crate::types::LayerType::L2);
Ok(crate::types::MemoryLayer {
id: row.get(0)?,
node_id: row.get(1)?,
layer_type,
content: row.get(3)?,
token_count: row.get(4)?,
embedding_id: row.get(5)?,
created_at: row.get::<_, String>(6)?.parse().unwrap_or_default(),
source_chunk_id: row.get(7)?,
})
});
match result {
Ok(layer) => Ok(Some(layer)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(MemoryError::Database(e)),
}
}
pub async fn create_layer(
&self,
node_id: &str,
layer_type: crate::types::LayerType,
content: &str,
token_count: i64,
source_chunk_id: Option<&str>,
) -> MemoryResult<String> {
let id = uuid::Uuid::new_v4().to_string();
let now = Utc::now().to_rfc3339();
let conn = self.conn.lock().await;
conn.execute(
"INSERT INTO memory_layers (id, node_id, layer_type, content, token_count, created_at, source_chunk_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![id, node_id, layer_type.to_string(), content, token_count, now, source_chunk_id],
)?;
Ok(id)
}
pub async fn get_children_tree(
&self,
parent_uri: &str,
max_depth: usize,
) -> MemoryResult<Vec<crate::types::TreeNode>> {
if max_depth == 0 {
return Ok(Vec::new());
}
let children = self.list_directory(parent_uri).await?;
let mut tree_nodes = Vec::new();
for child in children {
let layer_summary = self.get_layer_summary(&child.id).await?;
let grandchildren = if child.node_type == crate::types::NodeType::Directory {
Box::pin(self.get_children_tree(&child.uri, max_depth.saturating_sub(1))).await?
} else {
Vec::new()
};
tree_nodes.push(crate::types::TreeNode {
node: child,
children: grandchildren,
layer_summary,
});
}
Ok(tree_nodes)
}
async fn get_layer_summary(
&self,
node_id: &str,
) -> MemoryResult<Option<crate::types::LayerSummary>> {
let l0 = self.get_layer(node_id, crate::types::LayerType::L0).await?;
let l1 = self.get_layer(node_id, crate::types::LayerType::L1).await?;
let has_l2 = self
.get_layer(node_id, crate::types::LayerType::L2)
.await?
.is_some();
if l0.is_none() && l1.is_none() && !has_l2 {
return Ok(None);
}
Ok(Some(crate::types::LayerSummary {
l0_preview: l0.map(|l| truncate_string(&l.content, 100)),
l1_preview: l1.map(|l| truncate_string(&l.content, 200)),
has_l2,
}))
}
pub async fn node_exists(&self, uri: &str) -> MemoryResult<bool> {
let conn = self.conn.lock().await;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM memory_nodes WHERE uri = ?1",
params![uri],
|row| row.get(0),
)?;
Ok(count > 0)
}
}
fn row_to_knowledge_space(row: &Row) -> Result<KnowledgeSpaceRecord, rusqlite::Error> {
let scope = row
.get::<_, String>(1)?
.parse()
.unwrap_or(tandem_orchestrator::KnowledgeScope::Project);
let trust_level = row
.get::<_, String>(6)?
.parse()
.unwrap_or(tandem_orchestrator::KnowledgeTrustLevel::Promoted);
let metadata = row
.get::<_, Option<String>>(7)?
.and_then(|raw| serde_json::from_str(&raw).ok());
Ok(KnowledgeSpaceRecord {
id: row.get(0)?,
scope,
project_id: row.get(2)?,
namespace: row.get(3)?,
title: row.get(4)?,
description: row.get(5)?,
trust_level,
metadata,
created_at_ms: row.get::<_, i64>(8)? as u64,
updated_at_ms: row.get::<_, i64>(9)? as u64,
})
}
fn row_to_knowledge_item(row: &Row) -> Result<KnowledgeItemRecord, rusqlite::Error> {
let trust_level = row
.get::<_, String>(8)?
.parse()
.unwrap_or(tandem_orchestrator::KnowledgeTrustLevel::Promoted);
let status = row
.get::<_, String>(9)?
.parse()
.unwrap_or(KnowledgeItemStatus::Working);
let payload = row
.get::<_, String>(7)
.ok()
.and_then(|raw| serde_json::from_str(&raw).ok())
.unwrap_or(serde_json::Value::Null);
let artifact_refs = row
.get::<_, String>(11)
.ok()
.and_then(|raw| serde_json::from_str(&raw).ok())
.unwrap_or_default();
let source_memory_ids = row
.get::<_, String>(12)
.ok()
.and_then(|raw| serde_json::from_str(&raw).ok())
.unwrap_or_default();
let metadata = row
.get::<_, Option<String>>(14)?
.and_then(|raw| serde_json::from_str(&raw).ok());
Ok(KnowledgeItemRecord {
id: row.get(0)?,
space_id: row.get(1)?,
coverage_key: row.get(2)?,
dedupe_key: row.get(3)?,
item_type: row.get(4)?,
title: row.get(5)?,
summary: row.get(6)?,
payload,
trust_level,
status,
run_id: row.get(10)?,
artifact_refs,
source_memory_ids,
freshness_expires_at_ms: row.get::<_, Option<i64>>(13)?.map(|value| value as u64),
metadata,
created_at_ms: row.get::<_, i64>(15)? as u64,
updated_at_ms: row.get::<_, i64>(16)? as u64,
})
}
fn row_to_knowledge_coverage(row: &Row) -> Result<KnowledgeCoverageRecord, rusqlite::Error> {
let metadata = row
.get::<_, Option<String>>(7)?
.and_then(|raw| serde_json::from_str(&raw).ok());
Ok(KnowledgeCoverageRecord {
coverage_key: row.get(0)?,
space_id: row.get(1)?,
latest_item_id: row.get(2)?,
latest_dedupe_key: row.get(3)?,
last_seen_at_ms: row.get::<_, i64>(4)? as u64,
last_promoted_at_ms: row.get::<_, Option<i64>>(5)?.map(|value| value as u64),
freshness_expires_at_ms: row.get::<_, Option<i64>>(6)?.map(|value| value as u64),
metadata,
})
}
fn truncate_string(s: &str, max_len: usize) -> String {
if s.len() <= max_len {
s.to_string()
} else {
format!("{}...", &s[..max_len.saturating_sub(3)])
}
}
fn build_fts_query(query: &str) -> String {
let tokens = query
.split_whitespace()
.filter_map(|tok| {
let cleaned =
tok.trim_matches(|c: char| !c.is_ascii_alphanumeric() && c != '_' && c != '-');
if cleaned.is_empty() {
None
} else {
Some(format!("\"{}\"", cleaned))
}
})
.collect::<Vec<_>>();
if tokens.is_empty() {
"\"\"".to_string()
} else {
tokens.join(" OR ")
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::Value;
use tandem_orchestrator::{KnowledgeScope, KnowledgeTrustLevel};
use tempfile::TempDir;
async fn setup_test_db() -> (MemoryDatabase, TempDir) {
let temp_dir = TempDir::new().unwrap();
let db_path = temp_dir.path().join("test_memory.db");
let db = MemoryDatabase::new(&db_path).await.unwrap();
(db, temp_dir)
}
#[tokio::test]
async fn test_init_schema() {
let (db, _temp) = setup_test_db().await;
let stats = db.get_stats().await.unwrap();
assert_eq!(stats.total_chunks, 0);
}
#[tokio::test]
async fn test_knowledge_registry_roundtrip() {
let (db, _temp) = setup_test_db().await;
let space = KnowledgeSpaceRecord {
id: "space-1".to_string(),
scope: tandem_orchestrator::KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("support".to_string()),
title: Some("Support Knowledge".to_string()),
description: Some("Reusable support guidance".to_string()),
trust_level: tandem_orchestrator::KnowledgeTrustLevel::Promoted,
metadata: Some(serde_json::json!({"owner": "ops"})),
created_at_ms: 1,
updated_at_ms: 2,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-1".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1/support/debugging/slow-start".to_string(),
dedupe_key: "dedupe-1".to_string(),
item_type: "decision".to_string(),
title: "Restart service before retry".to_string(),
summary: Some("When the service is stale, restart before retrying.".to_string()),
payload: serde_json::json!({"action": "restart"}),
trust_level: tandem_orchestrator::KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-1".to_string()),
artifact_refs: vec!["artifact://run-1/report".to_string()],
source_memory_ids: vec!["memory-1".to_string()],
freshness_expires_at_ms: Some(10),
metadata: Some(serde_json::json!({"source": "run"})),
created_at_ms: 3,
updated_at_ms: 4,
};
db.upsert_knowledge_item(&item).await.unwrap();
let coverage = KnowledgeCoverageRecord {
coverage_key: item.coverage_key.clone(),
space_id: space.id.clone(),
latest_item_id: Some(item.id.clone()),
latest_dedupe_key: Some(item.dedupe_key.clone()),
last_seen_at_ms: 5,
last_promoted_at_ms: Some(6),
freshness_expires_at_ms: Some(10),
metadata: Some(serde_json::json!({"coverage": true})),
};
db.upsert_knowledge_coverage(&coverage).await.unwrap();
let loaded_space = db.get_knowledge_space(&space.id).await.unwrap().unwrap();
assert_eq!(loaded_space.namespace.as_deref(), Some("support"));
let loaded_items = db
.list_knowledge_items(&space.id, Some(&item.coverage_key))
.await
.unwrap();
assert_eq!(loaded_items.len(), 1);
assert_eq!(loaded_items[0].title, item.title);
let loaded_coverage = db
.get_knowledge_coverage(&item.coverage_key, &space.id)
.await
.unwrap()
.unwrap();
assert_eq!(loaded_coverage.latest_item_id.as_deref(), Some("item-1"));
}
#[tokio::test]
async fn test_store_and_retrieve_chunk() {
let (db, _temp) = setup_test_db().await;
let chunk = MemoryChunk {
id: "test-1".to_string(),
content: "Test content".to_string(),
tier: MemoryTier::Session,
session_id: Some("session-1".to_string()),
project_id: Some("project-1".to_string()),
source: "user_message".to_string(),
source_path: None,
source_mtime: None,
source_size: None,
source_hash: None,
created_at: Utc::now(),
token_count: 10,
metadata: None,
};
let embedding = vec![0.1f32; DEFAULT_EMBEDDING_DIMENSION];
db.store_chunk(&chunk, &embedding).await.unwrap();
let chunks = db.get_session_chunks("session-1").await.unwrap();
assert_eq!(chunks.len(), 1);
assert_eq!(chunks[0].content, "Test content");
}
#[tokio::test]
async fn test_store_and_retrieve_global_chunk() {
let (db, _temp) = setup_test_db().await;
let chunk = MemoryChunk {
id: "global-1".to_string(),
content: "Global note".to_string(),
tier: MemoryTier::Global,
session_id: None,
project_id: None,
source: "agent_note".to_string(),
source_path: None,
source_mtime: None,
source_size: None,
source_hash: None,
created_at: Utc::now(),
token_count: 7,
metadata: Some(serde_json::json!({"kind":"test"})),
};
let embedding = vec![0.2f32; DEFAULT_EMBEDDING_DIMENSION];
db.store_chunk(&chunk, &embedding).await.unwrap();
let chunks = db.get_global_chunks(10).await.unwrap();
assert_eq!(chunks.len(), 1);
assert_eq!(chunks[0].content, "Global note");
assert_eq!(chunks[0].source, "agent_note");
assert_eq!(chunks[0].token_count, 7);
assert_eq!(chunks[0].tier, MemoryTier::Global);
}
#[tokio::test]
async fn test_global_chunk_exists_by_source_hash() {
let (db, _temp) = setup_test_db().await;
let chunk = MemoryChunk {
id: "global-hash".to_string(),
content: "Global hash note".to_string(),
tier: MemoryTier::Global,
session_id: None,
project_id: None,
source: "chat_exchange".to_string(),
source_path: None,
source_mtime: None,
source_size: None,
source_hash: Some("hash-123".to_string()),
created_at: Utc::now(),
token_count: 5,
metadata: None,
};
let embedding = vec![0.3f32; DEFAULT_EMBEDDING_DIMENSION];
db.store_chunk(&chunk, &embedding).await.unwrap();
assert!(db
.global_chunk_exists_by_source_hash("hash-123")
.await
.unwrap());
assert!(!db
.global_chunk_exists_by_source_hash("missing-hash")
.await
.unwrap());
}
#[tokio::test]
async fn test_config_crud() {
let (db, _temp) = setup_test_db().await;
let config = db.get_or_create_config("project-1").await.unwrap();
assert_eq!(config.max_chunks, 10000);
let new_config = MemoryConfig {
max_chunks: 5000,
..Default::default()
};
db.update_config("project-1", &new_config).await.unwrap();
let updated = db.get_or_create_config("project-1").await.unwrap();
assert_eq!(updated.max_chunks, 5000);
}
#[tokio::test]
async fn test_global_memory_put_search_and_dedup() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let record = GlobalMemoryRecord {
id: "gm-1".to_string(),
user_id: "user-a".to_string(),
source_type: "user_message".to_string(),
content: "remember rust workspace layout".to_string(),
content_hash: "h1".to_string(),
run_id: "run-1".to_string(),
session_id: Some("s1".to_string()),
message_id: Some("m1".to_string()),
tool_name: None,
project_tag: Some("proj-x".to_string()),
channel_tag: None,
host_tag: None,
metadata: None,
provenance: None,
redaction_status: "passed".to_string(),
redaction_count: 0,
visibility: "private".to_string(),
demoted: false,
score_boost: 0.0,
created_at_ms: now,
updated_at_ms: now,
expires_at_ms: None,
};
let first = db.put_global_memory_record(&record).await.unwrap();
assert!(first.stored);
let second = db.put_global_memory_record(&record).await.unwrap();
assert!(second.deduped);
let hits = db
.search_global_memory("user-a", "rust workspace", 5, Some("proj-x"), None, None)
.await
.unwrap();
assert!(!hits.is_empty());
assert_eq!(hits[0].record.id, "gm-1");
}
#[tokio::test]
async fn test_knowledge_registry_round_trip() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-1".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("marketing/positioning".to_string()),
title: Some("Marketing positioning".to_string()),
description: Some("Reusable positioning guidance".to_string()),
trust_level: KnowledgeTrustLevel::ApprovedDefault,
metadata: Some(serde_json::json!({"owner":"marketing"})),
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let loaded_space = db.get_knowledge_space("space-1").await.unwrap().unwrap();
assert_eq!(loaded_space.id, "space-1");
assert_eq!(loaded_space.scope, KnowledgeScope::Project);
assert_eq!(loaded_space.project_id.as_deref(), Some("project-1"));
assert_eq!(
loaded_space.namespace.as_deref(),
Some("marketing/positioning")
);
let item = KnowledgeItemRecord {
id: "item-1".to_string(),
space_id: "space-1".to_string(),
coverage_key: "project-1::marketing/positioning::strategy::pricing".to_string(),
dedupe_key: "item-1-dedupe".to_string(),
item_type: "evidence".to_string(),
title: "Pricing sensitivity observation".to_string(),
summary: Some("Customers reacted to annual pricing changes".to_string()),
payload: serde_json::json!({"claim":"Annual pricing changes created friction"}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-1".to_string()),
artifact_refs: vec!["artifact://run-1/research-sources".to_string()],
source_memory_ids: vec!["memory-1".to_string()],
freshness_expires_at_ms: Some(now + 86_400_000),
metadata: Some(serde_json::json!({"source_kind":"web"})),
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let loaded_item = db.get_knowledge_item("item-1").await.unwrap().unwrap();
assert_eq!(loaded_item.id, "item-1");
assert_eq!(loaded_item.space_id, "space-1");
assert_eq!(
loaded_item.coverage_key,
"project-1::marketing/positioning::strategy::pricing"
);
assert_eq!(loaded_item.status, KnowledgeItemStatus::Promoted);
assert_eq!(
loaded_item.artifact_refs,
vec!["artifact://run-1/research-sources".to_string()]
);
let by_space = db.list_knowledge_items("space-1", None).await.unwrap();
assert_eq!(by_space.len(), 1);
let by_coverage = db
.list_knowledge_items(
"space-1",
Some("project-1::marketing/positioning::strategy::pricing"),
)
.await
.unwrap();
assert_eq!(by_coverage.len(), 1);
let coverage = KnowledgeCoverageRecord {
coverage_key: "project-1::marketing/positioning::strategy::pricing".to_string(),
space_id: "space-1".to_string(),
latest_item_id: Some("item-1".to_string()),
latest_dedupe_key: Some("item-1-dedupe".to_string()),
last_seen_at_ms: now,
last_promoted_at_ms: Some(now),
freshness_expires_at_ms: Some(now + 86_400_000),
metadata: Some(serde_json::json!({"reuse_reason":"same topic"})),
};
db.upsert_knowledge_coverage(&coverage).await.unwrap();
let loaded_coverage = db
.get_knowledge_coverage(
"project-1::marketing/positioning::strategy::pricing",
"space-1",
)
.await
.unwrap()
.unwrap();
assert_eq!(loaded_coverage.space_id, "space-1");
assert_eq!(loaded_coverage.latest_item_id.as_deref(), Some("item-1"));
assert_eq!(
loaded_coverage.latest_dedupe_key.as_deref(),
Some("item-1-dedupe")
);
}
#[tokio::test]
async fn test_knowledge_promotion_working_to_promoted_updates_coverage() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote-1".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("engineering/debugging".to_string()),
title: Some("Engineering debugging".to_string()),
description: Some("Reusable debugging guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote-1".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::engineering/debugging::startup::race".to_string(),
dedupe_key: "dedupe-promote-1".to_string(),
item_type: "decision".to_string(),
title: "Delay startup-dependent retries".to_string(),
summary: Some("Retry only after startup completed.".to_string()),
payload: serde_json::json!({"action":"delay_retry"}),
trust_level: KnowledgeTrustLevel::Working,
status: KnowledgeItemStatus::Working,
run_id: Some("run-1".to_string()),
artifact_refs: vec!["artifact://run-1/debug".to_string()],
source_memory_ids: vec!["memory-1".to_string()],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let promote = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::Promoted,
promoted_at_ms: now + 10,
freshness_expires_at_ms: Some(now + 86_400_000),
reviewer_id: None,
approval_id: None,
reason: Some("validated in workflow".to_string()),
};
let result = db.promote_knowledge_item(&promote).await.unwrap().unwrap();
assert!(result.promoted);
assert_eq!(result.item.status, KnowledgeItemStatus::Promoted);
assert_eq!(result.item.trust_level, KnowledgeTrustLevel::Promoted);
assert_eq!(
result.coverage.latest_item_id.as_deref(),
Some("item-promote-1")
);
assert_eq!(
result.coverage.latest_dedupe_key.as_deref(),
Some("dedupe-promote-1")
);
assert_eq!(result.coverage.last_promoted_at_ms, Some(now + 10));
}
#[tokio::test]
async fn test_knowledge_promotion_promoted_to_approved_default_requires_review() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote-2".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("marketing/positioning".to_string()),
title: Some("Marketing positioning".to_string()),
description: Some("Reusable positioning guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote-2".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::marketing/positioning::strategy::pricing".to_string(),
dedupe_key: "dedupe-promote-2".to_string(),
item_type: "evidence".to_string(),
title: "Pricing observation".to_string(),
summary: Some("Annual pricing changes created friction".to_string()),
payload: serde_json::json!({"claim":"pricing friction"}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-2".to_string()),
artifact_refs: vec!["artifact://run-2/research".to_string()],
source_memory_ids: vec!["memory-2".to_string()],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let promote = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::ApprovedDefault,
promoted_at_ms: now + 5,
freshness_expires_at_ms: None,
reviewer_id: None,
approval_id: None,
reason: Some("should require review".to_string()),
};
let err = db.promote_knowledge_item(&promote).await.unwrap_err();
match err {
MemoryError::InvalidConfig(_) => {}
other => panic!("unexpected error: {}", other),
}
}
#[tokio::test]
async fn test_knowledge_promotion_promoted_to_approved_default_updates_coverage() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote-3".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("support/runbooks".to_string()),
title: Some("Support runbooks".to_string()),
description: Some("Reusable runbook guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote-3".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::support/runbooks::oncall::restart".to_string(),
dedupe_key: "dedupe-promote-3".to_string(),
item_type: "runbook".to_string(),
title: "Restart service and verify".to_string(),
summary: Some("Restart then validate health endpoint.".to_string()),
payload: serde_json::json!({"steps":["restart","healthcheck"]}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-3".to_string()),
artifact_refs: vec!["artifact://run-3/runbook".to_string()],
source_memory_ids: vec!["memory-3".to_string()],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let promote = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::ApprovedDefault,
promoted_at_ms: now + 12,
freshness_expires_at_ms: Some(now + 172_800_000),
reviewer_id: Some("reviewer-1".to_string()),
approval_id: Some("approval-1".to_string()),
reason: Some("reviewed by ops".to_string()),
};
let result = db.promote_knowledge_item(&promote).await.unwrap().unwrap();
assert!(result.promoted);
assert_eq!(result.item.status, KnowledgeItemStatus::ApprovedDefault);
assert_eq!(
result.item.trust_level,
KnowledgeTrustLevel::ApprovedDefault
);
assert_eq!(result.coverage.last_promoted_at_ms, Some(now + 12));
assert_eq!(
result.coverage.latest_item_id.as_deref(),
Some("item-promote-3")
);
}
#[tokio::test]
async fn test_knowledge_promotion_rejects_deprecated() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote-4".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("ops".to_string()),
title: Some("Ops knowledge".to_string()),
description: Some("Reusable ops knowledge".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote-4".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::ops::incident::latency".to_string(),
dedupe_key: "dedupe-promote-4".to_string(),
item_type: "decision".to_string(),
title: "Ignore deprecated item".to_string(),
summary: None,
payload: serde_json::json!({"decision":"deprecated"}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Deprecated,
run_id: Some("run-4".to_string()),
artifact_refs: vec![],
source_memory_ids: vec![],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let promote = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::Promoted,
promoted_at_ms: now + 1,
freshness_expires_at_ms: None,
reviewer_id: None,
approval_id: None,
reason: None,
};
let err = db.promote_knowledge_item(&promote).await.unwrap_err();
match err {
MemoryError::InvalidConfig(_) => {}
other => panic!("unexpected error: {}", other),
}
}
#[tokio::test]
async fn test_knowledge_promotion_idempotent_updates_coverage() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote-5".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("engineering/ops".to_string()),
title: Some("Engineering ops".to_string()),
description: None,
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote-5".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::engineering/ops::deploy::guardrails".to_string(),
dedupe_key: "dedupe-promote-5".to_string(),
item_type: "pattern".to_string(),
title: "Deploy guardrails".to_string(),
summary: None,
payload: serde_json::json!({"pattern":"guardrails"}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-5".to_string()),
artifact_refs: vec![],
source_memory_ids: vec![],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let promote = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::Promoted,
promoted_at_ms: now + 20,
freshness_expires_at_ms: None,
reviewer_id: None,
approval_id: None,
reason: None,
};
let result = db.promote_knowledge_item(&promote).await.unwrap().unwrap();
assert!(!result.promoted);
assert_eq!(result.coverage.last_promoted_at_ms, Some(now + 20));
assert_eq!(
result.coverage.latest_item_id.as_deref(),
Some("item-promote-5")
);
}
#[tokio::test]
async fn test_knowledge_item_promotion_updates_coverage() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-promote".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("engineering/debugging".to_string()),
title: Some("Engineering debugging".to_string()),
description: Some("Reusable debugging guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-promote".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::engineering/debugging::startup::race".to_string(),
dedupe_key: "dedupe-promote".to_string(),
item_type: "decision".to_string(),
title: "Delay startup-dependent retries".to_string(),
summary: Some("Retry only after startup completes.".to_string()),
payload: serde_json::json!({"action": "delay_retry"}),
trust_level: KnowledgeTrustLevel::Working,
status: KnowledgeItemStatus::Working,
run_id: Some("run-promote".to_string()),
artifact_refs: vec!["artifact://run-promote/report".to_string()],
source_memory_ids: vec!["memory-promote".to_string()],
freshness_expires_at_ms: None,
metadata: Some(serde_json::json!({"source_kind":"run"})),
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let request = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::Promoted,
promoted_at_ms: now + 10,
freshness_expires_at_ms: Some(now + 86_400_000),
reviewer_id: None,
approval_id: None,
reason: Some("validated".to_string()),
};
let promoted = db
.promote_knowledge_item(&request)
.await
.unwrap()
.expect("promotion result");
assert_eq!(promoted.previous_status, KnowledgeItemStatus::Working);
assert!(promoted.promoted);
assert_eq!(promoted.item.status, KnowledgeItemStatus::Promoted);
assert_eq!(promoted.item.trust_level, KnowledgeTrustLevel::Promoted);
assert_eq!(
promoted.item.freshness_expires_at_ms,
Some(now + 86_400_000)
);
assert_eq!(
promoted
.item
.metadata
.as_ref()
.and_then(|value| value.get("promotion"))
.and_then(|value| value.get("to_status"))
.and_then(Value::as_str),
Some("promoted")
);
assert_eq!(
promoted.coverage.latest_item_id.as_deref(),
Some("item-promote")
);
assert_eq!(
promoted.coverage.latest_dedupe_key.as_deref(),
Some("dedupe-promote")
);
assert_eq!(promoted.coverage.last_promoted_at_ms, Some(now + 10));
assert_eq!(
promoted.coverage.freshness_expires_at_ms,
Some(now + 86_400_000)
);
let loaded = db
.get_knowledge_item("item-promote")
.await
.unwrap()
.unwrap();
assert_eq!(loaded.status, KnowledgeItemStatus::Promoted);
assert_eq!(
loaded
.metadata
.as_ref()
.and_then(|value| value.get("promotion"))
.and_then(|value| value.get("from_status"))
.and_then(Value::as_str),
Some("working")
);
}
#[tokio::test]
async fn test_knowledge_item_approved_default_requires_review() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-approved".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("marketing/positioning".to_string()),
title: Some("Marketing positioning".to_string()),
description: Some("Reusable positioning guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-approved".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::marketing/positioning::strategy::pricing".to_string(),
dedupe_key: "dedupe-approved".to_string(),
item_type: "evidence".to_string(),
title: "Pricing sensitivity observation".to_string(),
summary: Some("Customers reacted to annual pricing changes".to_string()),
payload: serde_json::json!({"claim":"Annual pricing changes created friction"}),
trust_level: KnowledgeTrustLevel::Promoted,
status: KnowledgeItemStatus::Promoted,
run_id: Some("run-approved".to_string()),
artifact_refs: vec!["artifact://run-approved/research".to_string()],
source_memory_ids: vec!["memory-approved".to_string()],
freshness_expires_at_ms: Some(now + 1234),
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let request = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::ApprovedDefault,
promoted_at_ms: now + 20,
freshness_expires_at_ms: Some(now + 90_000_000),
reviewer_id: Some("reviewer-1".to_string()),
approval_id: Some("approval-1".to_string()),
reason: Some("approved as default guidance".to_string()),
};
let promoted = db
.promote_knowledge_item(&request)
.await
.unwrap()
.expect("promotion result");
assert_eq!(promoted.previous_status, KnowledgeItemStatus::Promoted);
assert_eq!(promoted.item.status, KnowledgeItemStatus::ApprovedDefault);
assert_eq!(
promoted.item.trust_level,
KnowledgeTrustLevel::ApprovedDefault
);
assert_eq!(promoted.coverage.last_promoted_at_ms, Some(now + 20));
assert_eq!(
promoted
.item
.metadata
.as_ref()
.and_then(|value| value.get("promotion"))
.and_then(|value| value.get("approval_id"))
.and_then(Value::as_str),
Some("approval-1")
);
}
#[tokio::test]
async fn test_knowledge_item_promotion_rejects_invalid_transition() {
let (db, _temp) = setup_test_db().await;
let now = chrono::Utc::now().timestamp_millis() as u64;
let space = KnowledgeSpaceRecord {
id: "space-invalid".to_string(),
scope: KnowledgeScope::Project,
project_id: Some("project-1".to_string()),
namespace: Some("support".to_string()),
title: Some("Support".to_string()),
description: Some("Support guidance".to_string()),
trust_level: KnowledgeTrustLevel::Promoted,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_space(&space).await.unwrap();
let item = KnowledgeItemRecord {
id: "item-invalid".to_string(),
space_id: space.id.clone(),
coverage_key: "project-1::support::workflow::triage".to_string(),
dedupe_key: "dedupe-invalid".to_string(),
item_type: "decision".to_string(),
title: "Triage first".to_string(),
summary: None,
payload: serde_json::json!({"action":"triage"}),
trust_level: KnowledgeTrustLevel::Working,
status: KnowledgeItemStatus::Working,
run_id: Some("run-invalid".to_string()),
artifact_refs: vec![],
source_memory_ids: vec![],
freshness_expires_at_ms: None,
metadata: None,
created_at_ms: now,
updated_at_ms: now,
};
db.upsert_knowledge_item(&item).await.unwrap();
let request = KnowledgePromotionRequest {
item_id: item.id.clone(),
target_status: KnowledgeItemStatus::ApprovedDefault,
promoted_at_ms: now + 1,
freshness_expires_at_ms: None,
reviewer_id: Some("reviewer-1".to_string()),
approval_id: Some("approval-1".to_string()),
reason: Some("should fail".to_string()),
};
let err = db.promote_knowledge_item(&request).await.unwrap_err();
assert!(matches!(err, MemoryError::InvalidConfig(_)));
let loaded = db.get_knowledge_item(&item.id).await.unwrap().unwrap();
assert_eq!(loaded.status, KnowledgeItemStatus::Working);
}
}