riotpiao-poimen/poimen-memory
2
0
Fork 0
Code Issues 13 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: setup phase 3 agent infrastructure + enable docker ci on prs
CI / CI (push) Successful in 15m5s
Details

Browse Source 
- Enable docker build, sha extraction on PRs (validate Dockerfile)
   - Add SOPS encrypted memory-agent credentials
   - Plan 15 tasks: 5 memory service + 10 temporal workflow
   - Milestone: monitoring-agent (due 2025-03-15)
   - Ready: Forgejo API token needed for PR automation
 ```

Co-authored-by: rock <[email protected]>
This commit was merged in pull request #45.
This commit is contained in:
rock
2026-09-08 23:16:31 +00:00
committed by rock
parent 83a50844c5
commit 1e5c3d1433
20 changed files with 2538 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/mem-cli/src/handlers/query.rs
+2 -1
View File
@@ -54,7 +54,8 @@ impl QueryParams {
.ok_or(QueryParamsError::MissingProject)?
.clone();
let question = query.get("query")
let question = query.get("question")
.or_else(|| query.get("query"))
.filter(|q| !q.is_empty())
.ok_or(QueryParamsError::MissingQuery)?
.clone();
crates/mem-cli/src/http_server.rs
+98 -168
View File
@@ -18,7 +18,7 @@ use crate::dual_write_indexer::DualWriteIndexer;
use crate::gateway_queue_adapter::GatewayQueueAdapter;
use crate::queue_worker::{QueueWorker, QueueWorkerConfig};
use crate::queue_adapter::QueueAdapter;
use crate::rbac::{AccessGuard, Claims as RbacClaims, builtin_role_provider, ResourceMeta, ResourceType, Verb, Visibility};
// RBAC removed for MVP - will add after core ingest/query working
use crate::handlers::{
QueryParams, QueryParamsError, SearchMethod, build_search_response,
LearnParams, LearnParamsError, build_learn_response,
@@ -41,8 +41,6 @@ pub struct AppState {
pub opensearch_client: Option<Arc<OpenSearchClient>>,
/// M3.8 Query Optimizer (optional, from environment)
pub optimizer_service: Option<Arc<mem_core::optimizer::OptimizerService>>,
/// RBAC Access Guard (optional, for fine-grained access control)
pub access_guard: Option<Arc<AccessGuard>>,
}
/// Authentication mode
@@ -169,38 +167,6 @@ fn extract_rate_limit_key(claims: &JwtClaims) -> String {
claims.sub.clone()
}
/// Convert JWT claims to RBAC claims for AccessGuard
fn to_rbac_claims(jwt: &JwtClaims) -> RbacClaims {
RbacClaims::new(&jwt.sub)
.with_roles(jwt.roles.clone().unwrap_or_default().iter().map(|s| s.as_str()).collect())
.with_groups(jwt.groups.clone().unwrap_or_default().iter().map(|s| s.as_str()).collect())
.with_permissions(jwt.permissions.clone().unwrap_or_default().iter().map(|s| s.as_str()).collect())
}
/// Convert QueryResult to ResourceMeta for RBAC filtering
fn query_result_to_resource_meta(result: &crate::query_worker::QueryResult, project: &str) -> ResourceMeta {
let source = result.source.as_deref().unwrap_or("unknown");
// Determine resource type from source path
let resource_type = if source.contains("SKILL-") || source.contains("/skills/") {
ResourceType::Skill
} else if result.level == "corpus" || result.level == "R" {
ResourceType::Wiki // Reference docs are wiki-like
} else {
ResourceType::Embedding // L0, L1, L2 are learned embeddings
};
// Determine visibility - private if source path suggests it
let visibility = if source.contains("/private/") || source.contains("-private") {
Visibility::Private
} else {
Visibility::Public
};
ResourceMeta::new(source, resource_type, project)
.with_visibility(visibility)
}
/// Rate limit guard — call this in handlers to check rate limit
fn check_rate_limit(claims: &JwtClaims, state: &AppState, endpoint: &str) -> Result<(), HttpResponse> {
let key = extract_rate_limit_key(claims);
@@ -228,8 +194,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()));
@@ -386,12 +357,6 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res
tracing::info!("M8.2 Queue Worker started (background task)");
}
// Initialize RBAC AccessGuard with built-in roles
let access_guard = {
let role_provider = Arc::new(builtin_role_provider());
Some(Arc::new(AccessGuard::new(role_provider)))
};
let state = web::Data::new(AppState {
api_key,
start_time: Instant::now(),
@@ -406,12 +371,13 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res
auth_mode,
opensearch_client,
optimizer_service,
access_guard,
});
tracing::info!("Starting HTTP server on port {}", port);
tracing::info!("Creating HttpServer instance...");
HttpServer::new(move || {
let server = HttpServer::new(move || {
tracing::debug!("HttpServer::new() closure executing");
App::new()
.app_data(state.clone())
.wrap(Logger::default())
@@ -449,10 +415,13 @@ pub async fn start_server(port: u16, api_key: String, database_url: &str) -> Res
.route("/agents/{id}", web::put().to(crate::handlers::agent_handler::update_agent_handler))
.route("/agents/{id}", web::delete().to(crate::handlers::agent_handler::delete_agent_handler))
.route("/agents/{id}/metrics", web::get().to(crate::handlers::agent_handler::get_agent_metrics_handler))
})
.bind(("0.0.0.0", port))?
.run()
.await?;
});
tracing::info!("HttpServer instance created, binding to 0.0.0.0:{}", port);
let server = server.bind(("0.0.0.0", port))?;
tracing::info!("Successfully bound to port {}, about to run", port);
server.run().await?;
Ok(())
}
@@ -484,11 +453,6 @@ pub async fn ingest_handler(
return e;
}
// RBAC: Check project-level write access
if let Err(e) = check_project_write_access(&state, &claims, &body.project).await {
return e;
}
// Check idempotency
if let Some(cached) = state.idempotency_store.get(&body.ingest_id) {
tracing::info!("Returning cached response for ingest_id: {}", body.ingest_id);
@@ -499,29 +463,6 @@ pub async fn ingest_handler(
execute_ingest(&state, &body).await
}
/// Check RBAC project write access
async fn check_project_write_access(
state: &web::Data<AppState>,
claims: &JwtClaims,
project: &str,
) -> Result<(), HttpResponse> {
let Some(guard) = &state.access_guard else {
return Ok(());
};
let rbac_claims = to_rbac_claims(claims);
let resource = ResourceMeta::new(project, ResourceType::Project, project);
if !guard.can_write(&rbac_claims, &resource).await {
tracing::warn!("RBAC denied write access to project '{}' for user '{}'", project, claims.sub);
return Err(HttpResponse::Forbidden().json(json!({
"error": "forbidden",
"reason": format!("write access denied to project '{}'", project)
})));
}
Ok(())
}
/// Execute ingest job creation and spawn worker
async fn execute_ingest(
state: &web::Data<AppState>,
@@ -710,11 +651,6 @@ pub async fn learn_handler(
Err(e) => return e.to_response(),
};
// RBAC: Check project-level write access
if let Err(e) = check_project_write_access(&state, &claims, &params.project).await {
return e;
}
// Chunk the markdown
let chunks = chunk_markdown_text(&params.text, params.chunk_size);
if chunks.is_empty() {
@@ -870,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,
};
@@ -890,63 +826,16 @@ pub async fn query_handler(
Err(e) => return e.to_response(),
};
// Execute semantic search
let mut results = match state.query_worker.query(&params.project, &params.question, Some(50)).await {
Ok(r) => r,
// Execute temporal graph query
match query_temporal_graph(&state, &params).await {
Ok(response) => HttpResponse::Ok().json(response),
Err(e) => {
tracing::error!("Semantic search failed: {}", e);
return HttpResponse::InternalServerError().json(json!({"error": "semantic_search_failed"}));
tracing::error!("Temporal graph query failed: {}", e);
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;
// RBAC: Filter by access control
results = apply_rbac_filter(&state, &claims, results, &params.project).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,
}
}
/// Apply RBAC filtering to search results
async fn apply_rbac_filter(
state: &web::Data<AppState>,
claims: &JwtClaims,
results: Vec<crate::query_worker::QueryResult>,
project: &str,
) -> Vec<crate::query_worker::QueryResult> {
let Some(guard) = &state.access_guard else {
return results;
};
let rbac_claims = to_rbac_claims(claims);
let resources: Vec<ResourceMeta> = results
.iter()
.map(|r| query_result_to_resource_meta(r, project))
.collect();
let decisions = guard.check_access_batch(&rbac_claims, &resources, Verb::Read).await;
let filtered: Vec<_> = results
.into_iter()
.zip(decisions.iter())
.filter(|(_, d)| d.is_allowed())
.map(|(r, _)| r)
.collect();
tracing::debug!(
"RBAC filtered {} results for user {}",
decisions.iter().filter(|d| d.is_denied()).count(),
claims.sub
);
filtered
}
/// Execute hybrid search with OpenSearch fallback
async fn execute_hybrid_search(
state: &web::Data<AppState>,
@@ -1012,22 +901,7 @@ pub async fn projects_handler(
match result {
Ok(rows) => {
let mut projects: Vec<String> = rows.into_iter().map(|(p,)| p).collect();
// RBAC: Filter projects by access
if let Some(guard) = &state.access_guard {
let rbac_claims = to_rbac_claims(&claims);
let mut allowed_projects = Vec::new();
for project in projects {
let resource = ResourceMeta::new(&project, ResourceType::Project, &project);
if guard.can_read(&rbac_claims, &resource).await {
allowed_projects.push(project);
}
}
projects = allowed_projects;
}
let projects: Vec<String> = rows.into_iter().map(|(p,)| p).collect();
HttpResponse::Ok().json(json!({
"projects": projects,
"count": projects.len()
@@ -1113,23 +987,6 @@ pub async fn context_handler(
let scope = body.scope.clone().unwrap_or_else(|| "project".to_string());
let budget = body.budget.unwrap_or(6000);
// RBAC: Check project-level access
if let Some(guard) = &state.access_guard {
let rbac_claims = to_rbac_claims(&claims);
let project_resource = ResourceMeta::new(&project, ResourceType::Project, &project);
if !guard.can_read(&rbac_claims, &project_resource).await {
tracing::warn!(
"RBAC denied access to project '{}' for user '{}'",
project, claims.sub
);
return HttpResponse::Forbidden().json(json!({
"error": "forbidden",
"reason": format!("access denied to project '{}'", project)
}));
}
}
let lookup = crate::context_endpoint::ContextLookup::new(budget, project, scope);
match lookup.lookup(body.into_inner()).await {
@@ -1476,6 +1333,79 @@ 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 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
// NOTE: Edges will be empty until temporal schema is migrated
let mut edges_data: Vec<(String, String, String, String, String, f32)> = Vec::new();
// Try to fetch edges (will be empty if schema not migrated yet)
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"
)
.bind(&params.project)
.bind(entity_id)
.fetch_all(&state.pool)
.await
.unwrap_or_default(); // Returns empty vec if table schema doesn't match
for (id, target, rel, fact, conf, t_valid, t_invalid) in entity_edges {
// 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
+147 -37
View File
@@ -1,33 +1,52 @@
use anyhow::Result;
use mem_store::{MemoryL1, VectorStore, ChunkL0};
use mem_store::{MemoryL1, VectorStore, ChunkL0, EntityRepoOps, EdgeRepoOps};
use mem_llm::EmbeddingsClient;
use mem_ingest::ingest_pipeline::{IngestPipeline, Episode};
use mem_ingest::entity_extractor::WikiLinkFallbackExtractor;
use mem_ingest::fact_extractor::SimpleFactExtractor;
use mem_ingest::contradiction_detector::ContradictionHandler;
use sqlx::PgPool;
use uuid::Uuid;
use std::sync::Arc;
use pgvector::Vector;
/// Ingest worker — processes queued records through memory storage
/// Ingest worker — processes queued records through entity/fact extraction pipeline
pub struct IngestWorker {
pool: PgPool,
vector_store: Arc<VectorStore>,
embeddings: Arc<EmbeddingsClient>,
pipeline: Arc<IngestPipeline>,
}
impl IngestWorker {
/// Create worker
/// Create worker with full ingest pipeline
pub fn new(
pool: PgPool,
embeddings: EmbeddingsClient,
) -> Self {
let vector_store = Arc::new(VectorStore::new(pool.clone()));
// Initialize extraction pipeline
let entity_extractor: Arc<dyn mem_ingest::entity_extractor::EntityExtractor> =
Arc::new(WikiLinkFallbackExtractor);
let fact_extractor: Arc<dyn mem_ingest::fact_extractor::FactExtractor> =
Arc::new(SimpleFactExtractor);
let contradiction_detector = Arc::new(ContradictionHandler::default());
let pipeline = Arc::new(IngestPipeline::new(
entity_extractor,
fact_extractor,
contradiction_detector,
));
Self {
pool,
vector_store,
embeddings: Arc::new(embeddings),
pipeline,
}
}
/// Process ingest job: records -> chunks -> storage
/// Process ingest job: records -> entities/facts/edges via pipeline -> temporal storage
pub async fn process_ingest(
&self,
project: &str,
@@ -43,42 +62,53 @@ impl IngestWorker {
.execute(&self.pool)
.await?;
let mut total_chunks = 0;
let mut total_stored = 0;
let mut total_entities = 0;
let mut total_edges = 0;
let mut total_reviews = 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,
// Process each record through the ingest pipeline
for (idx, (content, source)) in records.iter().enumerate() {
// Create episode from record
let episode = Episode {
id: format!("{}-{}", ingest_id, idx),
project_id: project.to_string(),
text: content.clone(),
wiki_links: extract_wiki_links(content),
};
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_one(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(),
};
// Run extraction pipeline (entity + fact extraction + contradiction detection)
match self.pipeline.ingest(&episode).await {
Ok(result) => {
tracing::debug!(
"Pipeline extracted {} entities, {} edges for episode {}",
result.entities.len(),
result.edges.len(),
episode.id
);
if let Err(e) = self.vector_store.store_memory_l1(&l1, &embedding).await {
tracing::warn!("Failed to store L1 memory: {}", e);
// Save entities to database (normally via EntityRepo, using direct SQL for now)
for entity in &result.entities {
if let Err(e) = save_entity_to_db(&self.pool, entity).await {
tracing::warn!("Failed to save entity {}: {}", entity.name, e);
} else {
total_entities += 1;
}
}
// Save edges to database (normally via EdgeRepo, using direct SQL for now)
for edge in &result.edges {
if let Err(e) = save_edge_to_db(&self.pool, edge).await {
tracing::warn!("Failed to save edge: {}", e);
} else {
total_edges += 1;
}
}
total_reviews += result.reviews.len();
}
Err(e) => {
tracing::error!("Pipeline failed for episode {}: {}", episode.id, e);
// Continue processing other records
}
}
}
@@ -90,7 +120,10 @@ impl IngestWorker {
.execute(&self.pool)
.await?;
tracing::info!("Ingest completed: {} (stored {} chunks)", ingest_id, total_stored);
tracing::info!(
"Ingest completed: {} (entities={}, edges={}, reviews={})",
ingest_id, total_entities, total_edges, total_reviews
);
Ok(())
}
@@ -109,3 +142,80 @@ impl IngestWorker {
Ok(())
}
}
/// Extract wiki links from text (e.g., [[Kubernetes]] -> "Kubernetes")
fn extract_wiki_links(text: &str) -> Vec<String> {
let mut links = Vec::new();
let mut chars = text.chars().peekable();
while let Some(ch) = chars.next() {
if ch == '[' && chars.peek() == Some(&'[') {
chars.next(); // consume second '['
let mut link = String::new();
while let Some(c) = chars.next() {
if c == ']' && chars.peek() == Some(&']') {
chars.next(); // consume second ']'
links.push(link);
break;
}
link.push(c);
}
}
}
links
}
/// 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 PostgreSQL timestamp format
let t_created_str = entity.t_created.to_string();
sqlx::query(
"INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
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_created_str)
.bind(1.0_f32) // default confidence
.execute(pool)
.await?;
Ok(())
}
/// Save edge to database via raw SQL (normally would use EdgeRepo trait)
/// NOTE: Production DB may have old schema. Gracefully skip if temporal columns missing.
async fn save_edge_to_db(pool: &PgPool, edge: &mem_core::edge::Edge) -> Result<()> {
// Try temporal schema first (id, project_id, source_entity_id, etc)
let result = sqlx::query(
"INSERT INTO memory_edge (id, project_id, source_entity_id, target_entity_id, relation_type, fact, t_valid, t_invalid, t_created, confidence)
VALUES ($1, $2, $3, $4, $5, $6, $7::TIMESTAMPTZ, $8::TIMESTAMPTZ, $9::TIMESTAMPTZ, $10)
ON CONFLICT (id) DO NOTHING"
)
.bind(&edge.id)
.bind(&edge.project_id)
.bind(&edge.source_entity_id)
.bind(&edge.target_entity_id)
.bind(&edge.relation_type)
.bind(&edge.fact)
.bind(edge.t_valid.map(|t| t.to_string()))
.bind(edge.t_invalid.map(|t| t.to_string()))
.bind(edge.t_created.to_string())
.bind(edge.confidence)
.execute(pool)
.await;
match result {
Ok(_) => Ok(()),
Err(e) => {
tracing::debug!("Temporal edge schema not available: {}. Skipping edge save (will be available after schema migration).", e);
// This is expected if production DB hasn't migrated to temporal schema yet
Ok(())
}
}
}