poimen-memory/tasks/M3.8-CORRECTED-architecture.md

M3.8 — Context Optimizer: CORRECTED Architecture

Current Misunderstanding ❌

What I documented: M3.8 compresses chunks AFTER hybrid search, BEFORE sending to LLM

• PromptBuilder.build_cache_aligned() calls optimizer
• Only affects LLM input, not search indexes

Why this is incomplete: Ignores the ingest-time quality improvement

---

Correct Architecture ✅

M3.8 operates at INGEST, not query:

Raw Content (logs, JSON, transcripts, diffs)
       ↓
   M3.8 Optimization (clean, denoise, normalize)
       ├─ Stage 1: ContentRouter (detect type)
       ├─ Stage 2: Compressor (type-specific cleaning)
       ├─ Stage 3: CacheAligner (remove timestamps, IDs)
       └─ Stage 4: Output optimized chunk
       ↓
   Higher-quality document
       ↓
   Embed (nomic 768-dim) → pgvector (better embeddings)
   Index (OpenSearch BM25) → better ranking signals
       ↓
   Search retrieval improves (less noise, higher recall)
       ↓
   LLM receives better chunks (already clean)

Benefits of Ingest-Time Optimization

1. Better Embeddings: Clean text → higher semantic quality
2. Better Ranking: BM25 on signal-rich text (no temp paths/timestamps)
3. Reduced Noise: pgvector doesn't embed verbose noise
4. No Query-Path Overhead: Process once at ingest, use clean results in all queries
5. Consistent Results: Same optimized documents used for all users/queries

---

Implementation Plan

Phase 1: Ingest Pipeline Integration (NEW)

Goal: Wire M3.8 optimizer into ingest sources

Files to create/modify:

• crates/mem-ingest/src/optimizer_sink.rs (NEW, 100 LOC)

  • OptimizerSink: Wrapper around ingest sources
  • Applies M3.8 optimization before emission
  • Configuration: which compressors enabled per project

• crates/mem-ingest/src/lib.rs (modify)

  • Export OptimizerSink
  • Lazy-load ContextOptimizer

• tests/it_ingest_optimizer.rs (NEW, 150 LOC)

  • Test each ingest source with optimizer
  • Verify compression ratios on real sources
  • Assert cleaner output vs raw

Phase 2: Metrics/Monitoring (NEW)

Goal: Track optimization effectiveness in the ingest chain

Files to create:

• crates/mem-core/src/optimizer/metrics.rs (NEW, 80 LOC)

  • OptimizationMetrics: compression_ratio, input_bytes, output_bytes, compressor_used
  • MetricsCollector: Track per-source, per-type, per-project
  • Emit to structured logging (tracing)

• tests/it_optimizer_metrics.rs (NEW, 100 LOC)

  • Assert metrics collected for each chunk
  • Verify accuracy of compression ratio
  • Test aggregation across multiple chunks

Phase 3: Query Path Simplification (REFACTOR)

Goal: Remove query-path compression since ingest handles it

Currently in PromptBuilder:

• build_cache_aligned() calls optimizer (REMOVE)
• cache_metrics() calculates drift (KEEP, but simplified)

Change:

• PromptBuilder gets already-optimized chunks from search
• No additional compression needed
• cache_metrics() still tracks drift for observability (but on clean chunks)

---

Task Breakdown

Task	Phase	LOC	Tests	Purpose
M3.8.1	1	1100	62	Core compressor modules (DONE)
M3.8.2	2	80	10	Ingest pipeline integration
M3.8.3	2	80	10	Metrics & monitoring
M3.8.4	3	-50	0	Remove query-path compression
M3.8.5	2	—	20	Ingest sources + optimizer e2e tests
M3.8.6	3	—	—	M3.8 gate (compression targets + search quality)

---

Detailed Spec: M3.8.2 (Ingest Pipeline)

OptimizerSink Wrapper

pub struct OptimizerSink {
    inner: Box<dyn RecordSource>,  // original source (PiSession, DocCorpus, etc)
    optimizer: ContextOptimizer,
    config: OptimizerConfig,  // per-project settings
    metrics: MetricsCollector,
}

