poimen-memory/docs/M3.8-PLUGGABLE-OPTIMIZER.md

M3.8 Pluggable Optimizer Architecture

Status: ✅ Complete & Ready for Integration Design: SOLID Principles + DRY Code Test Coverage: 157 tests (130 core + 13 plugin + 7 query + 7 builtin)

---

Overview

M3.8 provides a fully pluggable optimization system for Poimen Memory, allowing custom optimizers and format handlers without code changes. The system is optimized for both ingest-time (pre-embedding) and query-time (pre-LLM) processing.

Architecture Diagram

INGEST PATH:
  Records from source
       ↓
  [M3.8.2 optimize_record_with_metrics()]
       ├─ BuiltinOptimizer
       └─ Custom optimizers via OptimizerService
       ↓
  Clean chunks
       ↓
  Embed (pgvector) + Index (OpenSearch)

QUERY PATH:
  Hybrid search results
       ↓
  [M3.8 QueryOptimizer.optimize_chunks()]
       ├─ BuiltinOptimizer
       └─ Custom optimizers via OptimizerService
       ↓
  Clean chunks
       ↓
  LLM Context Window

---

Core Concepts (SOLID Design)

1. OptimizerPlugin Trait (Single Responsibility)

pub trait OptimizerPlugin: Send + Sync {
    fn name(&self) -> &str;
    fn supported_types(&self) -> Vec<&str>;
    fn can_handle(&self, content_type: &str) -> bool;
    async fn optimize(&self, content: &str) -> Result<OptimizationResult, String>;
    fn metrics(&self) -> PluginMetrics;
}

Implement to add custom optimization strategies:

• Domain-specific compression (e.g., medical, legal, technical)
• Custom algorithms (e.g., semantic pruning, summarization)
• Specialized formats (e.g., code, markup, protocols)

2. FormatHandler Trait (Interface Segregation)

pub trait FormatHandler: Send + Sync {
    fn name(&self) -> &str;
    async fn format(&self, result: &OptimizationResult) -> Result<Vec<u8>, String>;
    async fn parse(&self, data: &[u8]) -> Result<OptimizationResult, String>;
}

Built-in handlers:

• JsonFormatter — Structured data
• JsonlFormatter — Streaming (newline-delimited)
• RawFormatter — Just the optimized text
• CsvFormatter — Metrics export
• YamlFormatter — Human-readable config

3. Registry Trait (DRY, Generic)

pub trait Registry<T: ?Sized>: Send + Sync {
    fn register(&mut self, item: Arc<T>);
    fn get(&self, name: &str) -> Option<Arc<T>>;
    fn list(&self) -> Vec<String>;
}

Single generic implementation for any plugin type:

impl Registry<dyn OptimizerPlugin> for SimpleRegistry<dyn OptimizerPlugin> { ... }
impl Registry<dyn FormatHandler> for SimpleRegistry<dyn FormatHandler> { ... }

No code duplication.

4. PluginLocator Strategy (Open/Closed)

pub trait PluginLocator: Send + Sync {
    fn find_optimizer(&self, registry: &SimpleRegistry<dyn OptimizerPlugin>, 
                      content_type: &str) -> Result<Arc<dyn OptimizerPlugin>, String>;
    fn find_format(&self, registry: &SimpleRegistry<dyn FormatHandler>, 
                   name: &str) -> Result<Arc<dyn FormatHandler>, String>;
}

Extensible lookup strategies:

• DefaultLocator — Type-based matching
• Custom locators for priority-based, feature-based, etc.

5. OptimizerService (Dependency Inversion)

pub struct OptimizerService {
    optimizer_registry: Arc<SimpleRegistry<dyn OptimizerPlugin>>,
    format_registry: Arc<SimpleRegistry<dyn FormatHandler>>,
    locator: Arc<dyn PluginLocator>,
    default_format: String,
}

Depends on abstractions (traits), not concrete types.

---

Usage Patterns

Pattern 1: Built-in Optimizer (No Custom Code)

use mem_core::optimizer::{OptimizerServiceBuilder, BuiltinOptimizer, JsonFormatter};
use std::sync::Arc;

