diff --git a/FIXME_CRITICAL.md b/FIXME_CRITICAL.md new file mode 100644 index 0000000..ca07a79 --- /dev/null +++ b/FIXME_CRITICAL.md @@ -0,0 +1,263 @@ +# CRITICAL FIXES NEEDED - Poimen Memory Service + +## STATUS: Service Non-Functional ❌ + +**Root Issues Blocking Service**: +1. ✅ HTTP handler deadlock fixed (schema init error handling) +2. ❌ Server initialization hangs during schema or startup (logs stop after `l2_l1_edges`) +3. ❌ Ingest pipeline NOT implemented (just raw vector storage, no entities/edges) +4. ❌ Temporal schema missing (no t_valid, t_invalid, version tracking) +5. ❌ GRM gate not integrated (no memorability scores, confidence) +6. ❌ Query doesn't use knowledge graph (just vector search) +7. ❌ Compaction disabled +8. ❌ Verification gates missing + +--- + +## STEP 1: Fix Server Startup Hang ⚠️ + +**Current Issue**: Server hangs during initialization after schema creation. + +**Suspected causes**: +- OptimizerServiceBuilder.build() getting stuck +- AccessGuard creation blocking +- Background task spawning deadlock + +**Fix**: +```rust +// In http_server.rs:316-325 +// Wrap in timeout or disable non-essentials +let optimizer_service = match tokio::time::timeout( + Duration::from_secs(5), + async { mem_core::optimizer::OptimizerServiceBuilder::new().build() } +).await { + Ok(Ok(service)) => Some(Arc::new(service)), + _ => { + tracing::warn!("Optimizer initialization skipped (timeout or error)"); + None + } +}; +``` + +**Test**: `./target/release/mem serve --port 9999` should reach "Starting HTTP server" within 10s + +--- + +## STEP 2: Implement Ingest Pipeline (HIGH PRIORITY) + +**Current Implementation** (`ingest_worker.rs`): +```rust +// Just stores raw chunks + embeddings +store_chunk_l0(&l0_chunk) +store_memory_l1(&l1_memory, &embedding) +``` + +**Expected Implementation**: +```rust +// 1. Extract entities (entity_extractor) +let entities = entity_extractor.extract(&content).await?; + +// 2. Extract facts + edges (fact_extractor) +let facts = fact_extractor.extract(&content, entities).await?; + +// 3. Create temporal edges with GRM gate +for fact in facts { + let edge = TemporalEdge { + source: fact.source_entity, + target: fact.target_entity, + relation: fact.relation, + fact: fact.text, + t_valid: now(), + t_invalid: None, + confidence: grm_gate.score(&fact)?, // ← GRM gate + version: 1, + }; + edge_repo.insert(&edge).await?; +} + +// 4. Check contradictions + queue for review +for edge in edges { + if contradiction_detector.detect(&edge, existing_edges)? { + review_queue.enqueue(&edge).await?; + } +} +``` + +**Files to modify**: +- `crates/mem-cli/src/ingest_worker.rs` (core ingest logic) +- `crates/mem-ingest/src/ingest_pipeline.rs` (entity + fact extraction) +- `crates/mem-ingest/src/contradiction_detector.rs` (pre-filter + review) + +--- + +## STEP 3: Update Storage Schema (MEDIUM PRIORITY) + +**Missing fields**: +```sql +ALTER TABLE memories_l1 ADD COLUMN ( + t_valid TIMESTAMP NOT NULL DEFAULT NOW(), + t_invalid TIMESTAMP, + confidence FLOAT DEFAULT 0.5, + version INT DEFAULT 1, + memorability_score INT, + contribution_date TIMESTAMP +); + +ALTER TABLE l1_l0_edges MODIFY TO ( + l1_id UUID, + l0_id UUID, + relation_type VARCHAR, + fact TEXT, + t_valid TIMESTAMP DEFAULT NOW(), + t_invalid TIMESTAMP, + confidence FLOAT, + contradiction_flag BOOL DEFAULT FALSE, + review_queue_id UUID, + version INT DEFAULT 1, + PRIMARY KEY (l1_id, l0_id, version) +); +``` + +**Migration script**: `crates/mem-store/migrations/003_temporal_grm_schema.sql` + +--- + +## STEP 4: Wire Query Handler to Knowledge Graph (MEDIUM PRIORITY) + +**Current** (`query_handler` in http_server.rs): +```rust +async fn query_handler(...) -> HttpResponse { + // Just semantic search + let results = vector_search(query)?; + HttpResponse::Ok().json(results) +} +``` + +**Expected**: +```rust +async fn query_handler(query: QueryRequest) -> HttpResponse { + // 1. Semantic search on embeddings + let initial_results = vector_search(&query.text)?; + + // 2. Follow edges (graph traversal) + let mut expanded = vec![]; + for result in initial_results { + expanded.push(result); + // Get related entities via edges + let related = edge_repo.find_by_source(&result.entity_id).await?; + expanded.extend(related); + } + + // 3. Apply temporal filters + expanded.retain(|e| e.t_valid <= now() && (e.t_invalid.is_none() || e.t_invalid > now())); + + // 4. Sort by confidence + recency + expanded.sort_by(|a, b| { + b.confidence.partial_cmp(&a.confidence) + .then_with(|| b.t_valid.cmp(&a.t_valid)) + }); + + // 5. Apply compaction/cache alignment + for item in &mut expanded { + item.text = optimizer.compress(item.text)?; + } + + HttpResponse::Ok().json(MemoryResponse { + entities: expanded, + confidence_scores: compute_scores(&expanded), + }) +} +``` + +--- + +## STEP 5: Enable Compaction Endpoint (LOW PRIORITY) + +**Current**: Code exists but never called. + +**Fix**: Add K8s CronJob that calls `POST /memory/compact` daily: +```yaml +apiVersion: batch/v1 +kind: CronJob +metadata: + name: memory-compaction +spec: + schedule: "0 2 * * *" # 2 AM UTC + jobTemplate: + spec: + template: + spec: + containers: + - name: compact + image: bitnami/curl:latest + command: + - curl + - -X POST + - -H "Authorization: Bearer $ADMIN_TOKEN" + - http://poimen-memory:8080/memory/compact + restartPolicy: OnFailure +``` + +--- + +## STEP 6: Add Verification Gates (LOW PRIORITY) + +**Missing**: `GET /memory/verify` endpoint that checks M1.8, M2.8, M3.7, M8.9 gates + +--- + +## IMPLEMENTATION ORDER + +1. **FIX STARTUP** (1 hour) → Get server running +2. **INGEST PIPELINE** (3 hours) → Wire entity + fact extraction +3. **TEMPORAL SCHEMA** (1 hour) → Add missing columns +4. **QUERY HANDLER** (2 hours) → Implement graph traversal +5. **COMPACTION** (1 hour) → Add CronJob +6. **GATES** (2 hours) → Quality verification + +**Total**: ~10 hours to full working system + +--- + +## TEST PLAN + +```bash +# 1. Server starts +curl http://localhost:9999/health +# Expected: {"status":"ok","uptime_seconds":N} + +# 2. Ingest works +curl -X POST http://localhost:9999/memory/ingest \ + -H "Content-Type: application/json" \ + -d '{"project":"test","source":"test://1","ingest_id":"i1","records":[{"role":"user","text":"Hello world","timestamp":"2026-01-08T16:00:00Z","source_position":0}]}' +# Expected: {"ingest_id":"i1","status":"pending",...} + +# 3. Query returns entities with edges +curl -X POST http://localhost:9999/memory/query \ + -H "Content-Type: application/json" \ + -d '{"project":"test","query":"hello"}' +# Expected: {"results":[{"type":"entity","name":"...","edges":[...]}]} + +# 4. Temporal filtering works +curl http://localhost:9999/memory/query?project=test&temporal_floor=2026-01-01 + +# 5. Compaction works +curl -X POST http://localhost:9999/memory/compact +# Expected: {"phase":"completed","records_deduplicated":N} +``` + +--- + +## FILES MODIFIED SO FAR + +✅ `crates/mem-cli/src/http_server.rs` - Added error handling for schema init + +--- + +## NEXT SESSION TODO + +- [ ] Fix server startup hang (debug OptimizerService) +- [ ] Implement ingest_worker to call entity_extractor + fact_extractor +- [ ] Add temporal columns to schema +- [ ] Update query_handler to traverse edges +- [ ] Test end-to-end with sample data diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index 2aee93d..4100bc6 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -228,8 +228,13 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res tracing::info!("Connected to database"); // Initialize schema - init_schema(&pool).await?; - tracing::info!("Schema initialized"); + match init_schema(&pool).await { + Ok(_) => tracing::info!("Schema initialized"), + Err(e) => { + tracing::warn!("Schema init error (may be non-fatal): {}", e); + // Continue anyway - tables might exist + } + } // Create workers let vector_store = Arc::new(VectorStore::new(pool.clone()));