stage 2

Cargo.toml

@@ -29,6 +29,7 @@ utoipa-swagger-ui = { version = "9", features = ["axum", "vendored"] }
 uuid = { version = "1", features = ["v4", "serde"] }
 anyhow = "1"
 openraft = { version = "0.9", features = ["serde", "storage-v2"] }
+petgraph = "0.6"
 tonic = "0.12"
 prost = "0.13"
 tokio-stream = "0.1"

benches/e2e_throughput.rs

@@ -175,6 +175,8 @@ impl BenchmarkHarness {
         ));
         let (embedding_job_sender, receiver) = embedding_job_channel(config.channel_size);
 
+        let (knowledge_job_sender, mut krx) = tokio::sync::mpsc::channel::<engram::knowledge::types::KnowledgeJob>(16);
+        tokio::spawn(async move { while krx.recv().await.is_some() {} });
         let state = Arc::new(AppState {
             short_term_memory: short_term_memory.clone(),
             vector_store: vector_store.clone(),
@@ -191,6 +193,10 @@ impl BenchmarkHarness {
             raft_addr: None,
             raft_advertise_addr: None,
             cluster_peers: vec![],
+            knowledge_graph: Arc::new(tokio::sync::RwLock::new(
+                engram::knowledge::graph::KnowledgeGraph::new(),
+            )),
+            knowledge_job_sender,
         });
 
         let _worker_handles = spawn_embedding_workers(

docker-compose.cluster.yml

@@ -25,7 +25,8 @@ services:
       REDIS_URL: "redis://redis-1:6379"
       LANCE_DB_PATH: "/data/lancedb"
       ENGRAM_BIND_ADDR: "0.0.0.0:3000"
-      OPENAI_API_KEY: "${OPENAI_API_KEY}"
+      OPENAI_API_KEY: "${OPENAI_API_KEY:-}"
+      KNOWLEDGE_EXTRACTOR: "${KNOWLEDGE_EXTRACTOR:-mock}"
       RUST_LOG: "info,openraft=debug"
     ports:
       - "3000:3000"
@@ -45,7 +46,8 @@ services:
       REDIS_URL: "redis://redis-2:6379"
       LANCE_DB_PATH: "/data/lancedb"
       ENGRAM_BIND_ADDR: "0.0.0.0:3000"
-      OPENAI_API_KEY: "${OPENAI_API_KEY}"
+      OPENAI_API_KEY: "${OPENAI_API_KEY:-}"
+      KNOWLEDGE_EXTRACTOR: "${KNOWLEDGE_EXTRACTOR:-mock}"
       RUST_LOG: "info,openraft=debug"
     ports:
       - "3001:3000"
@@ -65,7 +67,8 @@ services:
       REDIS_URL: "redis://redis-3:6379"
       LANCE_DB_PATH: "/data/lancedb"
       ENGRAM_BIND_ADDR: "0.0.0.0:3000"
-      OPENAI_API_KEY: "${OPENAI_API_KEY}"
+      OPENAI_API_KEY: "${OPENAI_API_KEY:-}"
+      KNOWLEDGE_EXTRACTOR: "${KNOWLEDGE_EXTRACTOR:-mock}"
       RUST_LOG: "info,openraft=debug"
     ports:
       - "3002:3000"

scripts/cluster-verify.sh

@@ -96,5 +96,97 @@ echo "$METRICS" | grep -q "engram_raft_term"         && pass "engram_raft_term p
 echo "$METRICS" | grep -q "engram_raft_commit_index" && pass "engram_raft_commit_index present" || fail "engram_raft_commit_index missing"
 echo "$METRICS" | grep -q "engram_raft_is_leader"    && pass "engram_raft_is_leader present"    || fail "engram_raft_is_leader missing"
 
+
+echo "[6] Knowledge replication (deterministic mock extraction)"
+# Find current leader — may have changed after failover in check [4]
+WRITE_LEADER="$N1"
+for port in 3000 3001 3002; do
+  ROLE=$(curl -sf "http://localhost:$port/cluster" | \
+    python3 -c "import sys,json; print(json.load(sys.stdin)['role'])" 2>/dev/null || echo "")
+  if [ "$ROLE" = "Leader" ]; then
+    WRITE_LEADER="http://localhost:$port"
+    break
+  fi
+done
+
+SESSION_K=$(curl -sf -X POST "$WRITE_LEADER/sessions" | \
+  python3 -c "import sys,json; print(json.load(sys.stdin)['session_id'])")
+
+# Three separate messages: one pattern per message so the mock extractor handles each cleanly
+curl -sf -X POST "$WRITE_LEADER/sessions/$SESSION_K/messages" \
+  -H "Content-Type: application/json" \
+  -d '{"role":"user","content":"Alice works at OpenAI"}' > /dev/null
+curl -sf -X POST "$WRITE_LEADER/sessions/$SESSION_K/messages" \
+  -H "Content-Type: application/json" \
+  -d '{"role":"user","content":"Bob works at OpenAI"}' > /dev/null
+curl -sf -X POST "$WRITE_LEADER/sessions/$SESSION_K/messages" \
+  -H "Content-Type: application/json" \
+  -d '{"role":"user","content":"Alice knows Bob"}' > /dev/null
+
+echo "  Waiting 3 seconds for extraction and Raft replication..."
+sleep 3
+
+LEADER_PORT=""
+for port in 3000 3001 3002; do
+  ROLE=$(curl -sf "http://localhost:$port/cluster" | \
+    python3 -c "import sys,json; print(json.load(sys.stdin)['role'])" 2>/dev/null || echo "")
+  if [ "$ROLE" = "Leader" ]; then
+    LEADER_PORT="$port"
+    break
+  fi
+done
+
+[ -z "$LEADER_PORT" ] && fail "no leader found for knowledge check"
+
+LEADER_ENTITIES=$(curl -sf "http://localhost:$LEADER_PORT/sessions/$SESSION_K/knowledge" | \
+  python3 -c "import sys,json; print(len(json.load(sys.stdin)['entities']))" 2>/dev/null || echo "-1")
+
+[ "$LEADER_ENTITIES" -ge 3 ] \
+  && pass "leader (:$LEADER_PORT) has $LEADER_ENTITIES entities" \
+  || fail "leader (:$LEADER_PORT) has $LEADER_ENTITIES entities (expected >= 3)"
+
+LEADER_EDGES=$(curl -sf "http://localhost:$LEADER_PORT/sessions/$SESSION_K/knowledge" | \
+  python3 -c "import sys,json; print(len(json.load(sys.stdin)['edges']))" 2>/dev/null || echo "-1")
+
+[ "$LEADER_EDGES" -ge 3 ] \
+  && pass "leader (:$LEADER_PORT) has $LEADER_EDGES relationships" \
+  || fail "leader (:$LEADER_PORT) has $LEADER_EDGES relationships (expected >= 3)"
+
+for port in 3000 3001 3002; do
+  [ "$port" -eq "$LEADER_PORT" ] && continue
+  FOLLOWER_ENTITIES=$(curl -sf "http://localhost:$port/sessions/$SESSION_K/knowledge" | \
+    python3 -c "import sys,json; print(len(json.load(sys.stdin)['entities']))" 2>/dev/null || echo "-1")
+  [ "$FOLLOWER_ENTITIES" -eq "$LEADER_ENTITIES" ] \
+    && pass "follower :$port converged to $FOLLOWER_ENTITIES entities (matches leader)" \
+    || fail "follower :$port has $FOLLOWER_ENTITIES entities (leader has $LEADER_ENTITIES)"
+done
+
+echo "[6b] Capability criterion: graph answers questions without LLM or vector search"
+RELATED=$(curl -sf "http://localhost:$LEADER_PORT/sessions/$SESSION_K/knowledge/entities/OpenAI" | \
+  python3 -c "import sys,json; d=json.load(sys.stdin); print([r['name'] for r in d['related']])" 2>/dev/null || echo "[]")
+echo "$RELATED" | grep -q "Alice" \
+  && pass "OpenAI is related to Alice (works_at)" \
+  || fail "OpenAI not related to Alice"
+echo "$RELATED" | grep -q "Bob" \
+  && pass "OpenAI is related to Bob (works_at)" \
+  || fail "OpenAI not related to Bob"
+
+PATH_RESP=$(curl -sf "http://localhost:$LEADER_PORT/sessions/$SESSION_K/knowledge/path?from=Alice&to=Bob" | \
+  python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('path'))" 2>/dev/null || echo "None")
+[ "$PATH_RESP" != "None" ] && [ "$PATH_RESP" != "null" ] \
+  && pass "shortest path Alice→Bob found via graph traversal" \
+  || fail "no path found from Alice to Bob"
+
+echo "[6c] Delete-session removes knowledge graph state from all nodes"
+curl -sf -X DELETE "$WRITE_LEADER/sessions/$SESSION_K" > /dev/null
+sleep 1
+for port in 3000 3001 3002; do
+  ENTITIES_AFTER=$(curl -sf "http://localhost:$port/sessions/$SESSION_K/knowledge" | \
+    python3 -c "import sys,json; print(len(json.load(sys.stdin)['entities']))" 2>/dev/null || echo "-1")
+  [ "$ENTITIES_AFTER" -eq 0 ] \
+    && pass "node :$port knowledge graph empty after delete" \
+    || fail "node :$port still has $ENTITIES_AFTER entities after delete"
+done
+
 echo ""
-echo "=== All Stage 1 criteria PASSED ==="
+echo "=== All Stage 1 + Stage 2 criteria PASSED ==="

src/app.rs

@@ -11,6 +11,10 @@ use crate::core::{
     StoreError, TokenCounter, VectorStore,
 };
 use crate::embedding::OpenAIEmbedder;
+use crate::config::KnowledgeExtractorType;
+use crate::knowledge::extractor::{MockKnowledgeExtractor, OpenAIKnowledgeExtractor};
+use crate::knowledge::graph::KnowledgeGraph;
+use crate::knowledge::worker::{knowledge_job_channel, spawn_knowledge_workers};
 use crate::metrics::AppMetrics;
 use crate::server::AppState;
 use crate::stores::{LanceDBStore, RedisCoreMemoryStore, RedisShortTermMemory};
@@ -36,6 +40,8 @@ pub async fn build_raft_node(
     core_memory: Arc<dyn CoreMemoryStore>,
     vector_store: Arc<dyn VectorStore>,
     embedding_tx: tokio::sync::mpsc::Sender<EmbeddingJob>,
+    knowledge_graph: Arc<tokio::sync::RwLock<crate::knowledge::graph::KnowledgeGraph>>,
+    knowledge_tx: tokio::sync::mpsc::Sender<crate::knowledge::types::KnowledgeJob>,
 ) -> anyhow::Result<Arc<crate::raft::types::RaftHandle>> {
     use crate::raft::{
         log_store::EngRaftLogStore, network::EngRaftNetwork,
@@ -61,7 +67,7 @@ pub async fn build_raft_node(
         raft_config,
         EngRaftNetwork,
         EngRaftLogStore::default(),
-        EngStateMachineStore::new(short_term, core_memory, vector_store, embedding_tx),
+        EngStateMachineStore::new(short_term, core_memory, vector_store, embedding_tx, knowledge_graph, knowledge_tx),
     )
     .await?;
 
@@ -96,12 +102,16 @@ pub fn spawn_raft_metrics_watcher(
 }
 
 pub async fn build_real_app_state(config: &Config) -> Result<Arc<AppState>, AppBuildError> {
-    let embedding_provider: Arc<dyn EmbeddingProvider> = match &config.openai_base_url {
-        Some(base_url) => Arc::new(OpenAIEmbedder::new_with_base_url(
-            config.openai_api_key.clone(),
-            base_url.clone(),
-        )?),
-        None => Arc::new(OpenAIEmbedder::new_with_api_key(config.openai_api_key.clone())?),
+    let embedding_provider: Arc<dyn EmbeddingProvider> = if config.openai_api_key.is_empty() {
+        Arc::new(crate::core::RandomEmbeddingProvider)
+    } else {
+        match &config.openai_base_url {
+            Some(base_url) => Arc::new(OpenAIEmbedder::new_with_base_url(
+                config.openai_api_key.clone(),
+                base_url.clone(),
+            )?),
+            None => Arc::new(OpenAIEmbedder::new_with_api_key(config.openai_api_key.clone())?),
+        }
     };
 
     build_app_state_with_embedding_provider(config, embedding_provider).await
@@ -136,13 +146,30 @@ pub async fn build_app_state_with_embedding_provider(
         config.embedding_max_concurrency,
     );
 
+    let knowledge_graph = Arc::new(tokio::sync::RwLock::new(KnowledgeGraph::new()));
+    let (knowledge_job_sender, knowledge_receiver) = knowledge_job_channel(config.knowledge_channel_size);
+
+    let knowledge_extractor: Arc<dyn crate::knowledge::extractor::KnowledgeExtractor> =
+        match config.knowledge_extractor {
+            KnowledgeExtractorType::Mock => Arc::new(MockKnowledgeExtractor),
+            KnowledgeExtractorType::OpenAI => match &config.openai_base_url {
+                Some(base_url) => Arc::new(OpenAIKnowledgeExtractor::new_with_base_url(
+                    config.openai_api_key.clone(),
+                    base_url.clone(),
+                )),
+                None => Arc::new(OpenAIKnowledgeExtractor::new(config.openai_api_key.clone())),
+            },
+        };
+
     let (raft, node_id, peer_http_addrs, raft_addr, raft_advertise_addr, cluster_peers) = if config.node_id.is_some() {
         let raft = build_raft_node(
             config,
             short_term_memory.clone(),
             core_memory_store.clone(),
             vector_store.clone(),
             embedding_job_sender.clone(),
+            knowledge_graph.clone(),
+            knowledge_job_sender.clone(),
         )
         .await
         .map_err(|e| AppBuildError::Other(e.into()))?;
@@ -160,6 +187,16 @@ pub async fn build_app_state_with_embedding_provider(
         spawn_raft_metrics_watcher(raft_handle.clone(), metrics.clone());
     }
 
+    let _knowledge_worker_handles = spawn_knowledge_workers(
+        knowledge_extractor,
+        raft.clone(),
+        config.node_id.unwrap_or(0),
+        knowledge_graph.clone(),
+        metrics.clone(),
+        knowledge_receiver,
+        config.knowledge_max_workers,
+    );
+
     Ok(Arc::new(AppState {
         short_term_memory,
         vector_store,
@@ -176,5 +213,7 @@ pub async fn build_app_state_with_embedding_provider(
         raft_addr,
         raft_advertise_addr,
         cluster_peers,
+        knowledge_graph,
+        knowledge_job_sender,
     }))
 }

src/cluster.rs

@@ -202,12 +202,17 @@ mod tests {
     async fn build_test_app_with_single_node_raft() -> TestServer {
         let c = build_test_components();
         let config = Config { node_id: Some(1), ..Config::default() };
+        let knowledge_graph = Arc::new(tokio::sync::RwLock::new(crate::knowledge::graph::KnowledgeGraph::new()));
+        let (knowledge_tx, mut knowledge_rx) = tokio::sync::mpsc::channel::<crate::knowledge::types::KnowledgeJob>(500);
+        tokio::spawn(async move { while knowledge_rx.recv().await.is_some() {} });
         let raft = build_raft_node(
             &config,
             c.short_term.clone(),
             c.core_memory.clone(),
             c.vector_store.clone(),
             c.embedding_job_sender.clone(),
+            knowledge_graph.clone(),
+            knowledge_tx.clone(),
         )
         .await
         .unwrap();
@@ -234,12 +239,16 @@ mod tests {
             raft_addr: Some("127.0.0.1:0".to_string()),
             raft_advertise_addr: None,
             cluster_peers: vec![],
+            knowledge_graph,
+            knowledge_job_sender: knowledge_tx,
         });
         TestServer::new(build_router(state)).unwrap()
     }
 
     fn build_test_app_standalone() -> TestServer {
         let c = build_test_components();
+        let (knowledge_job_sender, mut krx) = tokio::sync::mpsc::channel::<crate::knowledge::types::KnowledgeJob>(16);
+        tokio::spawn(async move { while krx.recv().await.is_some() {} });
         let state = Arc::new(AppState {
             short_term_memory: c.short_term,
             vector_store: c.vector_store,
@@ -256,6 +265,10 @@ mod tests {
             raft_addr: None,
             raft_advertise_addr: None,
             cluster_peers: vec![],
+            knowledge_graph: Arc::new(tokio::sync::RwLock::new(
+                crate::knowledge::graph::KnowledgeGraph::new(),
+            )),
+            knowledge_job_sender,
         });
         TestServer::new(build_router(state)).unwrap()
     }