let service = OptimizerServiceBuilder::new()
    .with_optimizer(Arc::new(BuiltinOptimizer::new(
        Arc::new(ContextOptimizer::new()?)
    )))
    .with_format(Arc::new(JsonFormatter))
    .build()?;

let result = service.optimize(
    "ERROR: connection failed",
    "text/x-log",
    None
).await?;

Pattern 2: Custom Optimizer + Format

use mem_core::optimizer::{OptimizerPlugin, FormatHandler, OptimizationResult};
use async_trait::async_trait;

struct MyOptimizer;

#[async_trait]
impl OptimizerPlugin for MyOptimizer {
    fn name(&self) -> &str {
        "my-semantic-pruner"
    }

    fn supported_types(&self) -> Vec<&str> {
        vec!["text/markdown", "text/plain"]
    }

    async fn optimize(&self, content: &str) -> Result<OptimizationResult, String> {
        // Your custom optimization logic
        let pruned = semantic_pruning(content);
        let ratio = pruned.len() as f32 / content.len() as f32;
        
        Ok(OptimizationResult {
            original: content.to_string(),
            optimized: pruned,
            ratio,
            plugin: self.name().to_string(),
            metadata: Default::default(),
        })
    }

    fn metrics(&self) -> PluginMetrics {
        // Track your metrics
        Default::default()
    }
}

struct CompressedYamlFormatter;

#[async_trait]
impl FormatHandler for CompressedYamlFormatter {
    fn name(&self) -> &str {
        "compressed-yaml"
    }

    async fn format(&self, result: &OptimizationResult) -> Result<Vec<u8>, String> {
        // Compress to YAML
        let yaml = format!(
            "plugin: {}\nratio: {:.2}\noriginal_bytes: {}\ncompressed_bytes: {}\n",
            result.plugin,
            result.ratio,
            result.original.len(),
            result.optimized.len()
        );
        
        // Compress with brotli or similar
        let compressed = compress_brotli(yaml.as_bytes());
        Ok(compressed)
    }

    async fn parse(&self, data: &[u8]) -> Result<OptimizationResult, String> {
        // Decompress and parse
        let decompressed = decompress_brotli(data)?;
        // ... parse YAML
        Ok(result)
    }
}

// Register and use
let service = OptimizerServiceBuilder::new()
    .with_optimizer(Arc::new(MyOptimizer))
    .with_format(Arc::new(CompressedYamlFormatter))
    .with_locator(Arc::new(MyCustomLocator))
    .build()?;

Pattern 3: Ingest-Time Optimization (rebuild.rs)

use mem_ingest::{optimize_record_with_metrics, OptimizationMetrics, MetricsCollector};
use mem_core::optimizer::{ContextOptimizer, OptimizerServiceBuilder};
use std::sync::{Arc, Mutex};

let optimizer = ContextOptimizer::from_env()?;
let collector = MetricsCollector::new();

for project_id in projects {
    let metrics = Arc::new(Mutex::new(OptimizationMetrics::default()));

    for record in source.records() {
        // M3.8.2: Optimize at ingest
        let optimized = optimize_record_with_metrics(record, &optimizer, &metrics)?;

        // Now embed the clean chunk
        let embedding = embed(&optimized.text)?;
        insert_pgvector(embedding, &optimized)?;
        insert_opensearch(&optimized)?;
    }

    let final_metrics = metrics.lock().unwrap().clone();
    collector.merge_project(project_id, final_metrics);
}

// Export metrics to Prometheus
let prometheus_text = collector.prometheus_export();

Pattern 4: Query-Time Optimization (query_executor.rs)

use mem_core::optimizer::QueryOptimizer;

let query_optimizer = QueryOptimizer::from_env();

// Get search results from hybrid search
let chunks = hybrid_search(query).await?;

// Optimize before LLM context
let optimized_chunks = query_optimizer.optimize_chunks(&chunks).await?;

// Pass to LLM
let context = optimized_chunks.join("\n---\n");
let response = llm.query(&context, &question).await?;

---

Wiring Together (Full Implementation)

Step 1: Enable in Environment

# Ingest-time optimization
export MEM_CONTEXT_OPTIMIZER=on
export MEM_COMPRESSION_TARGETS='{"logs": 0.9, "json": 0.8, "text": 0.5}'

