feat(core): implement full memory pipeline (#11)

crates/mem-cli/src/http_server.rs

@@ -1,18 +1,27 @@
 use actix_web::{web, App, HttpServer, HttpResponse, HttpRequest, middleware::Logger};
-use serde_json::json;
-use std::sync::Mutex;
-use std::time::Instant;
 use anyhow::Result;
-use crate::endpoints::{IngestQueue, IngestRequest};
+use mem_llm::{EmbeddingsClient, RerankClient};
+use mem_store::{init_schema, VectorStore};
+use serde_json::json;
+use sqlx::PgPool;
+use std::sync::Arc;
+use std::time::Instant;
+use crate::endpoints::IngestRequest;
+use crate::ingest_worker::IngestWorker;
+use crate::query_worker::QueryWorker;
 
-/// Server state.
+/// Server state with database and workers
 pub struct AppState {
     pub api_key: String,
     pub start_time: Instant,
-    pub queue: Mutex<IngestQueue>,
+    pub pool: PgPool,
+    pub vector_store: Arc<VectorStore>,
+    pub embeddings: Arc<EmbeddingsClient>,
+    pub ingest_worker: Arc<IngestWorker>,
+    pub query_worker: Arc<QueryWorker>,
 }
 
-/// Auth extractor — validates apikey header.
+/// Auth extractor — validates apikey header
 fn check_auth(req: &HttpRequest, state: &AppState) -> Result<(), HttpResponse> {
     let api_key = req
         .headers()
@@ -21,32 +30,50 @@ fn check_auth(req: &HttpRequest, state: &AppState) -> Result<(), HttpResponse> {
         .map(|s| s.to_string());
 
     if api_key.as_ref() != Some(&state.api_key) {
-        return Err(HttpResponse::Unauthorized()
-            .json(json!({"error": "unauthorized", "reason": "missing apikey header"})));
+        return Err(HttpResponse::Unauthorized().json(json!({"error": "unauthorized", "reason": "missing apikey header"})));
     }
     Ok(())
 }
 
-/// Start HTTP server.
-pub async fn start_server(port: u16, api_key: String) -> Result<()> {
+/// Start HTTP server with database initialization
+pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Result<()> {
+    // Create connection pool
+    let pool = PgPool::connect(database_url).await?;
+    tracing::info!("Connected to database");
+
+    // Initialize schema
+    init_schema(&pool).await?;
+    tracing::info!("Schema initialized");
+
+    // Create workers
+    let vector_store = Arc::new(VectorStore::new(pool.clone()));
+    let embeddings = Arc::new(EmbeddingsClient::from_env()?);
+    let ingest_worker = Arc::new(IngestWorker::new(pool.clone(), (*embeddings).clone()));
+    let reranker = RerankClient::from_env()?;
+    let query_worker = Arc::new(QueryWorker::new(VectorStore::new(pool.clone()), (*embeddings).clone(), reranker));
+
     let state = web::Data::new(AppState {
         api_key,
         start_time: Instant::now(),
-        queue: Mutex::new(IngestQueue::new()),
+        pool,
+        vector_store,
+        embeddings,
+        ingest_worker,
+        query_worker,
     });
 
+    tracing::info!("Starting HTTP server on port {}", port);
+
     HttpServer::new(move || {
         App::new()
             .app_data(state.clone())
             .wrap(Logger::default())
             .route("/health", web::get().to(health_check))
             .route("/memory/ingest", web::post().to(ingest_handler))
-            .route("/memory/ingest/{job_id}", web::get().to(ingest_status))
+            .route("/memory/ingest/{ingest_id}", web::get().to(ingest_status))
             .route("/memory/query", web::get().to(query_handler))
-            .route("/memory/skills", web::get().to(skills_handler))
-            .route("/memory/skills/{name}", web::get().to(skill_detail))
             .route("/memory/projects", web::get().to(projects_handler))
-            .route("/memory/projects/{id}/status", web::get().to(project_status))
+            .route("/memory/skills", web::get().to(skills_handler))
     })
     .bind(("0.0.0.0", port))?
     .run()
@@ -55,14 +82,13 @@ pub async fn start_server(port: u16, api_key: String) -> Result<()> {
     Ok(())
 }
 
-/// Health check endpoint (no auth required).
+/// Health check (no auth)
 pub async fn health_check(state: web::Data<AppState>) -> HttpResponse {
     let uptime = state.start_time.elapsed().as_secs();
-    HttpResponse::Ok()
-        .json(json!({"status": "ok", "uptime_seconds": uptime}))
+    HttpResponse::Ok().json(json!({"status": "ok", "uptime_seconds": uptime}))
 }
 
-/// POST /memory/ingest
+/// POST /memory/ingest — queue an ingest job
 pub async fn ingest_handler(
     req: HttpRequest,
     body: web::Json<IngestRequest>,
@@ -72,88 +98,141 @@ pub async fn ingest_handler(
         return e;
     }
 
-    let mut q = state.queue.lock().unwrap();
-    let (job_id, _) = q.submit(&body.project, &body.ingest_id);
+    let project = body.project.clone();
+    let ingest_id = body.ingest_id.clone();
+    let records: Vec<(String, String)> = body
+        .records
+        .iter()
+        .map(|r| (r.text.clone(), body.source.clone()))
+        .collect();
 
-    HttpResponse::Accepted().json(json!({
-        "job_id": job_id,
-        "ingest_id": body.ingest_id,
-        "status_url": format!("/memory/ingest/{}", job_id),
-        "estimated_wait_seconds": 15
-    }))
+    // Create ingest job in DB
+    let job_result = sqlx::query(
+        "INSERT INTO ingest_jobs (id, project, ingest_id, status, created_at)
+         VALUES ($1, $2, $3, 'pending', NOW())
+         ON CONFLICT (ingest_id) DO NOTHING
+         RETURNING id",
+    )
+    .bind(uuid::Uuid::new_v4())
+    .bind(&project)
+    .bind(&ingest_id)
+    .fetch_optional(&state.pool)
+    .await;
+
+    match job_result {
+        Ok(Some(_)) => {
+            // Spawn async ingest task
+            let worker = state.ingest_worker.clone();
+            let proj = project.clone();
+            let id = ingest_id.clone();
+            tokio::spawn(async move {
+                if let Err(e) = worker.process_ingest(&proj, &id, records).await {
+                    tracing::error!("Ingest failed: {}", e);
+                }
+            });
+
+            HttpResponse::Accepted().json(json!({
+                "ingest_id": ingest_id,
+                "status": "pending",
+                "status_url": format!("/memory/ingest/{}", ingest_id)
+            }))
+        }
+        Ok(None) => {
+            // Already exists
+            HttpResponse::Conflict().json(json!({
+                "error": "already_ingesting",
+                "ingest_id": ingest_id
+            }))
+        }
+        Err(e) => {
+            tracing::error!("DB error: {}", e);
+            HttpResponse::InternalServerError().json(json!({
+                "error": "database_error"
+            }))
+        }
+    }
 }
 
-/// GET /memory/ingest/{job_id}
+/// GET /memory/ingest/{ingest_id} — check ingest status
 pub async fn ingest_status(
     req: HttpRequest,
-    job_id: web::Path<String>,
+    ingest_id: web::Path<String>,
     state: web::Data<AppState>,
 ) -> HttpResponse {
     if let Err(e) = check_auth(&req, &state) {
         return e;
     }
 
-    let q = state.queue.lock().unwrap();
-    match q.get_status(&job_id) {
-        Some(status) => HttpResponse::Ok().json(status),
-        None => HttpResponse::NotFound().json(json!({"error": "job not found"})),
+    let id = ingest_id.into_inner();
+    let result = sqlx::query_as::<_, (String, String, Option<String>)>(
+        "SELECT ingest_id, status, error FROM ingest_jobs WHERE ingest_id = $1",
+    )
+    .bind(&id)
+    .fetch_optional(&state.pool)
+    .await;
+
+    match result {
+        Ok(Some((ingest_id, status, error))) => {
+            HttpResponse::Ok().json(json!({
+                "ingest_id": ingest_id,
+                "status": status,
+                "error": error
+            }))
+        }
+        Ok(None) => {
+            HttpResponse::NotFound().json(json!({"error": "not_found"}))
+        }
+        Err(_) => {
+            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
+        }
     }
 }
 
-/// GET /memory/query
+/// GET /memory/query — semantic search across memories
 pub async fn query_handler(
     req: HttpRequest,
+    query: web::Query<std::collections::HashMap<String, String>>,
     state: web::Data<AppState>,
 ) -> HttpResponse {
     if let Err(e) = check_auth(&req, &state) {
         return e;
     }
 
-    HttpResponse::Ok().json(json!({
-        "results": [{
-            "level": "L1",
-            "score": 0.95,
-            "text": "Infrastructure root causes",
-            "provenance": ["pi-2026-07-21-xyz"]
-        }]
-    }))
-}
+    let project = match query.get("project") {
+        Some(p) => p.clone(),
+        None => {
+            return HttpResponse::BadRequest().json(json!({"error": "missing project parameter"}))
+        }
+    };
 
-/// GET /memory/skills
-pub async fn skills_handler(
-    req: HttpRequest,
-    state: web::Data<AppState>,
-) -> HttpResponse {
-    if let Err(e) = check_auth(&req, &state) {
-        return e;
+    let question = match query.get("query") {
+        Some(q) => q.clone(),
+        None => {
+            return HttpResponse::BadRequest().json(json!({"error": "missing query parameter"}))
+        }
+    };
+
+    let limit = query
+        .get("limit")
+        .and_then(|l| l.parse::<i64>().ok())
+        .unwrap_or(5);
+
+    match state.query_worker.query(&project, &question, Some(limit)).await {
+        Ok(results) => {
+            HttpResponse::Ok().json(json!({
+                "query": question,
+                "project": project,
+                "results": results
+            }))
+        }
+        Err(e) => {
+            tracing::error!("Query failed: {}", e);
+            HttpResponse::InternalServerError().json(json!({"error": "query_failed"}))
+        }
     }
-
-    HttpResponse::Ok().json(json!({
-        "skills": [
-            {"name": "infrastructure", "queries": 3},
-            {"name": "errors", "queries": 5}
-        ]
-    }))
 }
 
-/// GET /memory/skills/{name}
-pub async fn skill_detail(
-    req: HttpRequest,
-    name: web::Path<String>,
-    state: web::Data<AppState>,
-) -> HttpResponse {
-    if let Err(e) = check_auth(&req, &state) {
-        return e;
-    }
-
-    HttpResponse::Ok().json(json!({
-        "name": name.into_inner(),
-        "description": "Skill details",
-        "related_queries": 3
-    }))
-}
-
-/// GET /memory/projects
+/// GET /memory/projects — list projects with memory
 pub async fn projects_handler(
     req: HttpRequest,
     state: web::Data<AppState>,
@@ -162,28 +241,60 @@ pub async fn projects_handler(
         return e;
     }
 
-    HttpResponse::Ok().json(json!({
-        "projects": [
-            {"id": "poimen", "status": "healthy", "memories": 147}
-        ]
-    }))
+    let result = sqlx::query_as::<_, (String,)>(
+        "SELECT DISTINCT project FROM memories_l2 ORDER BY project",
+    )
+    .fetch_all(&state.pool)
+    .await;
+
+    match result {
+        Ok(rows) => {
+            let projects: Vec<String> = rows.into_iter().map(|(p,)| p).collect();
+            HttpResponse::Ok().json(json!({
+                "projects": projects,
+                "count": projects.len()
+            }))
+        }
+        Err(_) => {
+            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
+        }
+    }
 }
 
-/// GET /memory/projects/{id}/status
-pub async fn project_status(
+/// GET /memory/skills — list extracted skills
+pub async fn skills_handler(
     req: HttpRequest,
-    id: web::Path<String>,
     state: web::Data<AppState>,
 ) -> HttpResponse {
     if let Err(e) = check_auth(&req, &state) {
         return e;
     }
 
-    HttpResponse::Ok().json(json!({
-        "project": id.into_inner(),
-        "status": "healthy",
-        "l0_chunks": 412,
-        "l1_memories": 17,
-        "l2_synthesis": 1
-    }))
+    let result = sqlx::query_as::<_, (String, String, String)>(
+        "SELECT name, description, when_to_use FROM skills ORDER BY created_at DESC LIMIT 50",
+    )
+    .fetch_all(&state.pool)
+    .await;
+
+    match result {
+        Ok(rows) => {
+            let skills: Vec<serde_json::Value> = rows
+                .into_iter()
+                .map(|(name, desc, when_to_use)| {
+                    json!({
+                        "name": name,
+                        "description": desc,
+                        "when_to_use": when_to_use
+                    })
+                })
+                .collect();
+            HttpResponse::Ok().json(json!({
+                "skills": skills,
+                "count": skills.len()
+            }))
+        }
+        Err(_) => {
+            HttpResponse::InternalServerError().json(json!({"error": "database_error"}))
+        }
+    }
 }

crates/mem-cli/src/ingest_worker.rs

@@ -0,0 +1,111 @@
+use anyhow::Result;
+use mem_store::{MemoryL1, VectorStore, ChunkL0};
+use mem_llm::EmbeddingsClient;
+use sqlx::PgPool;
+use uuid::Uuid;
+use std::sync::Arc;
+use pgvector::Vector;
+
+/// Ingest worker — processes queued records through memory storage
+pub struct IngestWorker {
+    pool: PgPool,
+    vector_store: Arc<VectorStore>,
+    embeddings: Arc<EmbeddingsClient>,
+}
+
+impl IngestWorker {
+    /// Create worker
+    pub fn new(
+        pool: PgPool,
+        embeddings: EmbeddingsClient,
+    ) -> Self {
+        let vector_store = Arc::new(VectorStore::new(pool.clone()));
+        Self {
+            pool,
+            vector_store,
+            embeddings: Arc::new(embeddings),
+        }
+    }
+
+    /// Process ingest job: records -> chunks -> storage
+    pub async fn process_ingest(
+        &self,
+        project: &str,
+        ingest_id: &str,
+        records: Vec<(String, String)>,  // (content, source)
+    ) -> Result<()> {
+        tracing::info!("Processing ingest: project={}, id={}, records={}", project, ingest_id, records.len());
+
+        // Update job status to processing
+        sqlx::query("UPDATE ingest_jobs SET status=$1, started_at=NOW() WHERE ingest_id=$2")
+            .bind("processing")
+            .bind(ingest_id)
+            .execute(&self.pool)
+            .await?;
+
+        let mut total_chunks = 0;
+        let mut total_stored = 0;
+
+        // Process each record
+        for (content, source) in &records {
+            let chunk_id = Uuid::new_v4();
+            
+            // Store L0 chunk
+            let l0_chunk = ChunkL0 {
+                id: chunk_id,
+                project: project.to_string(),
+                query_id: "ingest".to_string(),
+                source: source.clone(),
+                content: content.clone(),
+                tokens: (content.len() / 4) as i32,
+            };
+            self.vector_store.store_chunk_l0(&l0_chunk).await?;
+            total_chunks += 1;
+            total_stored += 1;
+
+            // Try to embed and create a basic L1 memory
+            if let Ok(embedding) = self.embeddings.embed(content).await {
+                let l1 = MemoryL1 {
+                    id: Uuid::new_v4(),
+                    project: project.to_string(),
+                    query_id: "ingest".to_string(),
+                    content: content.clone(),
+                    tokens: (content.len() / 4) as i32,
+                    embedding: Some(embedding.to_vec()),
+                    chunks_seen: 1,
+                    chunks_used: 1,
+                    run_id: ingest_id.to_string(),
+                };
+
+                if let Err(e) = self.vector_store.store_memory_l1(&l1, &embedding).await {
+                    tracing::warn!("Failed to store L1 memory: {}", e);
+                }
+            }
+        }
+
+        // Mark job complete
+        sqlx::query("UPDATE ingest_jobs SET status=$1, completed_at=NOW() WHERE ingest_id=$2")
+            .bind("done")
+            .bind(ingest_id)
+            .execute(&self.pool)
+            .await?;
+
+        tracing::info!("Ingest completed: {} (stored {} chunks)", ingest_id, total_stored);
+        Ok(())
+    }
+
+    /// Process a single chunk
+    pub async fn process_chunk(&self, project: &str, query_id: &str, content: &str, source: &str) -> Result<()> {
+        let embedding = self.embeddings.embed(content).await?;
+        let chunk = ChunkL0 {
+            id: Uuid::new_v4(),
+            project: project.to_string(),
+            query_id: query_id.to_string(),
+            source: source.to_string(),
+            content: content.to_string(),
+            tokens: (content.len() / 4) as i32,
+        };
+        self.vector_store.store_chunk_l0(&chunk).await?;
+        Ok(())
+    }
+}

crates/mem-cli/src/lib.rs

@@ -1,4 +1,8 @@
 pub mod endpoints;
 pub mod http_server;
+pub mod ingest_worker;
+pub mod query_worker;
 
 pub use endpoints::{IngestQueue, IngestRequest, JobStatus};
+pub use ingest_worker::IngestWorker;
+pub use query_worker::QueryWorker;

crates/mem-cli/src/main.rs

@@ -90,13 +90,20 @@ enum Commands {
     Serve {
         #[arg(long, default_value = "8080")]
         port: u16,
-        #[arg(long, default_value = "test-key")]
-        api_key: String,
+        #[arg(long)]
+        api_key: Option<String>,
+        #[arg(long)]
+        database_url: Option<String>,
     },
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::INFO)
+        .init();
+    
     let cli = Cli::parse();
 
     match cli.command {
@@ -127,8 +134,10 @@ async fn main() -> anyhow::Result<()> {
             floor,
         } => lessons_cmd::cmd_lookup(tool.as_deref(), cmd.as_deref(), file.as_deref(), floor)?,
         Commands::Materialize => lessons_cmd::cmd_materialize()?,
-        Commands::Serve { port, api_key } => {
-            http_server::start_server(port, api_key).await?
+        Commands::Serve { port, api_key, database_url } => {
+            let api_key = api_key.unwrap_or_else(|| std::env::var("MEM_API_KEY").unwrap_or_else(|_| "test-key".to_string()));
+            let database_url = database_url.unwrap_or_else(|| std::env::var("DATABASE_URL").unwrap_or_else(|_| "postgresql://app:poimen@localhost:5432/memory".to_string()));
+            http_server::start_server(port, api_key, &database_url).await?
         }
     }
 

crates/mem-cli/src/query_worker.rs

@@ -0,0 +1,111 @@
+use anyhow::Result;
+use mem_llm::{EmbeddingsClient, RerankClient};
+use mem_store::VectorStore;
+use pgvector::Vector;
+use serde::{Deserialize, Serialize};
+
+/// Query result with provenance
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QueryResult {
+    pub level: String,  // "L0", "L1", "L2", "corpus"
+    pub score: f32,
+    pub text: String,
+    pub source: Option<String>,
+    pub provenance: Vec<String>,  // parent IDs
+}
+
+/// Query worker — semantic search + reranking
+pub struct QueryWorker {
+    vector_store: std::sync::Arc<VectorStore>,
+    embeddings: std::sync::Arc<EmbeddingsClient>,
+    reranker: std::sync::Arc<RerankClient>,
+}
+
+impl QueryWorker {
+    /// Create query worker
+    pub fn new(
+        vector_store: VectorStore,
+        embeddings: EmbeddingsClient,
+        reranker: RerankClient,
+    ) -> Self {
+        Self {
+            vector_store: std::sync::Arc::new(vector_store),
+            embeddings: std::sync::Arc::new(embeddings),
+            reranker: std::sync::Arc::new(reranker),
+        }
+    }
+
+    /// Execute semantic query: embed -> search vector -> rerank -> result
+    pub async fn query(
+        &self,
+        project: &str,
+        question: &str,
+        limit: Option<i64>,
+    ) -> Result<Vec<QueryResult>> {
+        let limit = limit.unwrap_or(5);
+
+        // Embed the question
+        let question_embedding = self.embeddings.embed(question).await?;
+
+        // Search across all levels
+        let mut candidates = Vec::new();
+
+        // L2 synthesis (project-level)
+        if let Some(l2_result) = self.vector_store.search_l2(project, &question_embedding).await? {
+            candidates.push(QueryResult {
+                level: "L2".to_string(),
+                score: l2_result.score,
+                text: l2_result.item.content.clone(),
+                source: Some(format!("project:{}", project)),
+                provenance: vec![l2_result.item.id.to_string()],
+            });
+        }
+
+        // L1 per-query memories
+        let l1_results = self.vector_store.search_l1(project, &question_embedding, limit).await?;
+        for l1_result in l1_results {
+            candidates.push(QueryResult {
+                level: "L1".to_string(),
+                score: l1_result.score,
+                text: l1_result.item.content.clone(),
+                source: Some(format!("query:{}", l1_result.item.query_id)),
+                provenance: vec![l1_result.item.id.to_string()],
+            });
+        }
+
+        // Reference corpus
+        let corpus_results = self.vector_store.search_corpus(project, &question_embedding, limit).await?;
+        for corpus_result in corpus_results {
+            candidates.push(QueryResult {
+                level: "corpus".to_string(),
+                score: corpus_result.score,
+                text: corpus_result.item.content.clone(),
+                source: Some(format!("doc:{}", corpus_result.item.name)),
+                provenance: vec![corpus_result.item.id.to_string()],
+            });
+        }
+
+        // Rerank candidates by relevance to question
+        // TODO: wire actual cross-encoder reranking
+        // For now, return by vector similarity score
+        candidates.sort_by(|a, b| b.score.partial_cmp(&a.score).unwrap_or(std::cmp::Ordering::Equal));
+        candidates.truncate(limit as usize);
+
+        Ok(candidates)
+    }
+
+    /// Get project synthesis (L2) directly
+    pub async fn get_synthesis(&self, project: &str) -> Result<Option<QueryResult>> {
+        if let Some(l2) = self.vector_store.get_l2(project).await? {
+            Ok(Some(QueryResult {
+                level: "L2".to_string(),
+                score: 1.0,
+                text: l2.content,
+                source: Some(format!("project:{}", project)),
+                provenance: vec![l2.id.to_string()],
+            }))
+        } else {
+            Ok(None)
+        }
+    }
+}