diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index e35c386..7baaad9 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -806,7 +806,7 @@ pub async fn query_handler( state: web::Data, ) -> HttpResponse { // Auth + capability check - let (claims, token) = match validate_auth(&req, &state).await { + let (claims, _token) = match validate_auth(&req, &state).await { Ok(c) => c, Err(e) => return e, }; @@ -826,22 +826,13 @@ pub async fn query_handler( Err(e) => return e.to_response(), }; - // Execute semantic search - let mut results = match state.query_worker.query(¶ms.project, ¶ms.question, Some(50)).await { - Ok(r) => r, + // Execute temporal graph query + match query_temporal_graph(&state, ¶ms).await { + Ok(response) => HttpResponse::Ok().json(response), Err(e) => { - tracing::error!("Semantic search failed: {}", e); - return HttpResponse::InternalServerError().json(json!({"error": "semantic_search_failed"})); + tracing::error!("Temporal graph query failed: {}", e); + HttpResponse::InternalServerError().json(json!({"error": "query_failed", "reason": e.to_string()})) } - }; - - // M3.8: Optimize results - results = optimize_search_results(results, state.optimizer_service.as_ref()).await; - - // Route by search method - match params.method { - SearchMethod::Semantic => build_search_response(¶ms, results, None), - SearchMethod::Hybrid => execute_hybrid_search(&state, ¶ms, results, &token).await, } } @@ -1342,6 +1333,75 @@ pub async fn vault_file_handler( } } +/// Query temporal knowledge graph +/// 1. Find entities via semantic search +/// 2. Traverse edges from entities +/// 3. Apply temporal filtering (t_valid/t_invalid) +/// 4. Return graph with confidence scores +async fn query_temporal_graph( + state: &web::Data, + params: &QueryParams, +) -> 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 AND t_expired IS NULL LIMIT $2" + ) + .bind(¶ms.project) + .bind(params.limit as i32) + .fetch_all(&state.pool) + .await + .unwrap_or_default(); + + // Step 2: Traverse edges from found entities + let mut edges_data: Vec<(String, String, String, String, String, f32)> = Vec::new(); + 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 AND t_expired IS NULL" + ) + .bind(¶ms.project) + .bind(entity_id) + .fetch_all(&state.pool) + .await + .unwrap_or_default(); + + for (id, target, rel, fact, conf, t_valid, t_invalid) in entity_edges { + // Step 3: Apply temporal filtering + let now = chrono::Utc::now(); + let valid = t_valid.as_ref().map(|t| *t <= now).unwrap_or(true); + let not_invalid = t_invalid.as_ref().map(|t| *t > now).unwrap_or(true); + + if valid && not_invalid { + edges_data.push((id, entity_id.clone(), target, rel, fact, conf)); + } + } + } + + // Step 4: Build response + let response = json!({ + "query": params.question, + "project": params.project, + "entities": entities_rows.iter().map(|(id, name, etype)| json!({ + "id": id, + "name": name, + "type": etype + })).collect::>(), + "edges": edges_data.iter().map(|(id, src, tgt, rel, fact, conf)| json!({ + "id": id, + "source": src, + "target": tgt, + "relation": rel, + "fact": fact, + "confidence": conf + })).collect::>(), + "count": json!({ + "entities": entities_rows.len(), + "edges": edges_data.len() + }) + }); + + Ok(response) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/mem-cli/src/ingest_worker.rs b/crates/mem-cli/src/ingest_worker.rs index b1317b7..000e17d 100644 --- a/crates/mem-cli/src/ingest_worker.rs +++ b/crates/mem-cli/src/ingest_worker.rs @@ -167,21 +167,22 @@ fn extract_wiki_links(text: &str) -> Vec { /// Save entity to database via raw SQL (normally would use EntityRepo trait) async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) -> Result<()> { - // Convert OffsetDateTime to string in RFC3339 format, then back to chrono for sqlx + // Convert OffsetDateTime to PostgreSQL timestamp format let t_created_str = entity.t_created.to_string(); - let t_expired_str = entity.t_expired.map(|t| t.to_string()); sqlx::query( - "INSERT INTO memory_entity (id, project_id, name, entity_type, t_created, t_expired) - VALUES ($1, $2, $3, $4, $5::TIMESTAMPTZ, $6::TIMESTAMPTZ) + "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) ON CONFLICT (id) DO NOTHING" ) .bind(&entity.id) .bind(&entity.project_id) .bind(&entity.name) .bind(entity.entity_type.as_str()) + .bind(entity.summary.as_deref()) .bind(&t_created_str) - .bind(&t_expired_str) + .bind(&t_created_str) + .bind(1.0_f32) // default confidence .execute(pool) .await?; Ok(()) @@ -189,17 +190,13 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) -> /// Save edge to database via raw SQL (normally would use EdgeRepo trait) async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> { + let t_created_str = edge.t_created.to_string(); let t_valid_str = edge.t_valid.map(|t| t.to_string()); let t_invalid_str = edge.t_invalid.map(|t| t.to_string()); - let t_created_str = edge.t_created.to_string(); - let t_expired_str = edge.t_expired.map(|t| t.to_string()); sqlx::query( - "INSERT INTO memory_edge ( - id, project_id, source_entity_id, target_entity_id, - relation_type, fact, t_valid, t_invalid, t_created, t_expired, - contradiction_status, confidence - ) VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10::TIMESTAMPTZ, $11, $12) + "INSERT INTO memory_edge (id, project_id, source_entity_id, target_entity_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) ON CONFLICT (id) DO NOTHING" ) .bind(&edge.id) @@ -211,8 +208,6 @@ async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<( .bind(&t_valid_str) .bind(&t_invalid_str) .bind(&t_created_str) - .bind(&t_expired_str) - .bind(edge.contradiction_status.as_str()) .bind(edge.confidence) .execute(pool) .await?;