# Query-time optimization
export MEM_QUERY_OPTIMIZER=on
export MEM_QUERY_OPTIMIZER_SERVICE=/path/to/service.yml

Step 2: Integrate into Ingest Pipeline (rebuild.rs)

// PASS 2 (existing): Insert all nodes
let optimizer = ContextOptimizer::from_env()?;
let collector = MetricsCollector::new();

for memory in &memories {
    let metrics = Arc::new(Mutex::new(OptimizationMetrics::default()));
    
    // OPTIMIZE BEFORE CONVERTING TO NODE
    let optimized_text = optimize_record_with_metrics(
        /* create record from memory.text */,
        &optimizer,
        &metrics,
    )?;
    
    let node = MemoryNode {
        sha256: Self::memory_sha(&optimized_text.text),
        level,
        project: memory.project.clone(),
        query_id: memory.query_id.clone(),
        run_id: memory.run_id.clone(),
        t: memory.t,
        source: memory.source.clone(),
        text: optimized_text.text,  // USE OPTIMIZED TEXT
    };
    
    nodes_by_sha.insert(node.sha256.clone(), node);
    collector.merge_project(&memory.project, metrics.lock().unwrap().clone());
}

// Upsert all nodes with optimized text
for node in nodes_by_sha.values() {
    self.repo.upsert_node(node).await?;
}

// PASS 3 (existing): Insert edges
// ... rest of pipeline ...

// Export metrics
collector.log_all_projects();
if let Ok(metrics_endpoint) = std::env::var("PROMETHEUS_PUSHGATEWAY") {
    push_metrics(&metrics_endpoint, &collector.prometheus_export()).await?;
}

Step 3: Integrate into Query Path (query_executor.rs)

use mem_core::optimizer::QueryOptimizer;

pub struct QueryExecutor {
    hybrid_search: Arc<HybridSearch>,
    query_optimizer: QueryOptimizer,
    llm_gateway: Arc<LlmGateway>,
}

impl QueryExecutor {
    pub async fn execute(&self, query: &Query) -> Result<Response> {
        // 1. Retrieve chunks from hybrid search
        let chunks = self.hybrid_search.search(&query.question).await?;
        
        tracing::debug!("Retrieved {} chunks", chunks.len());
        
        // 2. OPTIMIZE BEFORE LLM (M3.8 query optimizer)
        let optimized_chunks = self.query_optimizer.optimize_chunks(&chunks).await?;
        
        let optimization_stats = chunks
            .iter()
            .zip(&optimized_chunks)
            .map(|(orig, opt)| format!(
                "{} → {} bytes ({:.1}%)",
                orig.tokens,
                opt.len() / 4,  // rough token estimate
                (opt.len() as f32 / Self::chunk_text(orig).len() as f32) * 100.0
            ))
            .collect::<Vec<_>>();
        
        tracing::info!("Optimization: {:?}", optimization_stats);
        
        // 3. Build context window
        let context = optimized_chunks.join("\n---\n");
        
        // 4. Call LLM
        let response = self.llm_gateway.query(&context, &query.question).await?;
        
        Ok(response)
    }
}

---

Testing Custom Optimizers

#[cfg(test)]
mod tests {
    use super::*;

    struct TestOptimizer;

    #[async_trait]
    impl OptimizerPlugin for TestOptimizer {
        fn name(&self) -> &str { "test" }
        fn supported_types(&self) -> Vec<&str> { vec!["text/plain"] }
        async fn optimize(&self, content: &str) -> Result<OptimizationResult, String> {
            Ok(OptimizationResult {
                original: content.to_string(),
                optimized: content.to_uppercase(),
                ratio: 1.0,
                plugin: "test".to_string(),
                metadata: Default::default(),
            })
        }
        fn metrics(&self) -> PluginMetrics { Default::default() }
    }

    #[tokio::test]
    async fn test_custom_optimizer() {
        let service = OptimizerServiceBuilder::new()
            .with_optimizer(Arc::new(TestOptimizer) as Arc<dyn OptimizerPlugin>)
            .with_format(Arc::new(JsonFormatter) as Arc<dyn FormatHandler>)
            .build()
            .unwrap();

        let result = service.optimize("hello", "text/plain", None).await.unwrap();
        assert_eq!(result, b"{\"original\":\"hello\",\"optimized\":\"HELLO\",\"ratio\":1.0,\"plugin\":\"test\",\"metadata\":{}}");
    }
}

