rock/poimen-memory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4d2dd6408b28fa443e1c8e14ca89d6b79e6a75a0
poimen-memory/docs/memory-wiki-graph-rag-optimization.md
T
rock 4d2dd6408b docs: add memory-wiki-graph-rag-optimization.md — complete RAG + RBAC design 
7 phases:
1. Wiki-link graph indexing (project scopes, skill links)
2. Multi-scope TF-IDF (global + project-local + chunk-level)
3. Hybrid retrieval (wiki-nav + TF-IDF + semantic search + RRF fusion)
4. LLM call optimization (budget-aware chunk selection)
5. Chunk-level metadata (category boost, key terms)
6. Cache alignment (KV cache hit ratio via wiki-link ordering)
7. OIDC + RBAC (JWT from Authentik, policy files in Vault)

JWT flow:
- Token validated against Authentik JWKS
- OIDC claims extracted (sub, groups, roles, permissions)
- Project-level RBAC check (403 if denied)
- Skill-level RBAC filtering (denied skills silently removed)
- All decisions logged to rbac_audit_log

3 access levels: private (owner only) | group (explicit list) | public
Policies stored in vault as YAML, any service can enforce.
2026-08-30 20:24:42 -07:00

1439 lines
49 KiB
Markdown
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
```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<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
```rust
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
```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<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
```rust
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
```rust
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
```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<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
```rust
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
```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 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** — RBAC (Role-Based Access Control)
- Project + skill ownership model
- Access level enforcement (private/group/public)
- RBAC check in retrieval pipeline
- Audit logging for compliance
---
## 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": "[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:
```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<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
```rust
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
```rust
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?**
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)
Reference in New Issue View Git Blame Copy Permalink