feat: Engram sync layer — swarm memory protocol, peer delta sync, distributed activation

2026-04-27 17:18:51 -05:00
parent 2454c83e82
commit 6601761cd9
17 changed files with 6637 additions and 11 deletions
@@ -0,0 +1,218 @@
+/// Core Engram API routes — nodes, edges, activation, search.
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    Json,
+};
+use engram_core::types::{Edge, MemoryTier, Node, NodeType, RelationType};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use uuid::Uuid;
+
+use crate::state::AppState;
+
+// ── Stats ─────────────────────────────────────────────────────────────────────
+
+#[derive(Serialize)]
+pub struct StatsResponse {
+    pub nodes: usize,
+    pub edges: usize,
+}
+
+pub async fn get_stats(
+    State(state): State<Arc<AppState>>,
+) -> Result<Json<StatsResponse>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let nodes = db.node_count().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let edges = db.edge_count().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(Json(StatsResponse { nodes, edges }))
+}
+
+// ── Nodes ─────────────────────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct CreateNodeRequest {
+    pub node_type: NodeType,
+    pub embedding: Vec<f32>,
+    pub content: Vec<u8>,
+    pub tier: MemoryTier,
+    pub importance: f32,
+}
+
+#[derive(Serialize)]
+pub struct CreateNodeResponse {
+    pub id: Uuid,
+}
+
+pub async fn create_node(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<CreateNodeRequest>,
+) -> Result<Json<CreateNodeResponse>, StatusCode> {
+    let node = Node::new(req.node_type, req.embedding, req.content, req.tier, req.importance);
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let id = db.put_node(node).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(Json(CreateNodeResponse { id }))
+}
+
+pub async fn get_node(
+    State(state): State<Arc<AppState>>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Node>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    match db.get_node(id).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? {
+        Some(node) => Ok(Json(node)),
+        None => Err(StatusCode::NOT_FOUND),
+    }
+}
+
+// ── Edges ─────────────────────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct CreateEdgeRequest {
+    pub from_id: Uuid,
+    pub to_id: Uuid,
+    pub relation: RelationType,
+    pub weight: f32,
+}
+
+pub async fn create_edge(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<CreateEdgeRequest>,
+) -> Result<StatusCode, StatusCode> {
+    let edge = Edge::new(req.from_id, req.to_id, req.relation, req.weight);
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    db.put_edge(edge).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(StatusCode::CREATED)
+}
+
+pub async fn get_edges_from(
+    State(state): State<Arc<AppState>>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Vec<Edge>>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let edges = db.get_edges_from(id).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(Json(edges))
+}
+
+// ── Activation ────────────────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct ActivateRequest {
+    pub seeds: Vec<Uuid>,
+    pub query_embedding: Vec<f32>,
+    #[serde(default = "default_depth")]
+    pub max_depth: u8,
+    #[serde(default = "default_limit")]
+    pub limit: usize,
+}
+
+fn default_depth() -> u8 { 3 }
+fn default_limit() -> usize { 10 }
+
+#[derive(Serialize)]
+pub struct ActivateResponse {
+    pub results: Vec<ActivatedNodeJson>,
+}
+
+#[derive(Serialize)]
+pub struct ActivatedNodeJson {
+    pub node: Node,
+    pub activation_strength: f32,
+    pub hops: u8,
+}
+
+pub async fn activate(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<ActivateRequest>,
+) -> Result<Json<ActivateResponse>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let results = db
+        .activate(&req.seeds, &req.query_embedding, req.max_depth, req.limit)
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    Ok(Json(ActivateResponse {
+        results: results
+            .into_iter()
+            .map(|a| ActivatedNodeJson {
+                node: a.node,
+                activation_strength: a.activation_strength,
+                hops: a.hops,
+            })
+            .collect(),
+    }))
+}
+
+// ── Search ────────────────────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct SearchRequest {
+    pub embedding: Vec<f32>,
+    #[serde(default = "default_limit")]
+    pub limit: usize,
+}
+
+#[derive(Serialize)]
+pub struct SearchResponse {
+    pub results: Vec<ScoredNodeJson>,
+}
+
+#[derive(Serialize)]
+pub struct ScoredNodeJson {
+    pub node: Node,
+    pub score: f32,
+}
+
+pub async fn search_embedding(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<SearchRequest>,
+) -> Result<Json<SearchResponse>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let results = db
+        .search_embedding(&req.embedding, req.limit)
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    Ok(Json(SearchResponse {
+        results: results
+            .into_iter()
+            .map(|s| ScoredNodeJson { node: s.node, score: s.score })
+            .collect(),
+    }))
+}
+
+// ── Decay / Consolidate ───────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct DecayRequest {
+    pub factor: f32,
+}
+
+#[derive(Serialize)]
+pub struct DecayResponse {
+    pub nodes_updated: usize,
+}
+
+pub async fn decay(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<DecayRequest>,
+) -> Result<Json<DecayResponse>, StatusCode> {
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let nodes_updated = db.decay(req.factor).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(Json(DecayResponse { nodes_updated }))
+}
+
+#[derive(Serialize)]
+pub struct ConsolidateResponse {
+    pub promoted: usize,
+}
+
+pub async fn consolidate(
+    State(state): State<Arc<AppState>>,
+) -> Result<Json<ConsolidateResponse>, StatusCode> {
+    use engram_core::ConsolidationConfig;
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    let report = db
+        .consolidate(&ConsolidationConfig::default())
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    Ok(Json(ConsolidateResponse { promoted: report.promoted }))
+}
+
@@ -0,0 +1,3 @@
+pub mod core;
+pub mod sync;
+pub mod swarm;
@@ -0,0 +1,124 @@
+/// Swarm routes — distributed activation across the peer network.
+///
+/// POST /swarm/activate   — SwarmActivateRequest → SwarmActivateResponse
+/// GET  /swarm/status     — peer health check and last sync times
+use axum::{
+    extract::State,
+    http::StatusCode,
+    Json,
+};
+use engram_sync::{
+    client::SyncClient, merge_activation_results, Peer, PeerActivationResult, PeerStatus,
+    SerializableActivatedNode, SwarmActivateRequest, SwarmActivateResponse,
+};
+use std::sync::Arc;
+use uuid::Uuid;
+
+use crate::state::AppState;
+
+/// POST /swarm/activate
+///
+/// Runs spreading activation locally, then (if include_peers=true) fans out
+/// to all trusted peers in parallel and returns merged results.
+pub async fn swarm_activate(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<SwarmActivateRequest>,
+) -> Result<Json<SwarmActivateResponse>, StatusCode> {
+    // Step 1: Run local activation. Lock db, compute, drop immediately.
+    let local_results: Vec<SerializableActivatedNode> = {
+        let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+        let activated = db
+            .activate(&req.seeds, &req.query_embedding, req.max_depth, req.limit)
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+        activated.into_iter().map(Into::into).collect()
+        // db lock released here
+    };
+
+    // Step 2: Snapshot peer list (lock, read, drop).
+    let (our_id, trusted_peers): (Uuid, Vec<Peer>) = {
+        let engine = state.sync_engine.lock().await;
+        let id = engine.our_id();
+        let peers = engine.list_peers().iter().filter(|p| p.trusted).cloned().collect();
+        (id, peers)
+        // engine lock released here
+    };
+
+    // Step 3: Fan out to peers — no locks held across these awaits.
+    let mut peer_results: Vec<PeerActivationResult> = Vec::new();
+
+    if req.include_peers {
+        let mut handles = Vec::new();
+        for peer in trusted_peers {
+            let seeds = req.seeds.clone();
+            let embedding = req.query_embedding.clone();
+            let max_depth = req.max_depth;
+            let limit = req.limit;
+
+            handles.push(tokio::spawn(async move {
+                let client = SyncClient::new(peer.clone(), our_id);
+                match client.remote_activate(&seeds, &embedding, max_depth, limit).await {
+                    Ok(results) => PeerActivationResult {
+                        peer_id: peer.id,
+                        peer_name: peer.name,
+                        results,
+                        error: None,
+                    },
+                    Err(e) => PeerActivationResult {
+                        peer_id: peer.id,
+                        peer_name: peer.name,
+                        results: Vec::new(),
+                        error: Some(e.to_string()),
+                    },
+                }
+            }));
+        }
+
+        for handle in handles {
+            if let Ok(result) = handle.await {
+                peer_results.push(result);
+            }
+        }
+    }
+
+    // Step 4: Merge and return.
+    let merged = merge_activation_results(&local_results, &peer_results, req.limit);
+
+    Ok(Json(SwarmActivateResponse {
+        local_results,
+        peer_results,
+        merged,
+    }))
+}
+
+/// GET /swarm/status
+///
+/// Returns peer list with reachability status and last sync times.
+pub async fn swarm_status(
+    State(state): State<Arc<AppState>>,
+) -> Result<Json<Vec<PeerStatus>>, StatusCode> {
+    // Snapshot peer list and our_id without holding the lock across awaits.
+    let (our_id, peers): (Uuid, Vec<Peer>) = {
+        let engine = state.sync_engine.lock().await;
+        let id = engine.our_id();
+        let peers = engine.list_peers().to_vec();
+        (id, peers)
+    };
+
+    let mut statuses: Vec<PeerStatus> = Vec::new();
+    for peer in peers {
+        use engram_core::types::now_ms;
+        let client = SyncClient::new(peer.clone(), our_id);
+        let reachable = client.pull_delta(now_ms()).await.is_ok();
+        statuses.push(PeerStatus {
+            peer_id: peer.id,
+            peer_name: peer.name,
+            address: peer.address,
+            last_sync_at: peer.last_sync_at,
+            reachable,
+            sync_tiers: peer.sync_tiers,
+            trusted: peer.trusted,
+        });
+    }
+
+    Ok(Json(statuses))
+}
@@ -0,0 +1,133 @@
+/// Sync routes — peer delta exchange and peer registry.
+///
+/// All routes under /sync require Authorization: Bearer {api_key}.
+///
+/// GET  /sync/delta?since={ms}&peer_id={uuid}  — generate delta for caller
+/// POST /sync/push                              — receive delta from peer
+/// POST /sync/peers                            — register a new peer
+/// GET  /sync/peers                            — list peers
+/// DELETE /sync/peers/{id}                     — remove peer
+use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    Json,
+};
+use engram_core::types::MemoryTier;
+use engram_sync::{Peer, SyncDelta};
+use serde::Deserialize;
+use std::sync::Arc;
+use uuid::Uuid;
+
+use crate::state::AppState;
+
+// ── GET /sync/delta ───────────────────────────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct DeltaParams {
+    pub since: Option<i64>,
+    pub peer_id: Option<Uuid>,
+}
+
+pub async fn get_delta(
+    State(state): State<Arc<AppState>>,
+    Query(params): Query<DeltaParams>,
+) -> Result<Json<SyncDelta>, StatusCode> {
+    let since = params.since.unwrap_or(0);
+
+    // Determine which tiers to expose based on caller's peer_id
+    let tiers: Vec<MemoryTier> = {
+        let engine = state.sync_engine.lock().await;
+        if let Some(peer_id) = params.peer_id {
+            if let Some(peer) = engine.get_peer(peer_id) {
+                if peer.trusted {
+                    peer.sync_tiers.clone()
+                } else {
+                    vec![MemoryTier::Semantic]
+                }
+            } else {
+                vec![MemoryTier::Semantic]
+            }
+        } else {
+            vec![MemoryTier::Semantic]
+        }
+        // engine lock released here
+    };
+
+    let engine = state.sync_engine.lock().await;
+    let delta = engine
+        .generate_delta(since, &tiers)
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    Ok(Json(delta))
+}
+
+// ── POST /sync/push ───────────────────────────────────────────────────────────
+
+pub async fn push_delta(
+    State(state): State<Arc<AppState>>,
+    Json(delta): Json<SyncDelta>,
+) -> Result<StatusCode, StatusCode> {
+    // Accepted tiers for incoming pushes (Semantic and Procedural by default)
+    let accepted_tiers = vec![MemoryTier::Semantic, MemoryTier::Procedural];
+
+    // Apply the delta using the DB handle directly (no async needed)
+    let db = state.db.lock().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    // Tombstones
+    for id in &delta.tombstones {
+        let _ = db.delete_node(*id);
+    }
+
+    // Nodes
+    for node in delta.nodes {
+        if !accepted_tiers.contains(&node.tier) {
+            continue;
+        }
+        if db.get_node(node.id).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?.is_some() {
+            continue;
+        }
+        db.put_node(node).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+    }
+
+    // Edges
+    for edge in delta.edges {
+        let from_ok = db.get_node(edge.from_id)
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
+            .is_some();
+        let to_ok = db.get_node(edge.to_id)
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
+            .is_some();
+        if from_ok && to_ok {
+            let _ = db.put_edge(edge);
+        }
+    }
+
+    Ok(StatusCode::OK)
+}
+
+// ── Peer registry ─────────────────────────────────────────────────────────────
+
+pub async fn list_peers(
+    State(state): State<Arc<AppState>>,
+) -> Result<Json<Vec<Peer>>, StatusCode> {
+    let engine = state.sync_engine.lock().await;
+    Ok(Json(engine.list_peers().to_vec()))
+}
+
+pub async fn register_peer(
+    State(state): State<Arc<AppState>>,
+    Json(peer): Json<Peer>,
+) -> Result<StatusCode, StatusCode> {
+    let mut engine = state.sync_engine.lock().await;
+    engine.add_peer(peer);
+    Ok(StatusCode::CREATED)
+}
+
+pub async fn delete_peer(
+    State(state): State<Arc<AppState>>,
+    Path(id): Path<Uuid>,
+) -> Result<StatusCode, StatusCode> {
+    let mut engine = state.sync_engine.lock().await;
+    engine.remove_peer(id);
+    Ok(StatusCode::NO_CONTENT)
+}