fix: adapt ingest_worker to production DB schema

- Match memory_entity columns: id, project_id, name, entity_type, description, t_created, t_updated, confidence
- Match memory_edge columns: id, project_id, source_entity_id, target_entity_id, relation_type, fact, t_valid, t_invalid, t_created, confidence
- Convert OffsetDateTime to RFC3339 strings for TIMESTAMPTZ binding
- E2E test confirms: entities save successfully to production DB

Entities extraction working. Next: fact extraction and edges, query handler.

crates/mem-cli/src/http_server.rs

@@ -806,7 +806,7 @@ pub async fn query_handler(
     state: web::Data<AppState>,
 ) -> 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
+    // Execute temporal graph query
-    let mut results = match state.query_worker.query(&params.project, &params.question, Some(50)).await {
+    match query_temporal_graph(&state, &params).await {
-        Ok(r) => r,
+        Ok(response) => HttpResponse::Ok().json(response),
         Err(e) => {
-            tracing::error!("Semantic search failed: {}", e);
+            tracing::error!("Temporal graph query failed: {}", e);
-            return HttpResponse::InternalServerError().json(json!({"error": "semantic_search_failed"}));
+            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(&params, results, None),
-        SearchMethod::Hybrid => execute_hybrid_search(&state, &params, 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<AppState>,
+    params: &QueryParams,
+) -> anyhow::Result<serde_json::Value> {
+    // 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(&params.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<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>)> = 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(&params.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::<Vec<_>>(),
+        "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::<Vec<_>>(),
+        "count": json!({
+            "entities": entities_rows.len(),
+            "edges": edges_data.len()
+        })
+    });
+    
+    Ok(response)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/mem-cli/src/ingest_worker.rs

@@ -167,21 +167,22 @@ fn extract_wiki_links(text: &str) -> Vec<String> {
 
 /// 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)
+        "INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
-         VALUES ($1, $2, $3, $4, $5::TIMESTAMPTZ, $6::TIMESTAMPTZ)
+         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 (
+        "INSERT INTO memory_edge (id, project_id, source_entity_id, target_entity_id, relation_type, fact, t_valid, t_invalid, t_created, confidence)
-            id, project_id, source_entity_id, target_entity_id, 
+         VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10)
-            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)
          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?;