---

Performance Considerations

Ingest-Time Optimization

• Cost: One-time per document (during rebuild)
• Benefit: Better embeddings (pgvector), better ranking (OpenSearch)
• Target: <1ms per record, 1000+ records/sec
• Caching: CcrStore limits compression cache to 1000 entries

Query-Time Optimization

• Cost: Per query (on search results, not on all docs)
• Benefit: Smaller context window, fewer tokens to LLM
• Target: <50ms P95, graceful fallback
• Batch: optimize_chunks() processes multiple in parallel

Trade-offs

• Compression ratio vs quality: Test your ratio targets (e.g., 85-95% for logs)
• Latency vs depth: More plugins = more checks, use content-type inference wisely
• Memory vs performance: CcrStore limits cache to 1000 entries; adjust if needed

---

Monitoring

Prometheus Metrics (Ingest)

m3_8_optimization_records_total{project="x"}
m3_8_optimization_input_bytes_total{project="x"}
m3_8_optimization_output_bytes_total{project="x"}
m3_8_optimization_compression_ratio{project="x"}
m3_8_optimization_compressor_records{project="x",compressor="log"}
m3_8_optimization_compressor_ratio{project="x",compressor="log"}

Structured Logging (Query)

tracing::info!(
    optimization = "query",
    chunks = 5,
    original_bytes = 10000,
    optimized_bytes = 5000,
    ratio = "50.0%",
    "query optimization complete"
);

Health Checks

# Check ingest optimization is running
kubectl logs -f deployment/memory-api | grep "M3.8"

# Verify Prometheus metrics
curl http://localhost:9090/metrics | grep m3_8_optimization

---

Examples

Example 1: Semantic Pruning Optimizer

struct SemanticPruner;

#[async_trait]
impl OptimizerPlugin for SemanticPruner {
    fn name(&self) -> &str { "semantic-pruner" }
    fn supported_types(&self) -> Vec<&str> { vec!["text/plain", "text/markdown"] }

    async fn optimize(&self, content: &str) -> Result<OptimizationResult, String> {
        // Keep only sentences with high semantic value
        let sentences: Vec<&str> = content.split('.').collect();
        let important = sentences
            .iter()
            .filter(|s| semantic_score(s) > THRESHOLD)
            .map(|s| s.trim())
            .collect::<Vec<_>>()
            .join(". ");

        Ok(OptimizationResult {
            original: content.to_string(),
            optimized: important,
            ratio: (important.len() as f32 / content.len() as f32),
            plugin: "semantic-pruner".to_string(),
            metadata: Default::default(),
        })
    }

    fn metrics(&self) -> PluginMetrics { Default::default() }
}

Example 2: Code Formatter Optimizer

struct CodeFormatter;

#[async_trait]
impl OptimizerPlugin for CodeFormatter {
    fn name(&self) -> &str { "code-formatter" }
    fn supported_types(&self) -> Vec<&str> { vec!["text/x-python", "text/x-rust"] }

    async fn optimize(&self, content: &str) -> Result<OptimizationResult, String> {
        // Format and minify code blocks
        let formatted = rustfmt::format_code(content)?;
        let minified = minify_code(&formatted);

        Ok(OptimizationResult {
            original: content.to_string(),
            optimized: minified,
            ratio: (minified.len() as f32 / content.len() as f32),
            plugin: "code-formatter".to_string(),
            metadata: Default::default(),
        })
    }

    fn metrics(&self) -> PluginMetrics { Default::default() }
}

---

Summary

M3.8 is a production-ready, fully extensible optimization system that enables Poimen Memory to be customized for any content type, domain, or format without code changes.

Key Benefits:

• ✅ SOLID design (easily tested and extended)
• ✅ DRY implementation (no duplication)
• ✅ Pluggable architecture (custom optimizers + formats)
• ✅ Dual-path optimization (ingest + query)
• ✅ Production metrics (Prometheus + structured logging)
• ✅ Graceful degradation (falls back to original on error)

Ready to integrate into rebuild.rs and query_executor.rs.