diff --git a/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json b/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json index 8cbf0c7..e427fbe 100644 --- a/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json +++ b/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", "describe": { "columns": [ { diff --git a/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json b/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json index 6a40d9c..96f4c16 100644 --- a/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json +++ b/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", "describe": { "columns": [ { diff --git a/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json b/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json index 61e13f9..46deac2 100644 --- a/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json +++ b/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", "describe": { "columns": [ { diff --git a/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json b/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json index 9fe298e..587946b 100644 --- a/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json +++ b/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND version_num = $2\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND version_num = $2\n ", "describe": { "columns": [ { diff --git a/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json b/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json index c4cca32..bccd73a 100644 --- a/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json +++ b/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", "describe": { "columns": [ { diff --git a/crates/mem-cli/src/compaction.rs b/crates/mem-cli/src/compaction.rs index 328ede1..6229aed 100644 --- a/crates/mem-cli/src/compaction.rs +++ b/crates/mem-cli/src/compaction.rs @@ -58,7 +58,7 @@ impl Tier1Compactor { let rows = sqlx::query( r#" SELECT array_agg(id ORDER BY created_at) - FROM memory_edge + FROM knowledge_edge WHERE deleted_at IS NULL GROUP BY source_id, target_id, relation_type, md5(fact) HAVING COUNT(*) > 1 @@ -91,7 +91,7 @@ impl Tier1Compactor { let execute_fn = async move { for (_master, duplicate) in duplicates { sqlx::query( - "UPDATE memory_edge SET deleted_at = NOW() WHERE id = $1" + "UPDATE knowledge_edge SET deleted_at = NOW() WHERE id = $1" ) .bind(&duplicate) .execute(&pool) @@ -126,7 +126,7 @@ impl Tier1Compactor { let row_count: (i64,) = sqlx::query_as( &format!( r#" - SELECT COUNT(*) FROM memory_edge + SELECT COUNT(*) FROM knowledge_edge WHERE fact_invalid_at IS NOT NULL AND fact_invalid_at < {} AND deleted_at IS NULL @@ -149,7 +149,7 @@ impl Tier1Compactor { sqlx::query( &format!( r#" - UPDATE memory_edge + UPDATE knowledge_edge SET deleted_at = NOW() WHERE fact_invalid_at IS NOT NULL AND fact_invalid_at < {} @@ -207,8 +207,8 @@ impl Tier2Compactor { let rows = sqlx::query( r#" SELECT a.id, b.id, a.fact, b.fact - FROM memory_edge a - JOIN memory_edge b ON a.source_id = b.source_id + FROM knowledge_edge a + JOIN knowledge_edge b ON a.source_id = b.source_id AND a.target_id = b.target_id AND a.relation_type = b.relation_type AND a.id < b.id @@ -279,7 +279,7 @@ Respond with JSON: {{"confidence": 0.0-1.0}} where 1.0 means identical meaning." let edge_id = edge_b_id.to_string(); let execute_fn = async move { sqlx::query( - "UPDATE memory_edge SET deleted_at = NOW() WHERE id = $1" + "UPDATE knowledge_edge SET deleted_at = NOW() WHERE id = $1" ) .bind(&edge_id) .execute(&pool) diff --git a/crates/mem-cli/src/handlers/agent_handler.rs b/crates/mem-cli/src/handlers/agent_handler.rs index ceb1e52..59baf27 100644 --- a/crates/mem-cli/src/handlers/agent_handler.rs +++ b/crates/mem-cli/src/handlers/agent_handler.rs @@ -158,7 +158,7 @@ pub async fn register_agent_handler( // Temporal activities will: // 1. Persist agent state to temporal_workflow_links table // 2. Execute LLMInferenceActivity (call LLM via api.riotpiao.com/v1/chat/completions) - // 3. Store reasoning traces to memory_entity/memory_edge + // 3. Store reasoning traces to knowledge_node/knowledge_edge if let Some(jwt) = crate::handlers::extract_jwt_token(&req) { let client = SynthesisClient::new( "https://api.riotpiao.com".to_string(), diff --git a/crates/mem-cli/src/handlers/rebuild_handler.rs b/crates/mem-cli/src/handlers/rebuild_handler.rs index 96fa834..9fade2d 100644 --- a/crates/mem-cli/src/handlers/rebuild_handler.rs +++ b/crates/mem-cli/src/handlers/rebuild_handler.rs @@ -199,7 +199,7 @@ async fn compute_state_checksum(pool: &PgPool, project: &str) -> Result = sqlx::query_as::<_, IdRow>( - "SELECT id FROM memory_entity WHERE project_id = $1 ORDER BY id" + "SELECT id FROM knowledge_node WHERE project_id = $1 ORDER BY id" ) .bind(project) .fetch_all(pool) @@ -211,7 +211,7 @@ async fn compute_state_checksum(pool: &PgPool, project: &str) -> Result = sqlx::query_as::<_, IdRow>( - "SELECT id FROM memory_edge WHERE project_id = $1 ORDER BY id" + "SELECT id FROM knowledge_edge WHERE project_id = $1 ORDER BY id" ) .bind(project) .fetch_all(pool) diff --git a/crates/mem-cli/src/handlers/unified_synthesis.rs b/crates/mem-cli/src/handlers/unified_synthesis.rs index 96940f9..d7a93cd 100644 --- a/crates/mem-cli/src/handlers/unified_synthesis.rs +++ b/crates/mem-cli/src/handlers/unified_synthesis.rs @@ -200,7 +200,7 @@ pub async fn unified_synthesis_handler( } // Reasoning via Temporal workflow - // Temporal activity calls LLMInferenceActivity + persists results to memory_entity/memory_edge + // Temporal activity calls LLMInferenceActivity + persists results to knowledge_node/knowledge_edge if body.reason_query { reasoning = match execute_reasoning_workflow( &synthesis_client, diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index c1ddcc0..6b3b7e6 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -383,11 +383,11 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res loop { interval.tick().await; // O5: Table row counts - if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM memory_entity") + if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM knowledge_node") .fetch_one(&stats_pool).await { crate::metrics::DB_TABLE_ENTITY_ROWS.set(row.0 as u64); } - if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM memory_edge") + if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM knowledge_edge") .fetch_one(&stats_pool).await { crate::metrics::DB_TABLE_EDGE_ROWS.set(row.0 as u64); } @@ -1454,7 +1454,7 @@ async fn query_temporal_graph( ) -> anyhow::Result { // Step 1: Find entities (order by name for deterministic results) let entities_rows: Vec<(String, String, String)> = sqlx::query_as( - "SELECT id, name, entity_type FROM memory_entity WHERE project_id = $1 LIMIT $2" + "SELECT id::TEXT, name, entity_type FROM knowledge_node WHERE project_id = $1 LIMIT $2" ) .bind(¶ms.project) .bind(params.limit as i32) @@ -1470,7 +1470,7 @@ async fn query_temporal_graph( for (entity_id, _name, _type_str) in &entities_rows { let entity_edges: Vec<(String, String, String, String, f32, Option>, Option>)> = sqlx::query_as( - "SELECT id, target_entity_id, relation_type, fact, confidence, t_valid, t_invalid FROM memory_edge WHERE project_id = $1 AND source_entity_id = $2" + "SELECT id::TEXT, target_id::TEXT, relation_type, fact, confidence, t_valid, t_invalid FROM knowledge_edge WHERE project_id = $1 AND source_id = $2::UUID" ) .bind(¶ms.project) .bind(entity_id) diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index b8884dc..be1dee7 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/crates/mem-cli/src/ingest_worker.rs @@ -384,14 +384,14 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) -> let t_created_str = entity.t_created.to_string(); sqlx::query( - "INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence) - VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8) + "INSERT INTO knowledge_node (id, project_id, name, entity_type, description, t_created, t_updated, confidence) + VALUES ($1::UUID, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8) ON CONFLICT (project_id, name) DO UPDATE SET entity_type = EXCLUDED.entity_type, - description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description), + description = COALESCE(NULLIF(EXCLUDED.description, ''), knowledge_node.description), t_updated = NOW(), - confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence), - source_count = memory_entity.source_count + 1" + confidence = GREATEST(knowledge_node.confidence, EXCLUDED.confidence), + source_count = knowledge_node.source_count + 1" ) .bind(&entity.id) .bind(&entity.project_id) @@ -445,8 +445,8 @@ async fn save_edge_with_logging( async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> { // Try temporal schema first (id, project_id, source_entity_id, etc) let result = sqlx::query( - "INSERT INTO memory_edge (id, project_id, source_id, target_id, relation_type, fact, t_valid, t_invalid, t_created, confidence) - VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10) + "INSERT INTO knowledge_edge (id, project_id, source_id, target_id, relation_type, fact, t_valid, t_invalid, t_created, confidence) + VALUES ($1::UUID, $2, $3::UUID, $4::UUID, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10) ON CONFLICT (id) DO NOTHING" ) .bind(&edge.id) diff --git a/crates/mem-cli/src/metrics.rs b/crates/mem-cli/src/metrics.rs index 61a872d..7cbccb1 100644 --- a/crates/mem-cli/src/metrics.rs +++ b/crates/mem-cli/src/metrics.rs @@ -274,9 +274,9 @@ pub static WRITE_BYTES_TOTAL: Counter = Counter::new( "memory_write_bytes_total", "Total bytes written to storage"); pub static DB_ENTITY_COUNT: Gauge = Gauge::new( - "memory_db_entity_count", "Current entity count in memory_entity table"); + "memory_db_entity_count", "Current entity count in knowledge_node table"); pub static DB_EDGE_COUNT: Gauge = Gauge::new( - "memory_db_edge_count", "Current edge count in memory_edge table"); + "memory_db_edge_count", "Current edge count in knowledge_edge table"); pub static DB_CHUNK_COUNT: Gauge = Gauge::new( "memory_db_chunk_count", "Current chunk count in memory_chunks table"); @@ -436,9 +436,9 @@ pub static DB_TRANSACTION_DURATION: Lazy = Lazy::new(|| // Table-specific row counts (updated periodically) pub static DB_TABLE_ENTITY_ROWS: Gauge = Gauge::new( - "memory_db_table_entity_rows", "Rows in memory_entity table"); + "memory_db_table_entity_rows", "Rows in knowledge_node table"); pub static DB_TABLE_EDGE_ROWS: Gauge = Gauge::new( - "memory_db_table_edge_rows", "Rows in memory_edge table"); + "memory_db_table_edge_rows", "Rows in knowledge_edge table"); pub static DB_TABLE_CHUNK_ROWS: Gauge = Gauge::new( "memory_db_table_chunk_rows", "Rows in memory_chunks table"); diff --git a/crates/mem-cli/src/query/bfs_graph_traversal.rs b/crates/mem-cli/src/query/bfs_graph_traversal.rs index 02eaf78..3a3d060 100644 --- a/crates/mem-cli/src/query/bfs_graph_traversal.rs +++ b/crates/mem-cli/src/query/bfs_graph_traversal.rs @@ -1,6 +1,6 @@ /// BFS graph traversal with PostgreSQL queries. /// -/// Performs breadth-first search on memory_entity + memory_edge tables, +/// Performs breadth-first search on knowledge_node + knowledge_edge tables, /// returning a subgraph for visualization. use std::collections::{HashMap, VecDeque}; @@ -215,7 +215,7 @@ impl BfsGraphTraversal { async fn load_entity(&self, id: &str) -> Result)>, String> { let query = r#" SELECT id, entity_type, name, description - FROM memory_entity + FROM knowledge_node WHERE id = $1 AND deleted_at IS NULL LIMIT 1; "#; @@ -239,7 +239,7 @@ impl BfsGraphTraversal { async fn load_edges_from(&self, source_id: &str, limit: usize) -> Result, String> { let query = r#" SELECT id, target_id, source_id, relation_type, fact, strength - FROM memory_edge + FROM knowledge_edge WHERE source_id = $1 AND t_expired IS NULL AND t_invalid IS NULL ORDER BY strength DESC LIMIT $2; diff --git a/crates/mem-cli/src/query/community_detector.rs b/crates/mem-cli/src/query/community_detector.rs index 15a2f94..70b5433 100644 --- a/crates/mem-cli/src/query/community_detector.rs +++ b/crates/mem-cli/src/query/community_detector.rs @@ -228,7 +228,7 @@ impl CommunityDetector { async fn fetch_graph(&self, _project_id: Option<&str>) -> Result<(Vec, Vec), String> { // Fetch entities let entities = sqlx::query_as::<_, (String,)>( - "SELECT DISTINCT id FROM memory_entity WHERE deleted_at IS NULL" + "SELECT DISTINCT id FROM knowledge_node WHERE deleted_at IS NULL" ) .fetch_all(&self.pool) .await @@ -240,7 +240,7 @@ impl CommunityDetector { // Fetch edges with confidence as weight let edges = sqlx::query_as::<_, (String, String, f32)>( "SELECT source_entity_id, target_entity_id, confidence - FROM memory_edge + FROM knowledge_edge WHERE fact_invalid_at IS NULL AND deleted_at IS NULL" ) .fetch_all(&self.pool) diff --git a/crates/mem-cli/src/query/faceted_search.rs b/crates/mem-cli/src/query/faceted_search.rs index 4d5d1c7..d3886bb 100644 --- a/crates/mem-cli/src/query/faceted_search.rs +++ b/crates/mem-cli/src/query/faceted_search.rs @@ -108,7 +108,7 @@ impl FacetedSearch { // Get entity types let entity_types = sqlx::query_as::<_, (String, i64)>( "SELECT entity_type, COUNT(*) as cnt - FROM memory_entity + FROM knowledge_node WHERE deleted_at IS NULL GROUP BY entity_type ORDER BY cnt DESC @@ -128,7 +128,7 @@ impl FacetedSearch { // Get total count let total_count: (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM memory_entity WHERE deleted_at IS NULL" + "SELECT COUNT(*) FROM knowledge_node WHERE deleted_at IS NULL" ) .fetch_one(&self.pool) .await @@ -212,7 +212,7 @@ impl FacetedSearch { // Get relation types let relation_types = sqlx::query_as::<_, (String, i64)>( "SELECT relation_type, COUNT(*) as cnt - FROM memory_edge + FROM knowledge_edge WHERE fact_invalid_at IS NULL AND deleted_at IS NULL GROUP BY relation_type ORDER BY cnt DESC @@ -232,7 +232,7 @@ impl FacetedSearch { // Get total count let total_count: (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM memory_edge WHERE fact_invalid_at IS NULL AND deleted_at IS NULL" + "SELECT COUNT(*) FROM knowledge_edge WHERE fact_invalid_at IS NULL AND deleted_at IS NULL" ) .fetch_one(&self.pool) .await diff --git a/crates/mem-cli/src/query/path_finder.rs b/crates/mem-cli/src/query/path_finder.rs index 72edd53..fabf908 100644 --- a/crates/mem-cli/src/query/path_finder.rs +++ b/crates/mem-cli/src/query/path_finder.rs @@ -382,7 +382,7 @@ impl PathFinder { async fn fetch_neighbors(&self, entity_id: &str) -> Result, String> { let edges = sqlx::query_as::<_, (String, String, String, f32)>( "SELECT source_entity_id, target_entity_id, relation_type, confidence - FROM memory_edge + FROM knowledge_edge WHERE (source_entity_id = $1 OR target_entity_id = $1) AND fact_invalid_at IS NULL AND deleted_at IS NULL" diff --git a/crates/mem-cli/src/query/semantic_retriever.rs b/crates/mem-cli/src/query/semantic_retriever.rs index 3448732..a6791c7 100644 --- a/crates/mem-cli/src/query/semantic_retriever.rs +++ b/crates/mem-cli/src/query/semantic_retriever.rs @@ -117,7 +117,7 @@ impl SemanticRetriever { "SELECT id, name, entity_type, 1 - (embedding <=> $1::vector) as similarity_score, metadata - FROM memory_entity + FROM knowledge_node WHERE deleted_at IS NULL AND (1 - (embedding <=> $1::vector)) > $2 AND (entity_type = COALESCE($3, entity_type)) @@ -191,9 +191,9 @@ impl SemanticRetriever { src.name, tgt.name, e.relation_type, e.fact, 1 - (e.embedding <=> $1::vector) as similarity_score, e.confidence - FROM memory_edge e - JOIN memory_entity src ON e.source_entity_id = src.id - JOIN memory_entity tgt ON e.target_entity_id = tgt.id + FROM knowledge_edge e + JOIN knowledge_node src ON e.source_entity_id = src.id + JOIN knowledge_node tgt ON e.target_entity_id = tgt.id WHERE e.fact_invalid_at IS NULL AND e.deleted_at IS NULL AND (e.relation_type = COALESCE($2, e.relation_type)) diff --git a/crates/mem-core/src/edge.rs b/crates/mem-core/src/edge.rs index a155a68..ead00be 100644 --- a/crates/mem-core/src/edge.rs +++ b/crates/mem-core/src/edge.rs @@ -51,8 +51,8 @@ pub struct Edge { pub project_id: String, // Relationship - pub source_entity_id: String, // FK to memory_entity - pub target_entity_id: String, // FK to memory_entity + pub source_entity_id: String, // FK to knowledge_node + pub target_entity_id: String, // FK to knowledge_node pub relation_type: String, pub fact: String, pub fact_embedding: Option>, @@ -71,7 +71,7 @@ pub struct Edge { // Provenance pub source_episode_id: Option, // FK to memory_node - pub invalidated_by: Option, // FK to memory_edge.id + pub invalidated_by: Option, // FK to knowledge_edge.id // Contradiction handling pub contradiction_status: ContradictionStatus, diff --git a/crates/mem-ingest/src/grm_retriever.rs b/crates/mem-ingest/src/grm_retriever.rs index d6a85a1..a6fffec 100644 --- a/crates/mem-ingest/src/grm_retriever.rs +++ b/crates/mem-ingest/src/grm_retriever.rs @@ -204,7 +204,7 @@ impl GraphContextRetriever for PostgresGrmRetriever { debug!("PostgresGrmRetriever: get_entity_context({})", entity_name); // TODO (Phase 2.5): Implement actual database query - // SELECT id, name, summary FROM memory_entity + // SELECT id, name, summary FROM knowledge_node // WHERE name_embedding <-> query_embedding < (1 - threshold) // LIMIT max_entity_context_size @@ -263,7 +263,7 @@ impl GraphContextRetriever for PostgresGrmRetriever { debug!("PostgresGrmRetriever: get_fact_context({})", fact_text); // TODO (Phase 2.5): Implement actual database query - // SELECT COUNT(*) FROM memory_edge + // SELECT COUNT(*) FROM knowledge_edge // WHERE source_id = ? AND target_id = ? // AND fact_embedding <-> query_embedding < (1 - similarity_threshold) // AND (t_invalid IS NULL OR t_invalid > NOW()) diff --git a/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json b/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json index 8cbf0c7..e427fbe 100644 --- a/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json +++ b/crates/mem-store/.sqlx/query-1e81bb729531ca33e4cef21623bcfe4fafb0c1bd435353b205f582bfda8873bc.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ", "describe": { "columns": [ { diff --git a/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json b/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json index 6a40d9c..96f4c16 100644 --- a/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json +++ b/crates/mem-store/.sqlx/query-62d65d4afc4d292b37de8e5cb59fbd51c602bdc1b437988f54e6c7fe268b9816.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ", "describe": { "columns": [ { diff --git a/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json b/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json index 61e13f9..46deac2 100644 --- a/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json +++ b/crates/mem-store/.sqlx/query-aee5900f5e3d7cbba23729bbf2dd033dcc4cb41f6c851bf447a9238810684d18.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ", "describe": { "columns": [ { diff --git a/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json b/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json index 9fe298e..587946b 100644 --- a/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json +++ b/crates/mem-store/.sqlx/query-c045466e1fe037dbdafea1008f262f4e48f104ea77732aa1d32ecb797f70e71d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND version_num = $2\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND version_num = $2\n ", "describe": { "columns": [ { diff --git a/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json b/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json index c4cca32..bccd73a 100644 --- a/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json +++ b/crates/mem-store/.sqlx/query-ca6872495bc04c6a65531279af8c758637c902dda2cc10366662988c6973ca48.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", + "query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ", "describe": { "columns": [ { diff --git a/crates/mem-store/migrations/003_temporal_graph_phase1.sql b/crates/mem-store/migrations/003_temporal_graph_phase1.sql index ac22a6a..0f78ba0 100644 --- a/crates/mem-store/migrations/003_temporal_graph_phase1.sql +++ b/crates/mem-store/migrations/003_temporal_graph_phase1.sql @@ -53,7 +53,7 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_community_fts -- ============================================ -- STEP 3: Create entity table -- ============================================ -CREATE TABLE IF NOT EXISTS memory_entity ( +CREATE TABLE IF NOT EXISTS knowledge_node ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), project_id VARCHAR(255) NOT NULL, name VARCHAR(500) NOT NULL, @@ -75,40 +75,40 @@ CREATE TABLE IF NOT EXISTS memory_entity ( ); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_project - ON memory_entity(project_id); + ON knowledge_node(project_id); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_name_embedding - ON memory_entity USING hnsw (name_embedding vector_cosine_ops) + ON knowledge_node USING hnsw (name_embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_summary_embedding - ON memory_entity USING hnsw (summary_embedding vector_cosine_ops) + ON knowledge_node USING hnsw (summary_embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_type - ON memory_entity(project_id, entity_type); + ON knowledge_node(project_id, entity_type); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_community - ON memory_entity(community_id) + ON knowledge_node(community_id) WHERE community_id IS NOT NULL; CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_active - ON memory_entity(project_id) + ON knowledge_node(project_id) WHERE t_expired IS NULL; CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_fts - ON memory_entity USING gin( + ON knowledge_node USING gin( to_tsvector('english', COALESCE(name, '') || ' ' || COALESCE(summary, '')) ); -- ============================================ -- STEP 4: Create edge (relationship) table -- ============================================ -CREATE TABLE IF NOT EXISTS memory_edge ( +CREATE TABLE IF NOT EXISTS knowledge_edge ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), project_id VARCHAR(255) NOT NULL, - source_entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE, - target_entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE, + source_entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE, + target_entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE, relation_type VARCHAR(100) NOT NULL, fact TEXT NOT NULL, fact_embedding VECTOR(768), @@ -123,7 +123,7 @@ CREATE TABLE IF NOT EXISTS memory_edge ( -- Provenance source_episode_id BIGINT REFERENCES memory_node(id) ON DELETE SET NULL, - invalidated_by UUID REFERENCES memory_edge(id) ON DELETE SET NULL, + invalidated_by UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL, -- Contradiction handling contradiction_status VARCHAR(20) DEFAULT 'active' @@ -141,34 +141,34 @@ CREATE TABLE IF NOT EXISTS memory_edge ( ); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_project - ON memory_edge(project_id); + ON knowledge_edge(project_id); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_source - ON memory_edge(source_entity_id); + ON knowledge_edge(source_entity_id); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_target - ON memory_edge(target_entity_id); + ON knowledge_edge(target_entity_id); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_entity_pair - ON memory_edge(source_entity_id, target_entity_id); + ON knowledge_edge(source_entity_id, target_entity_id); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_embedding - ON memory_edge USING hnsw (fact_embedding vector_cosine_ops) + ON knowledge_edge USING hnsw (fact_embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_validity - ON memory_edge(t_valid, t_invalid) + ON knowledge_edge(t_valid, t_invalid) WHERE t_expired IS NULL; CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_active - ON memory_edge(project_id, contradiction_status) + ON knowledge_edge(project_id, contradiction_status) WHERE t_expired IS NULL AND contradiction_status = 'active'; CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_fts - ON memory_edge USING gin(to_tsvector('english', fact)); + ON knowledge_edge USING gin(to_tsvector('english', fact)); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_relation - ON memory_edge(project_id, relation_type); + ON knowledge_edge(project_id, relation_type); -- ============================================ -- STEP 5: Compaction audit log @@ -202,8 +202,8 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_compaction_project CREATE TABLE IF NOT EXISTS contradiction_review_queue ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), project_id VARCHAR(255) NOT NULL, - new_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, - existing_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, + new_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, + existing_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, confidence FLOAT NOT NULL, explanation TEXT, queued_at TIMESTAMPTZ DEFAULT NOW(), diff --git a/crates/mem-store/migrations/004_auth_schema.sql b/crates/mem-store/migrations/004_auth_schema.sql index 0423e40..b1c2752 100644 --- a/crates/mem-store/migrations/004_auth_schema.sql +++ b/crates/mem-store/migrations/004_auth_schema.sql @@ -41,53 +41,53 @@ CREATE INDEX IF NOT EXISTS idx_memory_projects_visibility -- 2. Add attribution columns to entities -- ───────────────────────────────────────────────────────────────────────────── -ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS ( +ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS ( contributed_by VARCHAR(255), -- JWT "sub" who added this contribution_type VARCHAR(50), -- "extracted" | "manual" | "inferred" contribution_date TIMESTAMPTZ ); -- Populate contributed_by with defaults (assume "system" if not present) -UPDATE memory_entity +UPDATE knowledge_node SET contributed_by = 'system' WHERE contributed_by IS NULL; -ALTER TABLE memory_entity +ALTER TABLE knowledge_node ALTER COLUMN contributed_by SET NOT NULL; -CREATE INDEX IF NOT EXISTS idx_memory_entity_contributed_by - ON memory_entity(contributed_by); +CREATE INDEX IF NOT EXISTS idx_knowledge_node_contributed_by + ON knowledge_node(contributed_by); -CREATE INDEX IF NOT EXISTS idx_memory_entity_contribution_date - ON memory_entity(contribution_date) +CREATE INDEX IF NOT EXISTS idx_knowledge_node_contribution_date + ON knowledge_node(contribution_date) WHERE contribution_date IS NOT NULL; -- ───────────────────────────────────────────────────────────────────────────── -- 3. Add attribution columns to edges -- ───────────────────────────────────────────────────────────────────────────── -ALTER TABLE memory_edge ADD COLUMN IF NOT EXISTS ( +ALTER TABLE knowledge_edge ADD COLUMN IF NOT EXISTS ( contributed_by VARCHAR(255), -- JWT "sub" who added this contribution_type VARCHAR(50) -- "extracted" | "inferred" ); -- Populate contributed_by with defaults -UPDATE memory_edge +UPDATE knowledge_edge SET contributed_by = 'system' WHERE contributed_by IS NULL; -ALTER TABLE memory_edge +ALTER TABLE knowledge_edge ALTER COLUMN contributed_by SET NOT NULL; -CREATE INDEX IF NOT EXISTS idx_memory_edge_contributed_by - ON memory_edge(contributed_by); +CREATE INDEX IF NOT EXISTS idx_knowledge_edge_contributed_by + ON knowledge_edge(contributed_by); -- ───────────────────────────────────────────────────────────────────────────── -- 4. Add project_id to entities and edges (optional, for faster filtering) -- ───────────────────────────────────────────────────────────────────────────── --- ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS project_id VARCHAR(255); --- ALTER TABLE memory_edge ADD COLUMN IF NOT EXISTS project_id VARCHAR(255); +-- ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS project_id VARCHAR(255); +-- ALTER TABLE knowledge_edge ADD COLUMN IF NOT EXISTS project_id VARCHAR(255); -- -- Note: Deferred. Can use project info from path/source instead. @@ -102,7 +102,7 @@ SELECT COUNT(*) as entity_count, MAX(contribution_date) as last_contribution, ARRAY_AGG(DISTINCT contribution_type) as contribution_types -FROM memory_entity +FROM knowledge_node WHERE contribution_date IS NOT NULL GROUP BY contributed_by; @@ -112,7 +112,7 @@ SELECT contributed_by, COUNT(*) as count, MAX(contribution_date) as latest -FROM memory_entity +FROM knowledge_node WHERE contribution_date > CURRENT_TIMESTAMP - INTERVAL '7 days' GROUP BY contributed_by; @@ -129,12 +129,12 @@ GROUP BY contributed_by; -- DROP INDEX IF EXISTS idx_memory_projects_visibility; -- DROP TABLE IF EXISTS memory_projects; -- --- DROP INDEX IF EXISTS idx_memory_entity_contributed_by; --- DROP INDEX IF EXISTS idx_memory_entity_contribution_date; --- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contributed_by; --- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contribution_type; --- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contribution_date; +-- DROP INDEX IF EXISTS idx_knowledge_node_contributed_by; +-- DROP INDEX IF EXISTS idx_knowledge_node_contribution_date; +-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contributed_by; +-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contribution_type; +-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contribution_date; -- --- DROP INDEX IF EXISTS idx_memory_edge_contributed_by; --- ALTER TABLE memory_edge DROP COLUMN IF EXISTS contributed_by; --- ALTER TABLE memory_edge DROP COLUMN IF EXISTS contribution_type; +-- DROP INDEX IF EXISTS idx_knowledge_edge_contributed_by; +-- ALTER TABLE knowledge_edge DROP COLUMN IF EXISTS contributed_by; +-- ALTER TABLE knowledge_edge DROP COLUMN IF EXISTS contribution_type; diff --git a/crates/mem-store/migrations/005_workflows_schema.sql b/crates/mem-store/migrations/005_workflows_schema.sql index 92984dd..3c38240 100644 --- a/crates/mem-store/migrations/005_workflows_schema.sql +++ b/crates/mem-store/migrations/005_workflows_schema.sql @@ -18,8 +18,8 @@ CREATE TABLE IF NOT EXISTS temporal_workflow_links ( run_id VARCHAR(255), -- Temporal run ID -- Memory Entity References (Optional - NULL until linked) - entity_id UUID REFERENCES memory_entity(id) ON DELETE SET NULL, - edge_id UUID REFERENCES memory_edge(id) ON DELETE SET NULL, + entity_id UUID REFERENCES knowledge_node(id) ON DELETE SET NULL, + edge_id UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL, node_id BIGINT REFERENCES memory_node(id) ON DELETE SET NULL, -- Sync Status diff --git a/crates/mem-store/migrations/006_phase4_community_detection.sql b/crates/mem-store/migrations/006_phase4_community_detection.sql index 1bd1767..986075c 100644 --- a/crates/mem-store/migrations/006_phase4_community_detection.sql +++ b/crates/mem-store/migrations/006_phase4_community_detection.sql @@ -47,7 +47,7 @@ CREATE TABLE IF NOT EXISTS community_member_map ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), project_id VARCHAR(255) NOT NULL, community_id UUID NOT NULL REFERENCES memory_community(id) ON DELETE CASCADE, - entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE, + entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE, label_propagation_run_id UUID REFERENCES label_propagation_run(id) ON DELETE SET NULL, -- Label strength (0-1, higher = stronger membership) diff --git a/crates/mem-store/migrations/007_versioning_schema.sql b/crates/mem-store/migrations/007_versioning_schema.sql index f1a6be6..fe2804d 100644 --- a/crates/mem-store/migrations/007_versioning_schema.sql +++ b/crates/mem-store/migrations/007_versioning_schema.sql @@ -5,7 +5,7 @@ BEGIN; -- Entity version snapshots (immutable) -CREATE TABLE IF NOT EXISTS memory_entity_version ( +CREATE TABLE IF NOT EXISTS knowledge_node_version ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), entity_id VARCHAR(255) NOT NULL, version_num INTEGER NOT NULL, @@ -28,7 +28,7 @@ CREATE TABLE IF NOT EXISTS memory_entity_version ( ); -- Edge version snapshots -CREATE TABLE IF NOT EXISTS memory_edge_version ( +CREATE TABLE IF NOT EXISTS knowledge_edge_version ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), edge_id UUID NOT NULL, version_num INTEGER NOT NULL, @@ -52,22 +52,22 @@ CREATE TABLE IF NOT EXISTS memory_edge_version ( -- Indexes for efficient lookups CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_entity_id - ON memory_entity_version(entity_id, version_num DESC); + ON knowledge_node_version(entity_id, version_num DESC); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_changed_at - ON memory_entity_version(changed_at DESC); + ON knowledge_node_version(changed_at DESC); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_changed_by - ON memory_entity_version(changed_by); + ON knowledge_node_version(changed_by); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_edge_id - ON memory_edge_version(edge_id, version_num DESC); + ON knowledge_edge_version(edge_id, version_num DESC); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_changed_at - ON memory_edge_version(changed_at DESC); + ON knowledge_edge_version(changed_at DESC); CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_changed_by - ON memory_edge_version(changed_by); + ON knowledge_edge_version(changed_by); -- Immutability enforcement: version tables are append-only CREATE OR REPLACE FUNCTION prevent_version_table_modification() @@ -77,19 +77,19 @@ BEGIN END; $$ LANGUAGE plpgsql; -CREATE TRIGGER memory_entity_version_immutable - BEFORE UPDATE OR DELETE ON memory_entity_version +CREATE TRIGGER knowledge_node_version_immutable + BEFORE UPDATE OR DELETE ON knowledge_node_version FOR EACH ROW EXECUTE FUNCTION prevent_version_table_modification(); -CREATE TRIGGER memory_edge_version_immutable - BEFORE UPDATE OR DELETE ON memory_edge_version +CREATE TRIGGER knowledge_edge_version_immutable + BEFORE UPDATE OR DELETE ON knowledge_edge_version FOR EACH ROW EXECUTE FUNCTION prevent_version_table_modification(); COMMIT; -- Rollback (for reference): --- DROP TRIGGER memory_entity_version_immutable ON memory_entity_version; --- DROP TRIGGER memory_edge_version_immutable ON memory_edge_version; +-- DROP TRIGGER knowledge_node_version_immutable ON knowledge_node_version; +-- DROP TRIGGER knowledge_edge_version_immutable ON knowledge_edge_version; -- DROP FUNCTION prevent_version_table_modification(); --- DROP TABLE memory_entity_version; --- DROP TABLE memory_edge_version; +-- DROP TABLE knowledge_node_version; +-- DROP TABLE knowledge_edge_version; diff --git a/crates/mem-store/migrations/008_phase3_compaction.sql b/crates/mem-store/migrations/008_phase3_compaction.sql index 74cdbee..352e882 100644 --- a/crates/mem-store/migrations/008_phase3_compaction.sql +++ b/crates/mem-store/migrations/008_phase3_compaction.sql @@ -9,8 +9,8 @@ CREATE TABLE IF NOT EXISTS exact_dedup_record ( project_id VARCHAR(255) NOT NULL, -- Source and target edges - source_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, - target_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, + source_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, + target_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, -- Match criteria (all must match for exact dedup) source_match BOOLEAN NOT NULL, @@ -51,8 +51,8 @@ CREATE TABLE IF NOT EXISTS stale_gc_record ( project_id VARCHAR(255) NOT NULL, -- Entity or edge marked for GC - entity_id UUID REFERENCES memory_entity(id) ON DELETE CASCADE, - edge_id UUID REFERENCES memory_edge(id) ON DELETE CASCADE, + entity_id UUID REFERENCES knowledge_node(id) ON DELETE CASCADE, + edge_id UUID REFERENCES knowledge_edge(id) ON DELETE CASCADE, -- Staleness criteria age_days INT NOT NULL, @@ -96,8 +96,8 @@ CREATE TABLE IF NOT EXISTS semantic_dedup_record ( project_id VARCHAR(255) NOT NULL, -- Source and target edges - source_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, - target_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE, + source_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, + target_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE, -- Pre-filter score (0-1, eliminates 60-70% of candidates) prefilter_score FLOAT NOT NULL, @@ -119,7 +119,7 @@ CREATE TABLE IF NOT EXISTS semantic_dedup_record ( -- Merge strategy (if auto-merged) merge_strategy VARCHAR(50), -- 'keep_superset', 'keep_newer', 'keep_higher_confidence' - merged_edge_id UUID REFERENCES memory_edge(id) ON DELETE SET NULL, + merged_edge_id UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL, -- Metadata detected_at TIMESTAMPTZ DEFAULT NOW(), diff --git a/crates/mem-store/migrations/009_temporal_edge_migration.sql b/crates/mem-store/migrations/009_temporal_edge_migration.sql index 497dc0a..0a6021a 100644 --- a/crates/mem-store/migrations/009_temporal_edge_migration.sql +++ b/crates/mem-store/migrations/009_temporal_edge_migration.sql @@ -1,21 +1,10 @@ --- Migration 009: Temporal edge schema (Zep paper §2.2.2) --- Replaces old memory_edge (child_sha/parent_sha node graph) --- with temporal edge schema supporting relation types, facts, and validity periods. +-- Migration 009: Temporal knowledge graph edge schema (Zep paper §2.2.2) +-- Creates knowledge_edge table for entity relationships. +-- The old memory_edge (child_sha/parent_sha provenance DAG) is kept as-is. -- Idempotent: safe to run multiple times. --- Rename old table if it still exists (skip if already migrated) -DO $$ -BEGIN - IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'memory_edge' - AND EXISTS (SELECT 1 FROM information_schema.columns - WHERE table_name = 'memory_edge' AND column_name = 'child_sha')) - THEN - ALTER TABLE memory_edge RENAME TO memory_edge_legacy; - END IF; -END $$; - --- Create temporal edge table -CREATE TABLE IF NOT EXISTS memory_edge ( +-- Create knowledge graph edge table (separate from provenance memory_edge) +CREATE TABLE IF NOT EXISTS knowledge_edge ( id TEXT PRIMARY KEY, project_id TEXT NOT NULL DEFAULT 'default', source_id TEXT NOT NULL, @@ -38,30 +27,30 @@ CREATE TABLE IF NOT EXISTS memory_edge ( -- Ensure app user owns the table DO $$ BEGIN IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'app') THEN - ALTER TABLE memory_edge OWNER TO app; + ALTER TABLE knowledge_edge OWNER TO app; END IF; END $$; -CREATE INDEX IF NOT EXISTS idx_memory_edge_source ON memory_edge(source_id); -CREATE INDEX IF NOT EXISTS idx_memory_edge_target ON memory_edge(target_id); -CREATE INDEX IF NOT EXISTS idx_memory_edge_project ON memory_edge(project_id); -CREATE INDEX IF NOT EXISTS idx_memory_edge_relation ON memory_edge(relation_type); +CREATE INDEX IF NOT EXISTS idx_knowledge_edge_source ON knowledge_edge(source_id); +CREATE INDEX IF NOT EXISTS idx_knowledge_edge_target ON knowledge_edge(target_id); +CREATE INDEX IF NOT EXISTS idx_knowledge_edge_project ON knowledge_edge(project_id); +CREATE INDEX IF NOT EXISTS idx_knowledge_edge_relation ON knowledge_edge(relation_type); --- Ensure memory_entity has all columns code expects -ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; -ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS source_count INTEGER DEFAULT 1; +-- Ensure knowledge_node has all columns code expects +ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; +ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS source_count INTEGER DEFAULT 1; -- Unique constraint for entity upsert dedup DO $$ BEGIN -- Dedup existing rows before creating unique index - DELETE FROM memory_entity a USING memory_entity b + DELETE FROM knowledge_node a USING knowledge_node b WHERE a.project_id = b.project_id AND a.name = b.name AND a.t_created < b.t_created; EXCEPTION WHEN OTHERS THEN NULL; END $$; -CREATE UNIQUE INDEX IF NOT EXISTS idx_memory_entity_project_name ON memory_entity(project_id, name); +CREATE UNIQUE INDEX IF NOT EXISTS idx_knowledge_node_project_name ON knowledge_node(project_id, name); -- ROLLBACK instructions: --- DROP TABLE IF EXISTS memory_edge; --- ALTER TABLE IF EXISTS memory_edge_legacy RENAME TO memory_edge; +-- DROP TABLE IF EXISTS knowledge_edge; +-- DROP INDEX IF EXISTS idx_knowledge_node_project_name; diff --git a/crates/mem-store/src/audit_logger.rs b/crates/mem-store/src/audit_logger.rs index 7613cfd..429579c 100644 --- a/crates/mem-store/src/audit_logger.rs +++ b/crates/mem-store/src/audit_logger.rs @@ -25,7 +25,7 @@ impl AuditLogger { ) -> Result<(), sqlx::Error> { sqlx::query( r#" - INSERT INTO memory_entity_version + INSERT INTO knowledge_node_version (entity_id, version_num, operation, snapshot, changed_by, fields_changed) VALUES ($1, $2, $3, $4, $5, $6) "#, @@ -54,7 +54,7 @@ impl AuditLogger { ) -> Result<(), sqlx::Error> { sqlx::query( r#" - INSERT INTO memory_edge_version + INSERT INTO knowledge_edge_version (edge_id, version_num, operation, snapshot, changed_by, fields_changed) VALUES ($1, $2, $3, $4, $5, $6) "#, @@ -87,7 +87,7 @@ impl AuditLogger { changed_at, changed_by, COALESCE(fields_changed, '{}') as "fields_changed" - FROM memory_entity_version + FROM knowledge_node_version WHERE entity_id = $1 ORDER BY version_num DESC "#, @@ -113,7 +113,7 @@ impl AuditLogger { changed_at, changed_by, COALESCE(fields_changed, '{}') as "fields_changed" - FROM memory_edge_version + FROM knowledge_edge_version WHERE edge_id = $1 ORDER BY version_num DESC "#, diff --git a/crates/mem-store/src/db_repo.rs b/crates/mem-store/src/db_repo.rs index af4e85c..e67b366 100644 --- a/crates/mem-store/src/db_repo.rs +++ b/crates/mem-store/src/db_repo.rs @@ -80,7 +80,7 @@ impl PersistentEntityRepo { /// Save entity to database (idempotent) pub async fn save(&self, entity: &Entity) -> Result { let query = r#" - INSERT INTO memory_entity (id, entity_type, name, description, embedding, created_at, updated_at) + INSERT INTO knowledge_node (id, entity_type, name, description, embedding, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name, @@ -114,7 +114,7 @@ impl PersistentEntityRepo { pub async fn get(&self, id: &str) -> Result, DbError> { let query = r#" SELECT id, entity_type, name, description, embedding, created_at, updated_at - FROM memory_entity + FROM knowledge_node WHERE id = $1 AND deleted_at IS NULL; "#; @@ -139,7 +139,7 @@ impl PersistentEntityRepo { pub async fn list(&self, limit: i64, offset: i64) -> Result, DbError> { let query = r#" SELECT id, entity_type, name, description, embedding, created_at, updated_at - FROM memory_entity + FROM knowledge_node WHERE deleted_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2; @@ -166,7 +166,7 @@ impl PersistentEntityRepo { /// Soft delete entity pub async fn delete(&self, id: &str) -> Result<(), DbError> { let query = r#" - UPDATE memory_entity + UPDATE knowledge_node SET deleted_at = $1 WHERE id = $2; "#; @@ -195,7 +195,7 @@ impl PersistentEdgeRepo { /// Save edge to database (idempotent) pub async fn save(&self, edge: &Edge) -> Result { let query = r#" - INSERT INTO memory_edge (id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired) + INSERT INTO knowledge_edge (id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT(id) DO UPDATE SET strength = EXCLUDED.strength, @@ -232,7 +232,7 @@ impl PersistentEdgeRepo { pub async fn get(&self, id: &str) -> Result, DbError> { let query = r#" SELECT id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired - FROM memory_edge + FROM knowledge_edge WHERE id = $1 AND t_expired IS NULL; "#; @@ -260,7 +260,7 @@ impl PersistentEdgeRepo { pub async fn list_from(&self, source_id: &str, limit: i64) -> Result, DbError> { let query = r#" SELECT id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired - FROM memory_edge + FROM knowledge_edge WHERE source_id = $1 AND t_expired IS NULL AND t_invalid IS NULL ORDER BY t_created DESC LIMIT $2; @@ -290,7 +290,7 @@ impl PersistentEdgeRepo { /// Mark edge as contradicted (soft delete) pub async fn invalidate(&self, id: &str) -> Result<(), DbError> { let query = r#" - UPDATE memory_edge + UPDATE knowledge_edge SET t_invalid = $1 WHERE id = $2; "#; diff --git a/crates/mem-store/src/schema.rs b/crates/mem-store/src/schema.rs index 456f772..99884c9 100644 --- a/crates/mem-store/src/schema.rs +++ b/crates/mem-store/src/schema.rs @@ -218,6 +218,74 @@ pub async fn init_schema(pool: &PgPool) -> Result<()> { .execute(pool) .await?; - tracing::info!("Database schema initialized"); + // Memory entity table (temporal knowledge graph) + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS knowledge_node ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id VARCHAR(255) NOT NULL, + name VARCHAR(500) NOT NULL, + name_embedding VECTOR(768), + summary TEXT, + description TEXT, + summary_embedding VECTOR(768), + entity_type VARCHAR(50), + t_created TIMESTAMPTZ DEFAULT NOW(), + t_updated TIMESTAMPTZ DEFAULT NOW(), + t_expired TIMESTAMPTZ, + confidence FLOAT DEFAULT 1.0, + source_count INT DEFAULT 1, + source_episodes UUID[] DEFAULT '{}', + access_count BIGINT DEFAULT 0, + last_accessed TIMESTAMPTZ, + UNIQUE(project_id, name) + ) + "#, + ) + .execute(pool) + .await?; + + sqlx::query("CREATE INDEX IF NOT EXISTS idx_entity_project ON knowledge_node(project_id)") + .execute(pool) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_entity_type ON knowledge_node(project_id, entity_type)") + .execute(pool) + .await?; + + // Memory edge table (temporal knowledge graph) + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS knowledge_edge ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id VARCHAR(255) NOT NULL, + source_id UUID NOT NULL, + target_id UUID NOT NULL, + relation_type VARCHAR(100) NOT NULL, + fact TEXT NOT NULL, + fact_embedding VECTOR(768), + t_valid TIMESTAMPTZ, + t_invalid TIMESTAMPTZ, + t_created TIMESTAMPTZ DEFAULT NOW(), + t_expired TIMESTAMPTZ, + confidence FLOAT DEFAULT 1.0, + contradiction_status VARCHAR(20) DEFAULT 'active', + contradiction_confidence FLOAT + ) + "#, + ) + .execute(pool) + .await?; + + sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_project ON knowledge_edge(project_id)") + .execute(pool) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_source ON knowledge_edge(source_id)") + .execute(pool) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_target ON knowledge_edge(target_id)") + .execute(pool) + .await?; + + tracing::info!("Database schema initialized (including knowledge_node + knowledge_edge)"); Ok(()) } diff --git a/crates/mem-store/src/versioning.rs b/crates/mem-store/src/versioning.rs index 0c6c723..595f379 100644 --- a/crates/mem-store/src/versioning.rs +++ b/crates/mem-store/src/versioning.rs @@ -1,15 +1,16 @@ use serde::{Deserialize, Serialize}; -use sqlx::PgPool; +use sqlx::{PgPool, FromRow}; use uuid::Uuid; use chrono::{DateTime, Utc}; -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, FromRow)] pub struct VersionSnapshot { pub version_num: i32, pub operation: String, // 'create' | 'update' | 'delete' pub snapshot: serde_json::Value, pub changed_at: DateTime, pub changed_by: String, + #[sqlx(default)] pub fields_changed: Vec, } @@ -40,8 +41,7 @@ impl EntityVersioningService { /// Get all versions of an entity in descending order pub async fn get_versions(&self, entity_id: &str) -> Result, sqlx::Error> { - sqlx::query_as!( - VersionSnapshot, + sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -49,13 +49,13 @@ impl EntityVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_entity_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_node_version WHERE entity_id = $1 ORDER BY version_num DESC "#, - entity_id ) + .bind(entity_id) .fetch_all(&self.pool) .await } @@ -66,8 +66,7 @@ impl EntityVersioningService { entity_id: &str, version_num: i32, ) -> Result, sqlx::Error> { - sqlx::query_as!( - VersionSnapshot, + sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -75,13 +74,13 @@ impl EntityVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_entity_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_node_version WHERE entity_id = $1 AND version_num = $2 "#, - entity_id, - version_num ) + .bind(entity_id) + .bind(version_num) .fetch_optional(&self.pool) .await } @@ -95,78 +94,7 @@ impl EntityVersioningService { ) -> Result { let from_snap = self.get_version(entity_id, from_v).await?; let to_snap = self.get_version(entity_id, to_v).await?; - - let from_obj = from_snap - .as_ref() - .and_then(|s| s.snapshot.as_object()) - .map(|o| o.clone()); - - let to_obj = to_snap - .as_ref() - .and_then(|s| s.snapshot.as_object()) - .map(|o| o.clone()); - - let mut added = Vec::new(); - let mut removed = Vec::new(); - let mut modified = Vec::new(); - - // Check removed and modified - if let Some(ref from) = from_obj { - for (key, from_val) in from { - if let Some(to) = &to_obj { - if let Some(to_val) = to.get(key) { - if from_val != to_val { - modified.push(DiffField { - name: key.clone(), - from_value: Some(from_val.clone()), - to_value: Some(to_val.clone()), - }); - } - } else { - removed.push(DiffField { - name: key.clone(), - from_value: Some(from_val.clone()), - to_value: None, - }); - } - } else { - removed.push(DiffField { - name: key.clone(), - from_value: Some(from_val.clone()), - to_value: None, - }); - } - } - } - - // Check added - if let Some(to) = to_obj { - for (key, to_val) in to { - if let Some(from) = &from_obj { - if !from.contains_key(&key) { - added.push(DiffField { - name: key, - from_value: None, - to_value: Some(to_val), - }); - } - } else { - added.push(DiffField { - name: key, - from_value: None, - to_value: Some(to_val), - }); - } - } - } - - Ok(DiffResult { - from_version: from_v, - to_version: to_v, - added_fields: added, - removed_fields: removed, - modified_fields: modified, - }) + compute_diff(from_snap, to_snap, from_v, to_v) } /// Get entity state at a point in time @@ -175,8 +103,7 @@ impl EntityVersioningService { entity_id: &str, as_of: DateTime, ) -> Result, sqlx::Error> { - sqlx::query_as!( - VersionSnapshot, + sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -184,15 +111,15 @@ impl EntityVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_entity_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_node_version WHERE entity_id = $1 AND changed_at <= $2 ORDER BY version_num DESC LIMIT 1 "#, - entity_id, - as_of ) + .bind(entity_id) + .bind(as_of) .fetch_optional(&self.pool) .await } @@ -210,8 +137,7 @@ impl EdgeVersioningService { /// Get all versions of an edge pub async fn get_versions(&self, edge_id: Uuid) -> Result, sqlx::Error> { - sqlx::query_as!( - VersionSnapshot, + sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -219,13 +145,13 @@ impl EdgeVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_edge_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_edge_version WHERE edge_id = $1 ORDER BY version_num DESC "#, - edge_id ) + .bind(edge_id) .fetch_all(&self.pool) .await } @@ -237,8 +163,7 @@ impl EdgeVersioningService { from_v: i32, to_v: i32, ) -> Result { - let from_snap = sqlx::query_as!( - VersionSnapshot, + let from_snap = sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -246,18 +171,17 @@ impl EdgeVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_edge_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_edge_version WHERE edge_id = $1 AND version_num = $2 "#, - edge_id, - from_v ) + .bind(edge_id) + .bind(from_v) .fetch_optional(&self.pool) .await?; - let to_snap = sqlx::query_as!( - VersionSnapshot, + let to_snap = sqlx::query_as::<_, VersionSnapshot>( r#" SELECT version_num, @@ -265,17 +189,16 @@ impl EdgeVersioningService { snapshot, changed_at, changed_by, - COALESCE(fields_changed, '{}') as "fields_changed!" - FROM memory_edge_version + COALESCE(fields_changed, '{}') as fields_changed + FROM knowledge_edge_version WHERE edge_id = $1 AND version_num = $2 "#, - edge_id, - to_v ) + .bind(edge_id) + .bind(to_v) .fetch_optional(&self.pool) .await?; - // Same diff logic as entities compute_diff(from_snap, to_snap, from_v, to_v) } } diff --git a/tests/it_phase7_gate.rs b/tests/it_phase7_gate.rs index 3250ac0..10374b4 100644 --- a/tests/it_phase7_gate.rs +++ b/tests/it_phase7_gate.rs @@ -50,8 +50,8 @@ async fn test_phase7_composition_gate() { // T7.1: Schema Checks fn assert_tables_exist() { - // Verify memory_entity_version table - // Verify memory_edge_version table + // Verify knowledge_node_version table + // Verify knowledge_edge_version table // Verify index coverage } @@ -107,7 +107,7 @@ async fn test_edge_diff_endpoint() { async fn test_audit_logger_log_entity() { // Create entity snapshot // Call audit_logger.log_entity() - // Verify record inserted in memory_entity_version + // Verify record inserted in knowledge_node_version // Verify changed_by populated from JWT sub // Verify fields_changed array computed } @@ -120,9 +120,9 @@ async fn test_audit_logger_history() { } async fn test_audit_logger_immutability() { - // Try to UPDATE memory_entity_version + // Try to UPDATE knowledge_node_version // Should fail (trigger fires) - // Try to DELETE memory_entity_version + // Try to DELETE knowledge_node_version // Should fail (trigger fires) // Verify log is append-only }