impl RecordSource for OptimizerSink {
    fn next_record(&mut self) -> Option<Record> {
        let record = self.inner.next_record()?;
        
        // Optimize each chunk in the record
        let optimized_content = self.optimizer.optimize(&record.content)?;
        
        // Collect metrics
        self.metrics.record(OptimizationMetrics {
            input_bytes: record.content.len(),
            output_bytes: optimized_content.compressed.len(),
            compressor: optimized_content.compressor_used,
            compression_ratio: ...,
        });
        
        Some(Record {
            content: optimized_content.compressed,  // use optimized
            ..record
        })
    }
}

pub fn optimize_source(
    source: Box<dyn RecordSource>,
    project: &str,
) -> Result<OptimizerSink> {
    let optimizer = ContextOptimizer::from_env()?;
    let config = OptimizerConfig::for_project(project);
    
    Ok(OptimizerSink {
        inner: source,
        optimizer,
        config,
        metrics: MetricsCollector::new(),
    })
}

Integration Points

In rebuild.rs:

let source = DocCorpusSource::new(vault_path)?;
let optimized = optimize_source(Box::new(source), &project)?;

// Use optimized source for embedding + indexing
for record in optimized {
    let embedding = embed(&record.content)?;  // embed clean text
    insert_pgvector(embedding, &record)?;
    insert_opensearch(&record)?;
}

In CLI:

# New command: re-optimize existing vault
mem rebuild --project=poimen --optimize --vault=/data/vault

# Inspect optimization effectiveness
mem stats --project=poimen --show=compression
  input_bytes: 52.4 MB
  output_bytes: 31.2 MB
  compression_ratio: 59.5%
  compressor_breakdown:
    - log: 85%
    - json: 72%
    - text: 41%
    - diff: 67%

---

Tests: M3.8.2 (10 tests)

#[test]
fn test_optimizer_sink_preserves_record_structure() { ... }

#[test]
fn test_optimizer_sink_reduces_bytes() { ... }

#[test]
fn test_optimizer_sink_content_type_routing() { ... }

#[test]
fn test_optimizer_sink_handles_errors_gracefully() { ... }

#[test]
fn test_optimizer_sink_disabled_via_env() { ... }

#[test]
fn test_pi_session_source_optimized() { ... }

#[test]
fn test_doc_corpus_source_optimized() { ... }

#[test]
fn test_claude_transcript_source_optimized() { ... }

#[test]
fn test_compression_ratio_accurate() { ... }

#[test]
fn test_metrics_collected_per_chunk() { ... }

---

Tests: M3.8.3 (Metrics, 10 tests)

#[test]
fn test_metrics_collector_tracks_bytes() { ... }

#[test]
fn test_metrics_collector_aggregates_per_project() { ... }

#[test]
fn test_metrics_collector_tracks_per_compressor() { ... }

#[test]
fn test_metrics_emitted_to_tracing() { ... }

#[test]
fn test_metrics_include_input_output_bytes() { ... }

#[test]
fn test_metrics_timestamp_accurate() { ... }

#[test]
fn test_metrics_thread_safe() { ... }

#[test]
fn test_metrics_query_interface() { ... }

#[test]
fn test_metrics_histogram_generation() { ... }

#[test]
fn test_metrics_output_to_prometheus() { ... }

---

Files Summary

New:

• crates/mem-ingest/src/optimizer_sink.rs (100 LOC)
• crates/mem-core/src/optimizer/metrics.rs (80 LOC)
• tests/it_ingest_optimizer.rs (150 LOC)
• tests/it_optimizer_metrics.rs (100 LOC)

Modified:

• crates/mem-ingest/src/lib.rs (+20 LOC)
• crates/mem-core/src/optimizer/mod.rs (+10 LOC config export)
• crates/mem-store/src/rebuild.rs (+30 LOC integration)

Removed:

• PromptBuilder.build_cache_aligned() query-path optimizer call (simplify)

---

Acceptance Criteria

✅ OptimizerSink integrated with all ingest sources ✅ Metrics collected and emitted for every chunk ✅ Compression ratios match targets (log 85-95%, json 70-90%, etc.) ✅ 20 new tests all passing ✅ Zero breaking changes to ingest interface ✅ Performance: <1ms per chunk optimization (measured via metrics)