Files

T

rock 2850907167 docs: merge ARCHITECTURE_REFACTORING into memory-wiki-graph-rag-optimization.md

Integrated SOLID + DRY optimizations as new section:
- Scoring pipeline (DocumentScorer trait, ScoringPipeline orchestrator)
- Policy provider (PolicyProvider trait, pluggable Vault/Postgres/Redis)
- RBAC decision engine (AccessChecker composition, short-circuit eval)
- Test fixtures (OidcClaimsBuilder, AccessPolicyBuilder)

Implementation priority:
1. ScoringPipeline (Phase 3)
2. PolicyProvider trait (Phase 7)
3. AccessChecker composition (Phase 7)
4. Test fixtures (All phases)

Unified doc now has: architecture + concrete implementation + SOLID refactoring.

2026-08-30 20:36:49 -07:00

74 KiB

Raw Blame History

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

-- 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

pub struct WikiLinkParser {
    project: String,
    vault_root: PathBuf,
}

impl WikiLinkParser {
    pub async fn parse_file(&self, path: &Path) -> Result<Vec<WikiLink>> {
        // 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

pub struct TfIdfIndex {
    pub global: BTreeMap<String, TermStats>,        // term → idf
    pub project_scope: HashMap<String, ProjectIndex>,  // project → {term → tf}
    pub chunk_metadata: HashMap<String, Vec<Term>>,  // doc_id → [term, ...]
}

pub struct TermStats {
    pub idf: f32,               // log(total_docs / docs_with_term)
    pub global_freq: u32,
    pub project_freqs: HashMap<String, u32>,
}

pub struct ProjectIndex {
    pub project: String,
    pub docs: HashMap<String, f32>,  // doc_id → term_freq
    pub idf: BTreeMap<String, f32>,   // 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

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

pub struct TfIdfBuilder {
    vault_root: PathBuf,
}

impl TfIdfBuilder {
    pub async fn build_indexes(self) -> Result<TfIdfIndex> {
        // 1. Scan all markdown files
        let files = self.scan_vault().await?;
        
        // 2. Extract terms per project
        let mut project_docs: HashMap<String, Vec<String>> = 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<String> {
        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

pub struct QueryRouter {
    wiki_index: WikiLinkIndex,
    tfidf: TfIdfIndex,
    embeddings: EmbeddingsClient,
}

impl QueryRouter {
    pub async fn route_query(
        &self,
        query: &str,
        project: &str,
    ) -> Result<RetrievalPlan> {
        // 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::<Vec<_>>()
        ).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

fn rrf_fusion(
    tfidf_results: &[(String, f32)],
    semantic_results: &[(String, f32)],
) -> Result<Vec<(String, f32)>> {
    let mut scores: HashMap<String, f32> = 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

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<Vec<String>> {
        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<MinimizedCallPlan> {
        // 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

pub struct ChunkMetadata {
    pub doc_id: String,
    pub heading: String,           // ## heading
    pub first_sentence: String,    // first 20 words
    pub key_terms: Vec<String>,    // extracted terms
    pub category: String,          // "error", "solution", "tool", "concept"
}

pub struct MetadataExtractor;

impl MetadataExtractor {
    pub fn extract(content: &str) -> Result<ChunkMetadata> {
        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<String> {
        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

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

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:

Same chunk (already in cache)
Adjacent chunks (likely prefetched)
Same document (same KV cache line)
Related documents (via wiki-link proximity)

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:

Wiki-Link Parser (wiki_link_parser.rs)

pub struct WikiLinkParser;
impl WikiLinkParser {
    // Extract [[target]] from markdown
    pub fn parse_links(content: &str) -> Vec<WikiLink>

    // 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
}

Wiki-Link Repository (wiki_link_repo.rs)

pub struct WikiLinkRepo;
impl WikiLinkRepo {
    // Bulk insert/upsert wiki links
    pub async fn upsert_links(project: &str, links: Vec<WikiLink>) -> Result<u32>

    // Query: all reachable docs from a project
    pub async fn reachable_docs(project: &str) -> Result<Vec<String>>

    // Query: backlinks to a doc (who links to this?)
    pub async fn backlinks(project: &str, path: &str) -> Result<Vec<String>>
}

Schema (wiki_links.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:

// 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:

Global Index Builder (global_index.rs)

pub struct GlobalTfIdfIndex {
    vocabulary: BTreeMap<String, TermStats>,
    idf_cache: Arc<RwLock<...>>,
}
impl GlobalTfIdfIndex {
    pub async fn build_from_vault(vault_root: &Path) -> Result<Self>
    pub async fn compute_idf(&self, term: &str) -> f32
    pub async fn reload() -> Result<()>  // hot-reload after ingest
}

Project Index (project_index.rs)

pub struct ProjectTfIdfIndex {
    project: String,
    doc_freqs: HashMap<String, f32>,  // doc_id → freq
    idf: BTreeMap<String, f32>,       // term → IDF
}
impl ProjectTfIdfIndex {
    pub async fn build(project: &str, vault_root: &Path) -> Result<Self>
    pub fn score_chunk(&self, query: &str, doc_id: &str) -> f32
}

Chunk Metadata Extractor (chunk_metadata.rs)

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<Self>
    pub fn extract_key_terms(heading: &str, first_line: &str) -> Vec<String>
}

Testing:

// 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:

Query Router with Wiki Scope (query_router.rs)

pub async fn route_query(
    &self,
    query: &str,
    project: &str,  // wiki scope defined here
) -> Result<RetrievalPlan> {
    // 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::<Vec<_>>();

    // 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 })
}

Retrieval Plan Executor (retrieval_plan.rs)

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<Vec<Chunk>>
}

Testing:

// 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:

# 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:

OIDC Claims Extractor (oidc_claims.rs)

pub struct OidcClaims {
    pub sub: String,
    pub groups: Vec<String>,
    pub roles: Vec<String>,
    pub permissions: Vec<String>,
}
impl OidcClaims {
    pub fn from_jwt(token: &str, jwks: &JWKS) -> Result<Self>
}

RBAC Engine (access_engine.rs)

pub struct RbacEngine {
    policy_cache: Arc<RwLock<PolicyCache>>,
}
impl RbacEngine {
    pub async fn check_access(
        &self,
        claims: &OidcClaims,
        resource_type: &str,  // "project" | "skill"
        resource_name: &str,
        action: &str,         // "read" | "write"
    ) -> Result<bool> { ... }
}

Testing:

// 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:

# 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

# 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 <token>" \
  -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

Phase 1 — Wiki-Link Graph Indexing
- Obsidian vault parser
- Store wiki-links in Postgres
- Build wiki-link traversal index
Phase 2 — Multi-Scope TF-IDF Indexing
- Global + project-scoped + chunk-level TF-IDF
- Async index builder
- Store term statistics
Phase 3 — Hybrid Retrieval (Wiki-Nav + TF-IDF + Semantic)
- Query router
- RRF fusion algorithm
- Integration with existing pgvector search
Phase 4 — LLM Call Optimization
- Chunk selector (budget-aware)
- Metrics tracking
- A/B test: old retrieval vs wiki-graph-aware
Phase 5 — Chunk-Level Metadata Index
- Metadata extractor (heading, key terms, category)
- Scoring boost for matches
- Category-aware retrieval
Phase 6 — Cache Alignment & KV Cache Optimization
- Cache metrics tracking
- Wiki-link ordering by cache locality
- Monitor KV cache hit ratio
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:

/// Single interface: one scorer, one job
pub trait DocumentScorer: Send + Sync {
    async fn score(&self, query: &str, doc_id: &str) -> Result<f32>;
    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<dyn DocumentScorer>,  // Composition, not inheritance
    metadata: Arc<ChunkMetadataCache>,
    boost_factor: f32,
}
impl DocumentScorer for MetadataBoostingScorer { ... }

Multi-Scorer Orchestrator:

pub struct ScoringPipeline {
    scorers: Vec<(String, f32, Arc<dyn DocumentScorer>)>,  // name, weight, scorer
}

impl ScoringPipeline {
    pub fn new() -> Self { ... }
    
    pub fn with_scorer(
        mut self,
        name: &str,
        weight: f32,
        scorer: Arc<dyn DocumentScorer>,
    ) -> 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<f32> {
        let scores = futures::stream::iter(&self.scorers)
            .then(|(_, _, scorer)| async move { scorer.score(query, doc_id).await })
            .collect::<Result<Vec<_>>>()
            .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::<f32>())
    }
}

// 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:

#[async_trait]
pub trait PolicyProvider: Send + Sync {
    async fn get_policy(&self, resource_type: &str, resource_name: &str) -> Result<AccessPolicy>;
    async fn invalidate_cache(&self, resource_type: &str, name: &str) -> Result<()>;
}

// Vault implementation
pub struct VaultPolicyProvider {
    vault_root: PathBuf,
    cache: Arc<RwLock<LruCache<String, AccessPolicy>>>,
}

// Alternative: Postgres backend
pub struct DatabasePolicyProvider {
    db: Arc<PostgresPool>,
}

// Decorator: Redis cache layer
pub struct CachedPolicyProvider {
    inner: Arc<dyn PolicyProvider>,
    redis: Arc<redis::aio::Connection>,
    ttl_secs: u64,
}

// Usage (all identical interface):
let provider: Arc<dyn PolicyProvider> = 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:

#[async_trait]
pub trait AccessChecker: Send + Sync {
    async fn check(&self, claims: &OidcClaims, policy: &AccessPolicy) -> Result<bool>;
    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<Arc<dyn AccessChecker>>,
    policy_provider: Arc<dyn PolicyProvider>,
    audit: Arc<dyn AuditLogger>,
}

impl AccessDecisionEngine {
    pub async fn check_access(
        &self,
        claims: &OidcClaims,
        resource_type: &str,
        resource_name: &str,
    ) -> Result<bool> {
        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:

// tests/fixtures/builders.rs
pub struct OidcClaimsBuilder {
    sub: String,
    groups: Vec<String>,
    roles: Vec<String>,
    permissions: Vec<String>,
}

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

ScoringPipeline (Phase 3) — unblocks all TF-IDF work
PolicyProvider trait (Phase 7) — enables Vault/Postgres/Redis swap
AccessChecker composition (Phase 7) — splits fat RBAC method
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)

{
  "sub": "charlie",
  "email": "[email protected]",
  "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:

# 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:

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)

pub struct OidcClaims {
    pub sub: String,                    // user ID (from Authentik)
    pub groups: Vec<String>,            // group memberships
    pub roles: Vec<String>,             // "viewer", "editor", "admin"
    pub permissions: Vec<String>,       // fine-grained perms
}

pub struct AccessPolicy {
    pub owner_group: String,
    pub access_level: String,           // "private" | "group" | "public"
    pub allowed_groups: Vec<String>,
    pub required_role: Option<String>,  // minimum role needed
    pub required_permission: Option<String>,  // 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<bool> {
        // 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>                 <jwt>                <jwt>                  │
│                                                                     │
└────────────────┬────────────────┬───────────────┬───────────────────┘
                 │                │               │
                 └────────────────┼───────────────┘
                                  │ 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

pub struct AuthorizedQueryRouter {
    router: QueryRouter,
    access_control: AccessControl,
}

impl AuthorizedQueryRouter {
    pub async fn route_query(
        &self,
        query: &str,
        project: &str,
        user: &User,
    ) -> Result<RetrievalPlan> {
        // 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

pub async fn handle_query(
    auth: BearerToken,  // from Authorization: Bearer <jwt>
    body: QueryRequest,
    rbac_engine: &RbacEngine,
    router: &QueryRouter,
) -> Result<Response> {
    // 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?

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
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)
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)
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)

# 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

-- 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)

74 KiB Raw Blame History

Memory Wiki-Graph RAG Optimization

Architecture Overview

Phase 1: Wiki-Link Graph Indexing

Schema

Ingestion: Wiki-Link Parser

Phase 2: Multi-Scope TF-IDF Indexing

TF-IDF Scopes

Data Structure

Scoring Algorithm

Indexing Pipeline

Phase 3: Hybrid Retrieval (Wiki-Nav + TF-IDF + Semantic)

Query Router

RRF Fusion

Phase 4: LLM Call Optimization

Chunk Selection Strategy

Metrics

Phase 5: Chunk-Level Metadata Index

Chunk Metadata Extraction

Chunk-Level Scoring Boost

Phase 6: Cache Alignment & KV Cache Optimization

KV Cache Hit Ratio Tracking

Wiki-Link Graph Ordering (Cache-Aligned)

Implementation Details & Testing Strategy

Phase 1: Wiki-Link Graph Indexing

Phase 2: Multi-Scope TF-IDF Indexing

Phase 3: Hybrid Retrieval

Phase 4-6: Chunk Metadata, Cache Alignment, Testing

Phase 7: OIDC + RBAC Implementation

End-to-End Test Scenario

Implementation Order

Architecture Refactoring: SOLID + DRY Optimization

Problem Analysis

Solution: Trait-Based Architecture

1. Scoring Pipeline (Single DocumentScorer trait)

2. Policy Provider (Pluggable backend)

3. RBAC Decision Engine (Composition of checkers)

4. Test Fixtures (Reusable builders)

Implementation Priority

Summary: SOLID + DRY Improvements

Expected Outcomes

Phase 7: OIDC + RBAC (Universal Authentication & Authorization)

Architecture: Authentik + Vault Policy Files

OIDC Token Format (from Authentik)

Access Model (Vault-based policies)

Policy Files (in Vault)

Audit Log (Postgres)

Universal RBAC Engine (OIDC-native)

JWT Token Flow & RBAC Checking

Integration with Retrieval Pipeline

HTTP API Integration

How JWT Access is Handled

Configuration (YAML)

Vault Organization (with RBAC metadata & JWT-ready)

Audit & Compliance

References

74 KiB

Raw Blame History