# Memory Wiki-Graph RAG Optimization **Goal:** Query routes via wiki-link graph → project-scoped TF-IDF + semantic search → minimal LLM calls with maximum relevance. --- ## Architecture Overview ``` ┌─────────────────────────────────────────────────────────────────────┐ │ Query Input │ │ "How to debug pod CrashLoopBackOff?" │ └────────────────────────────┬────────────────────────────────────────┘ │ ┌──────────────▼───────────────┐ │ Wiki-Link Graph Lookup │ │ │ │ project:poimen │ │ → tools/kubectl.md │ │ [[debugging.md]] │ │ [[root-cause.md]] │ │ │ │ Scope defined by graph │ └──────────────┬───────────────┘ │ ┌────────────────────┼────────────────────┐ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────────┐ ┌───────────────┐ │ Project- │ │ Chunk-level │ │ Shared Skills │ │ scoped │ │ TF-IDF │ │ (wiki-links) │ │ TF-IDF │ │ │ │ │ │ │ │ metadata: │ │ SKILL-k8s- │ │ Index: │ │ {doc_id, │ │ debugging │ │ poimen/ │ │ term_freq,│ │ │ │ tools/ │ │ idf} │ │ Backlinks to │ │ kubectl │ │ │ │ all projects │ │ │ │ Prioritize: │ │ using it │ │ Rank by: │ │ • Exact tool │ │ │ │ TF-IDF │ │ • Error type │ │ Deduplicate: │ │ + recency│ │ • Solution │ │ one SKILL, │ │ + hits │ │ │ │ many projects │ └──────────┘ └──────────────┘ └───────────────┘ │ │ │ └────────────────────┼────────────────────┘ │ ┌──────────────▼───────────────┐ │ Semantic Search │ │ (pgvector cosine sim) │ │ │ │ Only search within scoped │ │ doc_ids from TF-IDF │ │ │ │ Parallel execution: │ │ • Query embedding │ │ • Chunk embeddings (cached) │ │ • Cosine similarity │ └──────────────┬───────────────┘ │ ┌──────────────▼───────────────┐ │ RRF Fusion │ │ │ │ TF-IDF score + Semantic score│ │ → fused_rank │ │ │ │ score = 0.4 * tfidf_norm + │ │ 0.6 * semantic_norm │ └──────────────┬───────────────┘ │ ┌──────────────▼───────────────┐ │ Chunk Filtering │ │ │ │ • Threshold: score > 0.7 │ │ • Limit: top-10 chunks │ │ • Dedup: related chunks │ │ (shingle-based) │ └──────────────┬───────────────┘ │ ┌──────────────▼───────────────┐ │ Budget Verification │ │ │ │ total_tokens = sum(chunk) │ │ if > budget: drop lowest │ │ score chunks │ └──────────────┬───────────────┘ │ ┌──────────────▼───────────────┐ │ LLM Call (Optimized) │ │ │ │ Context = selected chunks │ │ No need to search full vault │ │ Only ~3-5 chunks per call │ │ → 70-80% fewer LLM calls │ └──────────────────────────────┘ ``` --- ## Phase 1: Wiki-Link Graph Indexing ### Schema ```sql -- Wiki-link graph (relationships between docs) CREATE TABLE wiki_links ( id BIGSERIAL PRIMARY KEY, project VARCHAR(256), -- "poimen" or NULL for shared source_path VARCHAR(1024), -- "tools/kubectl.md" target_path VARCHAR(1024), -- "debugging.md" or "shared:skills/SKILL-*" link_type VARCHAR(32), -- "memory", "skill", "concept", "tool" created_at TIMESTAMP DEFAULT NOW(), UNIQUE(project, source_path, target_path) ); -- Project-scoped index metadata CREATE TABLE project_index_metadata ( id BIGSERIAL PRIMARY KEY, project VARCHAR(256), root_path VARCHAR(1024), -- entry point (e.g., "tools/kubectl.md") doc_count INT, link_count INT, indexed_at TIMESTAMP DEFAULT NOW(), cache_hit_ratio FLOAT -- KV cache optimization metric ); ``` ### Ingestion: Wiki-Link Parser ```rust pub struct WikiLinkParser { project: String, vault_root: PathBuf, } impl WikiLinkParser { pub async fn parse_file(&self, path: &Path) -> Result> { // Extract [[links]] from markdown let content = tokio::fs::read_to_string(path).await?; let regex = Regex::new(r"\[\[([^\]]+)\]\]")?; let links = regex .captures_iter(&content) .map(|cap| { let target = cap[1].trim(); WikiLink { source_path: path.to_str().unwrap().to_string(), target_path: target.to_string(), link_type: self.infer_link_type(target), project: self.project.clone(), } }) .collect(); Ok(links) } fn infer_link_type(&self, target: &str) -> String { if target.contains("SKILL-") { "skill".to_string() } else if target.contains("shared:") { "shared_concept".to_string() } else if target.ends_with(".md") { "memory".to_string() } else { "unknown".to_string() } } } ``` --- ## Phase 2: Multi-Scope TF-IDF Indexing ### TF-IDF Scopes ``` Scope 1: Global (all docs) → vocabulary for "rare term" detection → used as fallback when project-scoped finds nothing Scope 2: Project-scoped (poimen/* only) → primary index for queries within project → higher weight for project-specific terms Scope 3: Chunk-level metadata → extract key terms from chunk headers, first sentence → high IDF = rare term = strong signal ``` ### Data Structure ```rust pub struct TfIdfIndex { pub global: BTreeMap, // term → idf pub project_scope: HashMap, // project → {term → tf} pub chunk_metadata: HashMap>, // doc_id → [term, ...] } pub struct TermStats { pub idf: f32, // log(total_docs / docs_with_term) pub global_freq: u32, pub project_freqs: HashMap, } pub struct ProjectIndex { pub project: String, pub docs: HashMap, // doc_id → term_freq pub idf: BTreeMap, // term → project-local idf } pub struct Term { pub text: String, pub tf: f32, pub idf: f32, pub category: String, // "error", "tool", "concept", "solution" } ``` ### Scoring Algorithm ```rust pub fn score_chunk( query: &str, chunk_id: &str, project: &str, tfidf: &TfIdfIndex, ) -> f32 { let mut score = 0.0; // 1. Project-scoped TF-IDF let project_idx = &tfidf.project_scope.get(project).unwrap(); for term in tokenize(query) { let tf = project_idx.docs .get(chunk_id) .copied() .unwrap_or(0.0); let idf = project_idx.idf .get(&term) .copied() .unwrap_or(0.1); // smoothing score += tf * idf; } // 2. Chunk-level metadata boost let chunk_terms = tfidf.chunk_metadata.get(chunk_id).unwrap_or(&vec![]); for term in chunk_terms { if query.contains(&term.text) { // Exact match in chunk metadata → big boost score += term.idf * 3.0; } } // 3. Recency + hit count (if available) let recency_weight = 0.1; // newer = higher score score *= (1.0 + recency_weight); score } ``` ### Indexing Pipeline ```rust pub struct TfIdfBuilder { vault_root: PathBuf, } impl TfIdfBuilder { pub async fn build_indexes(self) -> Result { // 1. Scan all markdown files let files = self.scan_vault().await?; // 2. Extract terms per project let mut project_docs: HashMap> = HashMap::new(); for file in &files { let project = self.extract_project(&file)?; let terms = self.extract_terms(&file).await?; project_docs.entry(project).or_insert_with(Vec::new).push(terms); } // 3. Compute IDF per project let mut index = TfIdfIndex::default(); for (project, docs) in project_docs { let project_idx = self.compute_project_idf(project, docs)?; index.project_scope.insert(project, project_idx); } // 4. Global IDF (for fallback) index.global = self.compute_global_idf(&files)?; Ok(index) } async fn extract_terms(&self, file: &Path) -> Result { let content = tokio::fs::read_to_string(file).await?; // Extract heading + first 100 chars as high-value terms let lines: Vec<&str> = content.lines().collect(); let mut terms = String::new(); for line in lines { if line.starts_with("##") || line.starts_with("###") { terms.push_str(&line.replace('#', " ")); terms.push(' '); } } Ok(terms) } } ``` --- ## Phase 3: Hybrid Retrieval (Wiki-Nav + TF-IDF + Semantic) ### Query Router ```rust pub struct QueryRouter { wiki_index: WikiLinkIndex, tfidf: TfIdfIndex, embeddings: EmbeddingsClient, } impl QueryRouter { pub async fn route_query( &self, query: &str, project: &str, ) -> Result { // Step 1: Wiki-link navigation let scoped_docs = self.wiki_index .reachable_docs(project) // all docs in project + shared/ .await?; // Step 2: Project-scoped TF-IDF pre-filter let tfidf_candidates: Vec<_> = scoped_docs .iter() .map(|doc_id| { let score = score_chunk(query, doc_id, project, &self.tfidf); (doc_id.clone(), score) }) .filter(|(_, score)| score > &0.1) // threshold .collect(); // Step 3: Semantic search (only on TF-IDF candidates) let query_embedding = self.embeddings.embed(query).await?; let semantic_results = self.semantic_search( &query_embedding, &tfidf_candidates.iter().map(|(id, _)| id.clone()).collect::>() ).await?; // Step 4: RRF Fusion let fused = self.rrf_fusion(&tfidf_candidates, &semantic_results)?; Ok(RetrievalPlan { candidates: fused.into_iter().take(10).collect(), project: project.to_string(), search_strategy: "wiki-nav + tfidf + semantic".to_string(), }) } } ``` ### RRF Fusion ```rust fn rrf_fusion( tfidf_results: &[(String, f32)], semantic_results: &[(String, f32)], ) -> Result> { let mut scores: HashMap = HashMap::new(); // TF-IDF contribution (40%) for (rank, (doc_id, score)) in tfidf_results.iter().enumerate() { let normalized = 1.0 / (rank as f32 + 1.0); *scores.entry(doc_id.clone()).or_insert(0.0) += 0.4 * normalized; } // Semantic contribution (60%) for (rank, (doc_id, score)) in semantic_results.iter().enumerate() { let normalized = 1.0 / (rank as f32 + 1.0); *scores.entry(doc_id.clone()).or_insert(0.0) += 0.6 * normalized; } let mut result: Vec<_> = scores.into_iter().collect(); result.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap()); Ok(result) } ``` --- ## Phase 4: LLM Call Optimization ### Chunk Selection Strategy ```rust pub struct ChunkSelector { vault_root: PathBuf, } impl ChunkSelector { pub async fn select_chunks( &self, candidates: &[(String, f32)], // doc_ids + scores from Phase 3 budget: usize, // token budget ) -> Result> { let mut selected = Vec::new(); let mut token_count = 0; for (doc_id, score) in candidates { if token_count >= budget { break; } let path = self.doc_id_to_path(doc_id)?; let content = tokio::fs::read_to_string(&path).await?; let tokens = self.count_tokens(&content); // Only select if score is high enough if *score > 0.6 && token_count + tokens < budget { selected.push(doc_id.clone()); token_count += tokens; } } Ok(selected) } } pub struct LlmCallOptimizer { chunk_selector: ChunkSelector, } impl LlmCallOptimizer { pub async fn minimize_calls( &self, query: &str, candidates: &[(String, f32)], budget: usize, ) -> Result { // Select top chunks let chunks = self.chunk_selector.select_chunks(candidates, budget).await?; // Metrics let call_reduction = 100.0 * (1.0 - chunks.len() as f32 / candidates.len() as f32); Ok(MinimizedCallPlan { chunks: chunks.clone(), token_budget: budget, tokens_used: self.estimate_tokens(&chunks)?, estimated_call_reduction_pct: call_reduction, strategy: format!( "Wiki-scoped + TF-IDF pre-filter + semantic re-rank → {} chunks ({}% reduction)", chunks.len(), call_reduction as u32 ), }) } } ``` ### Metrics ``` Before optimization: Query "How to debug pod?" → search all vault → 100+ candidates → pass all to LLM LLM calls: 5-10 (per search result) After optimization (Wiki-Graph RAG): Query "How to debug pod?" → Wiki scope: poimen/tools/* only (30 docs) → TF-IDF filter: 5 docs → Semantic re-rank: top 3 → LLM calls: 1-2 Reduction: 70-80% Quality: higher (only relevant docs reach LLM) Latency: 50-100ms (pre-filtering) + semantic search time ``` --- ## Phase 5: Chunk-Level Metadata Index ### Chunk Metadata Extraction ```rust pub struct ChunkMetadata { pub doc_id: String, pub heading: String, // ## heading pub first_sentence: String, // first 20 words pub key_terms: Vec, // extracted terms pub category: String, // "error", "solution", "tool", "concept" } pub struct MetadataExtractor; impl MetadataExtractor { pub fn extract(content: &str) -> Result { let lines: Vec<&str> = content.lines().collect(); // Find heading let heading = lines .iter() .find(|l| l.starts_with("##")) .map(|l| l.to_string()) .unwrap_or_default(); // Extract first sentence let first_sentence = lines .iter() .find(|l| !l.is_empty() && !l.starts_with("#")) .map(|l| l.to_string()) .unwrap_or_default(); // Extract key terms (nouns, verbs) let key_terms = self.extract_key_terms(&heading, &first_sentence)?; // Infer category let category = self.infer_category(&heading, &key_terms)?; Ok(ChunkMetadata { doc_id: "...".to_string(), heading, first_sentence, key_terms, category, }) } fn infer_category(&self, heading: &str, terms: &[String]) -> Result { if heading.contains("Error") || heading.contains("error") { Ok("error".to_string()) } else if heading.contains("How") || heading.contains("Fix") { Ok("solution".to_string()) } else if heading.contains("kubectl") || heading.contains("cargo") { Ok("tool".to_string()) } else { Ok("concept".to_string()) } } } ``` ### Chunk-Level Scoring Boost ```rust fn score_chunk_with_metadata( query: &str, chunk_id: &str, metadata: &ChunkMetadata, tfidf: &TfIdfIndex, ) -> f32 { let mut score = score_chunk(query, chunk_id, &tfidf); // Boost for exact key term match for key_term in &metadata.key_terms { if query.to_lowercase().contains(&key_term.to_lowercase()) { score *= 1.5; // 50% boost for key term match } } // Boost for category alignment if metadata.category == "solution" && query.contains("how") { score *= 1.3; } if metadata.category == "error" && query.contains("error") { score *= 1.3; } score } ``` --- ## Phase 6: Cache Alignment & KV Cache Optimization ### KV Cache Hit Ratio Tracking ```rust pub struct CacheMetrics { pub doc_id: String, pub project: String, pub accessed_count: u32, // times used in queries pub total_tokens: u32, pub cache_efficiency: f32, // accessed_count / total_tokens } pub async fn update_cache_metrics( doc_id: &str, project: &str, db: &PostgresPool, ) -> Result<()> { sqlx::query( "UPDATE project_index_metadata SET cache_hit_ratio = accessed_count / total_tokens WHERE project = $1" ) .bind(project) .execute(db) .await?; Ok(()) } ``` ### Wiki-Link Graph Ordering (Cache-Aligned) When traversing wiki-links, fetch documents in order of: 1. **Same chunk** (already in cache) 2. **Adjacent chunks** (likely prefetched) 3. **Same document** (same KV cache line) 4. **Related documents** (via wiki-link proximity) ```rust pub struct CacheAlignedTraversal { current_chunk: String, wiki_links: Vec<(String, f32)>, // (target, relevance) } impl CacheAlignedTraversal { pub fn prioritize_by_cache_locality(&mut self) { self.wiki_links.sort_by(|a, b| { let a_dist = self.cache_distance(&a.0); let b_dist = self.cache_distance(&b.0); a_dist.partial_cmp(&b_dist).unwrap() }); } fn cache_distance(&self, target: &str) -> f32 { // Same chunk = 0, same doc = 1, same project = 2, shared = 3 if target == self.current_chunk { 0.0 } else if target.split('/').next() == self.current_chunk.split('/').next() { 1.0 } else if target.contains("shared") { 3.0 } else { 2.0 } } } ``` --- ## Implementation Details & Testing Strategy ### Phase 1: Wiki-Link Graph Indexing **Code Location:** ``` crates/mem-ingest/src/ ├── wiki_link_parser.rs (NEW: extract [[links]] from markdown) ├── wiki_link_index.rs (NEW: build graph, traverse) └── wiki_link_repo.rs (NEW: Postgres storage) crates/mem-store/src/ └── event_log.rs (EXTEND: add wiki_link_indexed event) k8s/ └── migrations/ └── wiki_links.sql (NEW: schema for wiki_links table) ``` **Implementation Tasks:** 1. **Wiki-Link Parser** (`wiki_link_parser.rs`) ```rust pub struct WikiLinkParser; impl WikiLinkParser { // Extract [[target]] from markdown pub fn parse_links(content: &str) -> Vec // Infer link type: memory | skill | shared fn infer_link_type(target: &str) -> LinkType // Resolve relative paths: [[debugging.md]] → poimen/tools/debugging.md fn resolve_path(target: &str, source_dir: &Path) -> PathBuf } ``` 2. **Wiki-Link Repository** (`wiki_link_repo.rs`) ```rust pub struct WikiLinkRepo; impl WikiLinkRepo { // Bulk insert/upsert wiki links pub async fn upsert_links(project: &str, links: Vec) -> Result // Query: all reachable docs from a project pub async fn reachable_docs(project: &str) -> Result> // Query: backlinks to a doc (who links to this?) pub async fn backlinks(project: &str, path: &str) -> Result> } ``` 3. **Schema** (`wiki_links.sql`) ```sql CREATE TABLE wiki_links ( id BIGSERIAL PRIMARY KEY, project VARCHAR(256), source_path VARCHAR(1024), target_path VARCHAR(1024), link_type VARCHAR(32), -- 'memory' | 'skill' | 'shared' resolved_path VARCHAR(1024), -- fully qualified path created_at TIMESTAMP, UNIQUE(project, source_path, target_path), INDEX(project, source_path), INDEX(project, target_path) ); ``` **Testing:** ```rust // tests/it_wiki_link_parser.rs #[tokio::test] async fn test_parse_wiki_links_basic() { let content = r#" ## Borrowing See [[lifetimes.md]] for more. Also check [[../../shared/skills/SKILL-ownership]] "#; let links = WikiLinkParser::parse_links(content); assert_eq!(links.len(), 2); assert!(links[0].target_path.contains("lifetimes")); assert!(links[1].target_path.contains("SKILL-ownership")); } #[tokio::test] async fn test_wiki_link_resolution() { let parser = WikiLinkParser::new("poimen/tools"); let resolved = parser.resolve_path("[[../concepts/design-patterns.md]]"); assert_eq!(resolved, Path::new("poimen/concepts/design-patterns.md")); } #[tokio::test] async fn test_reachable_docs() { // Insert test links let repo = WikiLinkRepo::new(db.clone()); repo.upsert_links("poimen", vec![ WikiLink { source: "tools/kubectl.md", target: "debugging.md", ... }, WikiLink { source: "debugging.md", target: "../../shared/skills/SKILL-k8s" }, ]).await?; // Query reachable from tools/kubectl.md let reachable = repo.reachable_docs("poimen").await?; assert!(reachable.contains(&"tools/kubectl.md".to_string())); assert!(reachable.contains(&"debugging.md".to_string())); assert!(reachable.contains(&"shared/skills/SKILL-k8s".to_string())); } ``` **Homelab Structure (test vault):** ``` homelab/vault/ projects/ poimen/ _access.yaml index.md "[[tools/kubectl.md]]" "[[memories/ownership.md]]" tools/ kubectl.md "[[../debugging/pod-crashes.md]]" "[[../../shared/skills/SKILL-kubernetes-debugging]]" debugging/ pod-crashes.md "[[../../memories/ownership.md]]" memories/ ownership.md "[[../tools/kubectl.md]]" rust-guide/ _access.yaml index.md "[[memories/ownership.md]]" memories/ ownership.md "[[../../shared/concepts/design-patterns.md]]" shared/ concepts/ design-patterns.md skills/ SKILL-kubernetes-debugging/ _access.yaml SKILL.md "[[../../projects/poimen/tools/kubectl.md]]" ``` **Verification Checklist:** - [ ] Parser correctly extracts all [[link]] types - [ ] Path resolution handles relative paths (../../../) - [ ] Reachable docs includes transitive links (A→B→C) - [ ] Backlinks correct (reverse graph) - [ ] Wiki-links survive Postgres round-trip --- ### Phase 2: Multi-Scope TF-IDF Indexing **Code Location:** ``` crates/mem-core/src/ ├── tfidf/ │ ├── mod.rs (NEW: module root) │ ├── global_index.rs (NEW: global term vocab) │ ├── project_index.rs (NEW: project-local TF-IDF) │ └── chunk_metadata.rs (NEW: heading+term extraction) └── optimizer/ └── tfidf_scorer.rs (NEW: TF-IDF scoring) k8s/migrations/ └── tfidf_index.sql (NEW: schema for term stats) ``` **Implementation Tasks:** 1. **Global Index Builder** (`global_index.rs`) ```rust pub struct GlobalTfIdfIndex { vocabulary: BTreeMap, idf_cache: Arc>, } impl GlobalTfIdfIndex { pub async fn build_from_vault(vault_root: &Path) -> Result pub async fn compute_idf(&self, term: &str) -> f32 pub async fn reload() -> Result<()> // hot-reload after ingest } ``` 2. **Project Index** (`project_index.rs`) ```rust pub struct ProjectTfIdfIndex { project: String, doc_freqs: HashMap, // doc_id → freq idf: BTreeMap, // term → IDF } impl ProjectTfIdfIndex { pub async fn build(project: &str, vault_root: &Path) -> Result pub fn score_chunk(&self, query: &str, doc_id: &str) -> f32 } ``` 3. **Chunk Metadata Extractor** (`chunk_metadata.rs`) ```rust pub struct ChunkMetadata { heading: String, first_sentence: String, key_terms: Vec<(String, f32)>, // term, TF category: String, // error | solution | tool | concept } impl ChunkMetadata { pub fn extract(content: &str) -> Result pub fn extract_key_terms(heading: &str, first_line: &str) -> Vec } ``` **Testing:** ```rust // tests/it_tfidf_indexing.rs #[tokio::test] async fn test_global_idf_computation() { let docs = vec![ "kubernetes pod error CrashLoopBackOff", "kubernetes pod debugging guide", "docker container error", ]; let index = GlobalTfIdfIndex::from_docs(&docs).await?; // "kubernetes" appears in 2/3 docs → IDF = log(3/2) ≈ 0.176 // "error" appears in 2/3 docs → IDF ≈ 0.176 // "CrashLoopBackOff" appears in 1/3 docs → IDF = log(3/1) ≈ 1.099 (high!) assert!(index.compute_idf("kubernetes") < 0.3); assert!(index.compute_idf("CrashLoopBackOff") > 0.9); } #[tokio::test] async fn test_project_scoped_scoring() { let vault = setup_test_vault().await?; let project_idx = ProjectTfIdfIndex::build("poimen", &vault).await?; let query = "kubernetes pod crash debugging"; let doc1_score = project_idx.score_chunk(query, "tools/kubectl.md"); let doc2_score = project_idx.score_chunk(query, "memories/ownership.md"); // kubectl.md should rank higher (more pod-related terms) assert!(doc1_score > doc2_score); } #[tokio::test] async fn test_chunk_metadata_extraction() { let content = r#" ## Pod Crashes in CrashLoopBackOff When a Kubernetes pod enters CrashLoopBackOff status, it means the container is crashing repeatedly. "#; let metadata = ChunkMetadata::extract(content)?; assert_eq!(metadata.heading, "## Pod Crashes in CrashLoopBackOff"); assert!(metadata.category == "error"); assert!(metadata.key_terms.iter().any(|(t, _)| t == "CrashLoopBackOff")); } ``` **Verification Checklist:** - [ ] Global IDF correctly computed (rare terms high, common low) - [ ] Project-local TF-IDF differs from global (project-specific vocab) - [ ] Chunk metadata correctly extracts heading + key terms - [ ] Category inference (error/solution/tool) accurate - [ ] TF-IDF scores consistent across Postgres round-trips - [ ] Benchmark: TF-IDF scoring < 10ms per query on 1000-doc project --- ### Phase 3: Hybrid Retrieval **Code Location:** ``` crates/mem-cli/src/ ├── query_router.rs (EXTEND: wiki-scoped routing) ├── retrieval_plan.rs (NEW: three-stage plan execution) └── authorized_query_router.rs (NEW: RBAC filtering) tests/ └── it_wiki_graph_retrieval.rs (NEW: end-to-end tests) ``` **Implementation Tasks:** 1. **Query Router with Wiki Scope** (`query_router.rs`) ```rust pub async fn route_query( &self, query: &str, project: &str, // wiki scope defined here ) -> Result { // 1. Get reachable docs from wiki-link graph let scoped_docs = wiki_index.reachable_docs(project).await?; // 2. TF-IDF pre-filter on scoped docs let tfidf_candidates = project_tfidf .score_chunks(query, &scoped_docs) .into_iter() .filter(|(_, score)| score > &0.1) .take(20) .collect::>(); // 3. Semantic search (only on TF-IDF candidates) let semantic_results = semantic_search(query, &tfidf_candidates).await?; // 4. RRF Fusion let fused = rrf_fusion(&tfidf_candidates, &semantic_results)?; Ok(RetrievalPlan { candidates: fused }) } ``` 2. **Retrieval Plan Executor** (`retrieval_plan.rs`) ```rust pub struct RetrievalPlan { candidates: Vec<(String, f32)>, // doc_id, fused_score project: String, search_strategy: String, } impl RetrievalPlan { pub async fn fetch_chunks(&self, budget: usize) -> Result> } ``` **Testing:** ```rust // tests/it_wiki_graph_retrieval.rs #[tokio::test] async fn test_wiki_scoped_retrieval() { setup_test_vault_with_links().await?; let router = QueryRouter::new(wiki_index, tfidf_idx, embeddings); let plan = router.route_query( "how to fix pod crashes", "poimen" ).await?; // Should only include docs reachable from poimen/index.md for (doc_id, _) in &plan.candidates { assert!( doc_id.starts_with("poimen/") || doc_id.starts_with("shared/"), "Doc {} outside project scope", doc_id ); } } #[tokio::test] async fn test_rrf_fusion_scoring() { let tfidf_results = vec![ ("doc1".to_string(), 0.9), ("doc2".to_string(), 0.7), ]; let semantic_results = vec![ ("doc2".to_string(), 0.95), ("doc1".to_string(), 0.6), ]; let fused = rrf_fusion(&tfidf_results, &semantic_results)?; // doc2 should rank higher (high semantic + ok TF-IDF) assert_eq!(fused[0].0, "doc2"); assert!(fused[0].1 > fused[1].1); } #[tokio::test] async fn test_llm_call_reduction() { let query = "kubernetes debugging"; let candidates = router.route_query(query, "poimen").await?.candidates; // Before optimization: 100+ candidates passed to LLM // After: only top-10 selected let selected = ChunkSelector::select(candidates, budget=4096).await?; assert!(selected.len() <= 10); assert!(selected.len() > 0); } ``` **Homelab Benchmark:** ```bash # Load test vault into Postgres + OpenSearch cargo test --test it_wiki_graph_retrieval -- --nocapture --test-threads=1 # Metrics to log: # - Retrieval latency (wiki-nav + TF-IDF + semantic) # - LLM call reduction % (original vs optimized) # - Chunk accuracy (top-5 results match expected docs) ``` **Verification Checklist:** - [ ] Wiki scoping correctly filters candidates - [ ] TF-IDF pre-filter reduces search space 80%+ - [ ] RRF fusion improves ranking vs semantic-only - [ ] Retrieval latency < 500ms (wiki + TF-IDF + semantic) - [ ] Selected chunks fit budget (< max_tokens) --- ### Phase 4-6: Chunk Metadata, Cache Alignment, Testing [Abbreviated for space; similar pattern to above] --- ### Phase 7: OIDC + RBAC Implementation **Code Location:** ``` crates/mem-cli/src/ ├── rbac/ │ ├── mod.rs │ ├── oidc_claims.rs (NEW: JWT parsing) │ ├── access_engine.rs (NEW: policy evaluation) │ ├── vault_policy_loader.rs (NEW: load from Vault) │ └── audit_log.rs (NEW: decision logging) └── http_server.rs (EXTEND: add RBAC middleware) k8s/migrations/ └── rbac.sql (NEW: rbac_audit_log table) ``` **Implementation Tasks:** 1. **OIDC Claims Extractor** (`oidc_claims.rs`) ```rust pub struct OidcClaims { pub sub: String, pub groups: Vec, pub roles: Vec, pub permissions: Vec, } impl OidcClaims { pub fn from_jwt(token: &str, jwks: &JWKS) -> Result } ``` 2. **RBAC Engine** (`access_engine.rs`) ```rust pub struct RbacEngine { policy_cache: Arc>, } impl RbacEngine { pub async fn check_access( &self, claims: &OidcClaims, resource_type: &str, // "project" | "skill" resource_name: &str, action: &str, // "read" | "write" ) -> Result { ... } } ``` **Testing:** ```rust // tests/it_rbac_engine.rs #[tokio::test] async fn test_oidc_jwt_parsing() { let token = create_test_jwt(vec!["platform-team"], "charlie"); let claims = OidcClaims::from_jwt(&token, &test_jwks())?; assert_eq!(claims.sub, "charlie"); assert!(claims.groups.contains(&"platform-team".to_string())); } #[tokio::test] async fn test_rbac_project_access_group() { let mut engine = RbacEngine::new(vault_client); let claims = OidcClaims { sub: "charlie".to_string(), groups: vec!["platform-team".to_string()], roles: vec!["viewer".to_string()], permissions: vec!["memory:read".to_string()], }; // Policy: access_level="group", allowed_groups=[platform-team] let allowed = engine.check_access(&claims, "project", "poimen", "read").await?; assert!(allowed); // Audit log check let audit = engine.last_decision_log()?; assert_eq!(audit.decision, "allow"); assert_eq!(audit.reason, "in_allowed_group"); } #[tokio::test] async fn test_rbac_project_access_denied() { let claims = OidcClaims { sub: "alice".to_string(), groups: vec!["data-team".to_string()], // NOT in allowed_groups roles: vec!["viewer".to_string()], permissions: vec![], }; let allowed = engine.check_access(&claims, "project", "poimen", "read").await?; assert!(!allowed); let audit = engine.last_decision_log()?; assert_eq!(audit.decision, "deny"); assert_eq!(audit.reason, "not_in_allowed_groups"); } #[tokio::test] async fn test_rbac_skill_filtering_in_retrieval() { let claims = OidcClaims { sub: "charlie".to_string(), groups: vec!["platform-team".to_string()], roles: vec![], permissions: vec![], }; let router = AuthorizedQueryRouter::new(query_router, rbac_engine); let plan = router.route_query( "kubernetes debugging", "poimen", &claims ).await?; // SKILL-private-debug (owner=ml-team) should be filtered out for (doc_id, _) in &plan.candidates { if doc_id.contains("SKILL-private") { panic!("Private skill leaked to unauthorized user"); } } } ``` **Homelab Setup for RBAC Testing:** ```yaml # homelab/vault/projects/poimen/_access.yaml project: poimen owner_group: platform-team access_level: group allowed_groups: [platform-team, devops-team] required_role: null # homelab/vault/shared/skills/SKILL-private-debug/_access.yaml skill: SKILL-private-debug owner_group: ml-team access_level: private allowed_groups: [] required_role: null # Authentik test users (via API) - charlie: groups=[platform-team, devops-team], roles=[viewer] - alice: groups=[data-team], roles=[viewer] - bob: groups=[ml-team], roles=[editor] ``` **Verification Checklist:** - [ ] JWT validation rejects expired/invalid tokens - [ ] OIDC claims correctly parsed from token - [ ] Project access check respects access_level (public/group/private) - [ ] Skill filtering removes unauthorized results - [ ] Audit log records all decisions (allow/deny) - [ ] Policy hot-reload works without service restart --- ## End-to-End Test Scenario ```bash # 1. Set up homelab vault structure mkdir -p homelab/vault/{projects/poimen,shared/skills} cp test_vault_structure.sh homelab/vault/ ./test_vault_structure.sh # 2. Load vault into Postgres + OpenSearch cargo run --bin mem ingest --project poimen --vault homelab/vault # 3. Create test users in Authentik curl -X POST http://authentik:9000/api/v3/core/users/ \ -H "Authorization: Bearer " \ -d '{"username": "charlie", "groups": ["platform-team", "devops-team"]}' # 4. Test wiki-graph retrieval curl -X POST http://localhost:8080/memory/query \ -H "Authorization: Bearer $(get_jwt charlie)" \ -d '{"project": "poimen", "query": "fix pod crash"}' # Expected: results from poimen + shared skills, filtered by RBAC # 5. Run integration tests cargo test --test it_wiki_graph_retrieval -- --nocapture cargo test --test it_rbac_engine -- --nocapture # 6. Benchmark retrieval performance cargo bench --bench wiki_graph_retrieval # Expected: wiki-scoped retrieval 70-80% faster than full-vault search ``` ## Implementation Order 1. **Phase 1** — Wiki-Link Graph Indexing - Obsidian vault parser - Store wiki-links in Postgres - Build wiki-link traversal index 2. **Phase 2** — Multi-Scope TF-IDF Indexing - Global + project-scoped + chunk-level TF-IDF - Async index builder - Store term statistics 3. **Phase 3** — Hybrid Retrieval (Wiki-Nav + TF-IDF + Semantic) - Query router - RRF fusion algorithm - Integration with existing pgvector search 4. **Phase 4** — LLM Call Optimization - Chunk selector (budget-aware) - Metrics tracking - A/B test: old retrieval vs wiki-graph-aware 5. **Phase 5** — Chunk-Level Metadata Index - Metadata extractor (heading, key terms, category) - Scoring boost for matches - Category-aware retrieval 6. **Phase 6** — Cache Alignment & KV Cache Optimization - Cache metrics tracking - Wiki-link ordering by cache locality - Monitor KV cache hit ratio 7. **Phase 7** — OIDC + RBAC (Universal Auth + Policy Enforcement) - Project + skill ownership model - Access level enforcement (private/group/public) - RBAC check in retrieval pipeline - Audit logging for compliance --- ## Architecture Refactoring: SOLID + DRY Optimization ### Problem Analysis Without refactoring, the design has antipatterns: ``` Current (Tightly Coupled): TfIdfIndex ─┬─ score_chunk() [duplicated logic across GlobalTfIdfIndex, │ ProjectTfIdfIndex, ChunkMetadataIndex] ├─ cache management [responsibility mixing] └─ build_from_vault() [coupled to Vault directly] RbacEngine ─┬─ load_policy() [duplicated in multiple methods] ├─ check_access() [fat method, 300+ lines, multiple concerns] └─ audit_log() [mixed responsibility] WikiLinkParser ─ resolve_path() called in multiple places (logic varies, not DRY) ``` ### Solution: Trait-Based Architecture #### 1. Scoring Pipeline (Single DocumentScorer trait) **Problem:** TF-IDF scoring logic duplicated in GlobalTfIdfIndex, ProjectTfIdfIndex, SemanticScorer, ChunkMetadataIndex. **Solution:** ```rust /// Single interface: one scorer, one job pub trait DocumentScorer: Send + Sync { async fn score(&self, query: &str, doc_id: &str) -> Result; fn name(&self) -> &str; // for debugging/metrics } // All scoring variants implement the same trait pub struct GlobalTfIdfScorer { vocabulary: Arc<...> } impl DocumentScorer for GlobalTfIdfScorer { ... } pub struct ProjectTfIdfScorer { project: String, vocabulary: Arc<...> } impl DocumentScorer for ProjectTfIdfScorer { ... } pub struct SemanticScorer { embeddings: Arc<...>, pgvector: Arc<...> } impl DocumentScorer for SemanticScorer { ... } pub struct MetadataBoostingScorer { base_scorer: Arc, // Composition, not inheritance metadata: Arc, boost_factor: f32, } impl DocumentScorer for MetadataBoostingScorer { ... } ``` **Multi-Scorer Orchestrator:** ```rust pub struct ScoringPipeline { scorers: Vec<(String, f32, Arc)>, // name, weight, scorer } impl ScoringPipeline { pub fn new() -> Self { ... } pub fn with_scorer( mut self, name: &str, weight: f32, scorer: Arc, ) -> Self { self.scorers.push((name.to_string(), weight, scorer)); self } /// Execute all scorers in parallel, fuse with RRF pub async fn score(&self, query: &str, doc_id: &str) -> Result { let scores = futures::stream::iter(&self.scorers) .then(|(_, _, scorer)| async move { scorer.score(query, doc_id).await }) .collect::>>() .await?; // RRF: weighted sum of normalized scores let weighted_sum: f32 = self.scorers .iter() .zip(scores) .map(|((_, weight, _), score)| weight * score) .sum(); Ok(weighted_sum / self.scorers.iter().map(|(_, w, _)| w).sum::()) } } // Usage: let pipeline = ScoringPipeline::new() .with_scorer("global-tfidf", 0.1, Arc::new(global_tfidf)) .with_scorer("project-tfidf", 0.3, Arc::new(project_tfidf)) .with_scorer("semantic", 0.6, Arc::new(semantic)); let final_score = pipeline.score(query, doc_id).await?; ``` **Benefits:** ✅ DRY | ✅ S | ✅ O | ✅ L | ✅ D | ✅ Testable #### 2. Policy Provider (Pluggable backend) **Problem:** RbacEngine tightly coupled to Vault. To support Postgres or Redis requires modifying multiple methods. **Solution:** ```rust #[async_trait] pub trait PolicyProvider: Send + Sync { async fn get_policy(&self, resource_type: &str, resource_name: &str) -> Result; async fn invalidate_cache(&self, resource_type: &str, name: &str) -> Result<()>; } // Vault implementation pub struct VaultPolicyProvider { vault_root: PathBuf, cache: Arc>>, } // Alternative: Postgres backend pub struct DatabasePolicyProvider { db: Arc, } // Decorator: Redis cache layer pub struct CachedPolicyProvider { inner: Arc, redis: Arc, ttl_secs: u64, } // Usage (all identical interface): let provider: Arc = match env { "vault" => Arc::new(VaultPolicyProvider::new(vault_root)), "postgres" => Arc::new(DatabasePolicyProvider::new(db)), }; let policy = provider.get_policy("project", "poimen").await?; ``` **Benefits:** ✅ O | ✅ D | ✅ Composition | ✅ Testable #### 3. RBAC Decision Engine (Composition of checkers) **Problem:** RbacEngine.check_access() is 300+ lines mixing access level, role, permission, and audit concerns. **Solution:** ```rust #[async_trait] pub trait AccessChecker: Send + Sync { async fn check(&self, claims: &OidcClaims, policy: &AccessPolicy) -> Result; fn description(&self) -> &str; } // Single-purpose checkers pub struct AccessLevelChecker; // public | group | private pub struct RoleChecker; // verify required_role pub struct PermissionChecker; // verify required_permission impl AccessChecker for AccessLevelChecker { ... } impl AccessChecker for RoleChecker { ... } impl AccessChecker for PermissionChecker { ... } /// Orchestrator: compose all checkers (short-circuit evaluation) pub struct AccessDecisionEngine { checkers: Vec>, policy_provider: Arc, audit: Arc, } impl AccessDecisionEngine { pub async fn check_access( &self, claims: &OidcClaims, resource_type: &str, resource_name: &str, ) -> Result { let policy = self.policy_provider.get_policy(resource_type, resource_name).await?; // Evaluate all checkers (short-circuit on failure) for checker in &self.checkers { if !checker.check(claims, &policy).await? { self.audit.log_decision(deny(checker.description())).await?; return Ok(false); } } self.audit.log_decision(allow()).await?; Ok(true) } } ``` **Benefits:** ✅ S | ✅ I | ✅ D | ✅ Easy to extend | ✅ Easy to test #### 4. Test Fixtures (Reusable builders) **Problem:** Test setup code repeated (creating OidcClaims, AccessPolicy, etc.). **Solution:** ```rust // tests/fixtures/builders.rs pub struct OidcClaimsBuilder { sub: String, groups: Vec, roles: Vec, permissions: Vec, } impl OidcClaimsBuilder { pub fn new(sub: &str) -> Self { ... } pub fn group(mut self, group: &str) -> Self { ... } pub fn role(mut self, role: &str) -> Self { ... } pub fn permission(mut self, perm: &str) -> Self { ... } pub fn build(self) -> OidcClaims { ... } } pub struct AccessPolicyBuilder { ... } // Usage in tests: #[tokio::test] async fn test_rbac_project_access_group() { let claims = OidcClaimsBuilder::new("charlie") .group("platform-team") .permission("memory:read") .build(); let policy = AccessPolicyBuilder::public() .group(vec!["platform-team"]) .build(); let engine = AccessDecisionEngine::new(provider, audit); let allowed = engine.check_access(&claims, "project", "poimen").await?; assert!(allowed); } ``` **Benefits:** ✅ DRY | ✅ Readable | ✅ Maintainable | ✅ Flexible ### Implementation Priority 1. **ScoringPipeline** (Phase 3) — unblocks all TF-IDF work 2. **PolicyProvider trait** (Phase 7) — enables Vault/Postgres/Redis swap 3. **AccessChecker composition** (Phase 7) — splits fat RBAC method 4. **Test fixtures** (All phases) — immediate DRY wins ### Summary: SOLID + DRY Improvements | Aspect | Before | After | |---|---|---| | **Code duplication** | TF-IDF logic in 3+ places | 1 trait, N implementations | | **New scorer** | Modify TfIdfIndex | New struct + impl DocumentScorer | | **Policy source** | Vault-only | Swap PolicyProvider trait | | **RBAC checks** | 1 fat method (300 lines) | 3 single-purpose checkers | | **Audit logging** | Hardcoded to Postgres | Pluggable AuditLogger trait | | **Test setup** | Repeated code | Reusable builders | | **Testability** | Hard to mock | Mock any trait | --- ## Expected Outcomes | Metric | Before | After | Target | |---|---|---|---| | LLM calls per query | 5-10 | 1-2 | < 2 | | Retrieval latency | 500ms+ | 100-200ms | < 200ms | | Chunk accuracy | 0.72 (noisy) | 0.88 (scoped) | > 0.85 | | Token efficiency | 60-70% | 85-90% | > 85% | | KV cache hits | 30% | 70%+ | > 70% | --- --- ## Phase 7: OIDC + RBAC (Universal Authentication & Authorization) ### Architecture: Authentik + Vault Policy Files ``` ┌──────────────────────┐ │ Authentik │ │ (OIDC Provider) │ └─────────────┬────────┘ │ OIDC token with claims: { "sub": "charlie", "groups": ["platform-team", "devops-team"], "roles": ["viewer", "editor"], "permissions": ["memory:read", "memory:write"] } │ ┌─────────┼─────────┬──────────┐ │ │ │ │ ▼ ▼ ▼ ▼ CLI HTTP API Agent WebUI │ │ │ │ └────┬────┴────┬───┴──────┬───┘ │ │ │ └─────────┼──────────┘ │ Fetch policy from Vault │ vault/policies/*.yaml vault/projects/*/\_access.yaml vault/shared/skills/*/\_access.yaml │ ┌────────▼────────┐ │ Auth Service │ │ (RBAC engine) │ │ │ │ 1. Extract │ │ OIDC claims │ │ 2. Load policy │ │ from Vault │ │ 3. Evaluate │ │ rules │ │ 4. Allow/Deny │ └────────┬────────┘ │ ┌────────▼────────┐ │ Access Decision │ │ + Audit Log │ └─────────────────┘ ``` ### OIDC Token Format (from Authentik) ```json { "sub": "charlie", "email": "charlie@company.com", "groups": [ "platform-team", "devops-team" ], "roles": [ "viewer", "editor" ], "permissions": [ "memory:read", "memory:write", "skill:read" ], "aud": "poimen-memory", "iss": "https://authentik.riotpiao.com/application/o/poimen-memory/", "exp": 1735689600 } ``` ### Access Model (Vault-based policies) ``` Project/Skill: ├─ owner: group (e.g., "platform-team", "data-team") ├─ access_level: "private" | "group" | "public" └─ allowed_groups: [group1, group2, ...] (if access_level == "group") User (from OIDC token): ├─ id (sub): string ├─ groups: ["platform-team", "dev-team", ...] ├─ roles: ["viewer", "editor", "admin"] └─ permissions: ["memory:read", "memory:write", ...] ``` ### Policy Files (in Vault) All policies stored as YAML in Vault, readable by any service: ```yaml # vault/policies/default.yaml # Global access policy default_access: public # assume public if no specific policy # vault/projects/poimen/_access.yaml project: poimen owner_group: platform-team access_level: group # "private" | "group" | "public" allowed_groups: - platform-team - devops-team required_role: viewer # minimum role needed # vault/shared/skills/SKILL-kubernetes-debugging/_access.yaml skill: SKILL-kubernetes-debugging owner_group: platform-team access_level: group allowed_groups: - platform-team - devops-team required_permission: skill:read ``` ### Audit Log (Postgres) Note: Authentik **already logs** all OIDC token issues. We add **application-level audit** for authorization decisions: ```sql CREATE TABLE rbac_audit_log ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(256), -- from OIDC 'sub' user_groups TEXT[], -- from OIDC 'groups' resource_type VARCHAR(32), -- "project" | "skill" | "memory" resource_name VARCHAR(256), action VARCHAR(32), -- "read" | "write" | "denied" decision VARCHAR(32), -- "allow" | "deny" reason VARCHAR(256), -- "access_level_public", "in_allowed_group", "not_owner", etc. timestamp TIMESTAMP DEFAULT NOW(), trace_id VARCHAR(256) -- correlate with Authentik logs ); ``` ### Universal RBAC Engine (OIDC-native) ```rust pub struct OidcClaims { pub sub: String, // user ID (from Authentik) pub groups: Vec, // group memberships pub roles: Vec, // "viewer", "editor", "admin" pub permissions: Vec, // fine-grained perms } pub struct AccessPolicy { pub owner_group: String, pub access_level: String, // "private" | "group" | "public" pub allowed_groups: Vec, pub required_role: Option, // minimum role needed pub required_permission: Option, // fine-grained check } pub struct RbacEngine { vault: VaultClient, // read policies from Vault audit: PostgresPool, // log decisions } impl RbacEngine { /// Universal authorization check: works for any service pub async fn check_access( &self, claims: &OidcClaims, // from OIDC token resource_type: &str, // "project", "skill", "memory" resource_name: &str, action: &str, // "read", "write" ) -> Result { // 1. Fetch policy from Vault (can be cached) let policy = self.vault .get_policy(resource_type, resource_name) .await?; let mut reason = String::new(); let mut allowed = false; // 2. Check access level match policy.access_level.as_str() { "public" => { allowed = true; reason = "access_level_public".to_string(); } "private" => { // Only owner group allowed = claims.groups.contains(&policy.owner_group); reason = if allowed { "owner_group".to_string() } else { "not_in_owner_group".to_string() }; } "group" => { // Check allowed groups allowed = claims.groups .iter() .any(|g| policy.allowed_groups.contains(g)); reason = if allowed { "in_allowed_group".to_string() } else { "not_in_allowed_groups".to_string() }; } _ => { return Err(anyhow!("Unknown access level")); } } // 3. Check role requirement (if any) if let Some(required_role) = &policy.required_role { if !claims.roles.contains(required_role) { allowed = false; reason = format!("role_requirement_failed: need {}", required_role); } } // 4. Check fine-grained permission (if any) if let Some(required_perm) = &policy.required_permission { if !claims.permissions.contains(required_perm) { allowed = false; reason = format!("permission_required: {}", required_perm); } } // 5. Audit log (always) self.audit_log( &claims.sub, &claims.groups, resource_type, resource_name, action, allowed, &reason, ).await?; Ok(allowed) } async fn audit_log( &self, user_id: &str, user_groups: &[String], resource_type: &str, resource_name: &str, action: &str, allowed: bool, reason: &str, ) -> Result<()> { sqlx::query( "INSERT INTO rbac_audit_log (user_id, user_groups, resource_type, resource_name, action, decision, reason) VALUES ($1, $2, $3, $4, $5, $6, $7)" ) .bind(user_id) .bind(user_groups) .bind(resource_type) .bind(resource_name) .bind(action) .bind(if allowed { "allow" } else { "deny" }) .bind(reason) .execute(&self.audit) .await?; Ok(()) } } ``` ### JWT Token Flow & RBAC Checking ``` ┌─────────────────────────────────────────────────────────────────────┐ │ Service Entry Points │ │ │ │ HTTP API CLI Agent (Claude) │ │ ──────── ─── ────────────── │ │ Authorization:Bearer env MEM_API_TOKEN auth: BearerToken │ │ │ │ │ └────────────────┬────────────────┬───────────────┬───────────────────┘ │ │ │ └────────────────┼───────────────┘ │ JWT token ┌────────▼────────────┐ │ Token Validation │ │ │ │ 1. Fetch JWKS from │ │ Authentik: │ │ GET /jwks │ │ │ │ 2. Verify signature │ │ (RSA/ECDSA) │ │ │ │ 3. Check expiry, │ │ audience,issuer │ └────────┬────────────┘ │ ┌─────────────▼─────────────┐ │ Invalid/expired? │ └──┬──────────────────────┬─┘ no │ │ yes │ ▼ │ ┌──────────────┐ │ │ 401 Unauth │ │ │ Return │ │ └──────────────┘ │ ▼ ┌──────────────────────────────┐ │ Extract OIDC Claims │ │ │ │ { │ "sub": "charlie", │ "groups": ["platform-team", │ "devops-team"], │ "roles": ["viewer"], │ "permissions": ["memory:read"], │ "aud": "poimen-memory", │ "exp": 1735689600 │ } └──────┬───────────────────────┘ │ OidcClaims ▼ ┌─────────────────────────────┐ │ Query/Request arrives │ │ {project, resource, action} │ └──────┬──────────────────────┘ │ ┌──────────▼──────────────┐ │ RbacEngine │ │ .check_access() │ │ │ │ ┌───────────────────┐ │ │ │ 1. Load policy │ │ │ │ from Vault: │ │ │ │ vault/ │ │ │ │ projects/ │ │ │ │ poimen/ │ │ │ │ _access.yaml │ │ │ └──────┬────────────┘ │ │ │ Policy │ │ ┌──────▼────────────┐ │ │ │ 2. Access level? │ │ │ │ │ │ │ │ "public" → │ │ │ │ allow │ │ │ │ │ │ │ │ "group" → │ │ │ │ check │ │ │ │ claims.groups │ │ │ │ vs policy. │ │ │ │ allowed_groups │ │ │ │ │ │ │ │ "private" → │ │ │ │ check │ │ │ │ claims.groups │ │ │ │ contains owner │ │ │ └──────┬────────────┘ │ │ │ │ │ ┌──────▼────────────┐ │ │ │ 3. Role check │ │ │ │ (if required) │ │ │ │ │ │ │ │ claims.roles │ │ │ │ contains │ │ │ │ policy. │ │ │ │ required_role? │ │ │ └──────┬────────────┘ │ │ │ │ │ ┌──────▼────────────┐ │ │ │ 4. Permission │ │ │ │ check │ │ │ │ (if required) │ │ │ │ │ │ │ │ claims. │ │ │ │ permissions │ │ │ │ contains policy. │ │ │ │ required_perm? │ │ │ └──────┬────────────┘ │ │ │ │ │ ┌──────▼────────────┐ │ │ │ 5. Audit log │ │ │ │ (always) │ │ │ │ │ │ │ │ INSERT INTO │ │ │ │ rbac_audit_log │ │ │ │ {user_id, groups, │ │ │ │ resource, │ │ │ │ decision, │ │ │ │ reason} │ │ │ └──────┬────────────┘ │ │ │ │ └─────────┼───────────────┘ │ ┌──────────▼──────────┐ │ Authorization │ │ Decision │ └──┬──────────────┬───┘ yes │ │ no │ ▼ │ ┌────────────────┐ │ │ 403 Forbidden │ │ │ reason: ... │ │ │ Return │ │ └────────────────┘ │ ▼ ┌─────────────────────────────────┐ │ Authorized Query Router │ │ route_query() │ │ │ │ 1. Wiki-graph scoped to │ │ project:poimen │ │ │ │ 2. TF-IDF filter (within scope) │ │ │ │ 3. Semantic search (within scope)│ │ │ │ 4. For each candidate skill: │ │ RbacEngine.check_access( │ │ claims, │ │ "skill", │ │ "SKILL-k8s-debug", │ │ "read" │ │ ) │ │ │ │ 5. Filter results: keep only │ │ resources user can access │ │ │ └────────┬────────────────────────┘ │ ▼ ┌────────────────────┐ │ Return Results │ │ │ │ ✓ chunks authorized│ │ ✓ skills authorized│ │ ✓ memories indexed │ │ │ └────────────────────┘ ``` ### Integration with Retrieval Pipeline ```rust pub struct AuthorizedQueryRouter { router: QueryRouter, access_control: AccessControl, } impl AuthorizedQueryRouter { pub async fn route_query( &self, query: &str, project: &str, user: &User, ) -> Result { // 1. Check if user can access this project if !self.access_control.check_project_access(user, project, "read").await? { return Err(anyhow!("Access denied: user {} not authorized for project {}", user.id, project)); } // 2. Route query (wiki-graph + TF-IDF + semantic) let mut plan = self.router.route_query(query, project).await?; // 3. Filter results: remove skills/memories user cannot access plan.candidates = futures::stream::iter(plan.candidates) .filter_map(|candidate| async move { // Check if this is a skill or memory if candidate.0.contains("SKILL-") { if self.access_control.check_skill_access(user, &candidate.0).await.ok()? { Some(candidate) } else { None } } else { // Regular memory: accessible if project is accessible (already checked above) Some(candidate) } }) .collect() .await; Ok(plan) } } ``` ### HTTP API Integration ```rust pub async fn handle_query( auth: BearerToken, // from Authorization: Bearer body: QueryRequest, rbac_engine: &RbacEngine, router: &QueryRouter, ) -> Result { // 1. Validate JWT and extract OIDC claims let claims = validate_and_decode_jwt(&auth.token, &AUTHENTIK_JWKS).await? .into_oidc_claims(); // Extract sub, groups, roles, permissions // 2. Check if user can access the project resource let authorized = rbac_engine.check_access( &claims, "project", &body.project, "read" ).await?; if !authorized { return Err(anyhow!("403 Forbidden: insufficient permissions for project {}", body.project)); } // 3. Route query with scope = project let mut plan = router.route_query(&body.query, &body.project).await?; // 4. Filter candidates: remove skills/memories user cannot access let filtered_candidates = futures::stream::iter(plan.candidates) .filter_map(|(doc_id, score)| async move { // If this is a skill, check skill-level RBAC if doc_id.contains("SKILL-") { let authorized = rbac_engine.check_access( &claims, "skill", &doc_id, "read" ).await.ok()?; if authorized { Some((doc_id, score)) } else { None // Filtered out } } else { // Regular memory: already covered by project RBAC check above Some((doc_id, score)) } }) .collect() .await; plan.candidates = filtered_candidates; // 5. Fetch and return authorized chunks let results = plan.fetch_chunks().await?; Ok(Response::ok(results)) } ``` ### How JWT Access is Handled **When does RBAC checking happen?** 1. **At every service entry point** (HTTP API, CLI, Agent) - JWT token arrives (Authorization header, env var, or embedded auth) - Token is validated against Authentik JWKS - Claims extracted 2. **For project-level resources** - `RbacEngine.check_access(claims, "project", "poimen", "read")` - Load policy from `vault/projects/poimen/_access.yaml` - Check: access_level + user groups + roles + permissions - Log decision in `rbac_audit_log` - **If denied: 403 Forbidden immediately** (before any search) 3. **For skill-level resources** - After wiki-graph + TF-IDF + semantic search returns candidates - **For each skill in results:** - `RbacEngine.check_access(claims, "skill", "SKILL-k8s-debug", "read")` - Load policy from `vault/shared/skills/SKILL-k8s-debug/_access.yaml` - Filter in/out from results - **Denied skills are silently filtered** (not shown in results) 4. **For memory-level resources** (within project) - Already covered by project-level check - No separate skill-like RBAC per individual memory doc - (Optional: add granular memory RBAC later if needed) **Example scenarios:** ``` Scenario 1: Charlie queries project:poimen ───────────────────────────────────────── 1. JWT token arrives 2. Claims extracted: sub=charlie, groups=[platform-team, devops-team] 3. RbacEngine.check_access(claims, "project", "poimen", "read") 4. Load vault/projects/poimen/_access.yaml 5. access_level="group", allowed_groups=[platform-team, devops-team] 6. Check: [platform-team, devops-team] ∩ charlie's groups? YES 7. Audit log: user=charlie, resource=project:poimen, decision=allow, reason=in_allowed_group 8. Proceed to wiki-graph + search Scenario 2: Alice (not in platform-team) queries project:poimen ────────────────────────────────────────────────────────────────── 1. JWT token arrives 2. Claims extracted: sub=alice, groups=[data-team] 3. RbacEngine.check_access(claims, "project", "poimen", "read") 4. Load vault/projects/poimen/_access.yaml 5. access_level="group", allowed_groups=[platform-team, devops-team] 6. Check: [platform-team, devops-team] ∩ alice's groups? NO 7. Audit log: user=alice, resource=project:poimen, decision=deny, reason=not_in_allowed_groups 8. Return: 403 Forbidden Scenario 3: Charlie searches in poimen, results include SKILL-private-debug ────────────────────────────────────────────────────────────────────────── 1. Charlie queries → project access check PASSES 2. Wiki-graph + search returns candidates including SKILL-private-debug 3. For SKILL-private-debug: RbacEngine.check_access(claims, "skill", "SKILL-private-debug", "read") 4. Load vault/shared/skills/SKILL-private-debug/_access.yaml 5. access_level="private", owner_group=ml-team 6. Check: ml-team ∩ charlie's groups? NO 7. Audit log: decision=deny 8. Filter out SKILL-private-debug from results 9. Return: other allowed skills + memories ``` ### Configuration (YAML) ```yaml # vault/projects/poimen/_access.yaml project: poimen owner_group: platform-team access_level: group allowed_groups: - platform-team - devops-team required_role: null # any role can read required_permission: null # vault/projects/ai-infra/_access.yaml project: ai-infra owner_group: ml-team access_level: public # anyone can read allowed_groups: [] required_role: null required_permission: null # vault/shared/skills/SKILL-kubernetes-debugging/_access.yaml skill: SKILL-kubernetes-debugging owner_group: platform-team access_level: group allowed_groups: - platform-team - devops-team required_role: null required_permission: skill:read # vault/shared/skills/SKILL-testing/_access.yaml skill: SKILL-testing owner_group: engineering access_level: public allowed_groups: [] required_role: null required_permission: null ``` ### Vault Organization (with RBAC metadata & JWT-ready) ``` vault/ projects/ poimen/ _access.yaml owner: platform-team access_level: group allowed_groups: [platform-team, devops-team] index.md memories/ ownership.md tools/ kubectl.md ai-infra/ _access.yaml owner: ml-team access_level: public index.md shared/ skills/ SKILL-kubernetes-debugging/ _access.yaml owner: platform-team access_level: group allowed_groups: [platform-team, devops-team] SKILL.md SKILL-testing/ _access.yaml owner: engineering access_level: public SKILL.md ``` ### Audit & Compliance ```sql -- Query access audit SELECT * FROM access_log WHERE timestamp > NOW() - INTERVAL '7 days' AND resource_type = 'project' AND action = 'read' ORDER BY timestamp DESC; -- Who accessed what SELECT user_id, resource_name, COUNT(*) as access_count FROM access_log WHERE action = 'read' GROUP BY user_id, resource_name ORDER BY access_count DESC; -- Denied access attempts SELECT user_id, resource_name, reason, COUNT(*) as deny_count FROM access_log WHERE action = 'denied' GROUP BY user_id, resource_name, reason ORDER BY deny_count DESC; ``` --- ## References - **Wiki-Link Graph**: Obsidian backlinks, roam-research style - **TF-IDF**: Classic information retrieval, project-scoped variant - **RRF Fusion**: Reciprocal Rank Fusion (IR best practice) - **KV Cache**: Language model inference optimization - **RAG**: Retrieval-Augmented Generation (context grounding) - **RBAC**: Role-Based Access Control (zero-trust principle)