docs: CRITICAL CORRECTION — M3.8 architecture (ingest, not query)

ISSUE IDENTIFIED:
M3.8 was misplaced in query path (PromptBuilder), but should be in ingest path
- Current: Compress before LLM (query-time, only helps LLM input)
- Correct: Optimize before embed + index (ingest-time, improves search quality)

BENEFITS OF INGEST-TIME OPTIMIZATION:
✅ Better embeddings (pgvector gets clean text → higher semantic quality)
✅ Better ranking (OpenSearch gets signal-rich text → better BM25 scores)
✅ One-time processing at ingest, not per-query overhead
✅ All queries benefit from cleaner search results
✅ LLM receives already-optimized chunks

NEW PLAN:
- M3.8.1: 🟡 Core modules PARTIAL (1100 LOC, 62 tests done, needs ingest wiring)
- M3.8.2: ⬜ Ingest integration (OptimizerSink wrapper, 13 tests)
- M3.8.3: ⬜ Metrics & monitoring (20 tests, tracing + prometheus)
- M3.8.4: ⬜ Query cleanup (remove PromptBuilder optimizer call)
- M3.8.5: ⬜ Benchmarks (compression ratios + search quality metrics)
- M3.8.6: ⬜ Gate (ingest pipeline quality + search improvement)

ARCHITECTURE CORRECTED:
Raw content → M3.8 optimize → embed + index → search improves → LLM benefits

FILES UPDATED:
- tasks/M3.8-CORRECTED-architecture.md (NEW, comprehensive re-plan)
- tasks/M3.8.1-context-optimizer.md (REWRITTEN, marked PARTIAL)
- tasks/M3.8.2-cache-aligner-headers.md (REWRITTEN, now OptimizerSink)

NEXT IMMEDIATE STEP:
Implement M3.8.2 (OptimizerSink) to wire compressors into rebuild.rs ingest pipeline

tasks/M3.8-CORRECTED-architecture.md

@@ -0,0 +1,288 @@
+# 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
+
+```rust
+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**:
+```rust
+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**: 
+```bash
+# 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)
+
+```rust
+#[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)
+
+```rust
+#[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)

tasks/M3.8.1-context-optimizer.md

@@ -1,168 +1,139 @@
-# M3.8.1 — Context Optimizer: pre-LLM compression pipeline
+# M3.8.1 — Context Optimizer Core Modules
 
 | Field | Value |
 |---|---|
 | Phase | M3.8 — Context optimization |
 | Size | L — 3–5 days |
-| Status | ✅ COMPLETE (all 4 phases) |
+| Status | 🟡 PARTIAL (modules done, needs ingest wiring) |
-| Flags | — |
 | Spec | `docs/CONTEXT_OPTIMIZER.md` |
-| Blocks | M3.8.2 |
+| Blocks | M3.8.2 (ingest integration) |
 | Depends | M3.7.7 (lesson.rs patterns), M3.7.8 (stop words) |
 
-## Summary
+## Status: PARTIAL ⚠️
 
-✅ **M3.8.1 COMPLETE** — 4 phases, 62 unit tests, full PromptBuilder integration
+✅ **Core Compressor Modules Complete**: 1,100 LOC, 62 tests
+- ContentRouter (Magika ML detection)
+- LogCompressor, JsonCrusher, DiffCompressor, TextCompressor
+- CacheAligner (drift detection)
+- CcrStore (reversible compression)
+- ContextOptimizer orchestrator
 
-**Total Implementation**: ~1,100 LOC across 8 modules
+❌ **Integration in Wrong Place**:
+- Currently: PromptBuilder.build_cache_aligned() (query path)
+- Should be: rebuild.rs ingest pipeline (ingest path)
+- Result: Improves only LLM input, not search quality
 
-| Module | LOC | Purpose |
+## What Was Done Right
-|--------|-----|---------|
-| ContentRouter | 150 | Magika ML + regex content detection |
-| LogCompressor | 260 | Error line + stack trace preservation |
-| JsonCrusher | 300 | Field variance + boundary-aware compression |
-| DiffCompressor | 180 | Change-line extraction, context dropping |
-| TextCompressor | 320 | Token importance scoring with stop words |
-| CacheAligner | 180 | Dynamic pattern detection + prefix stabilization |
-| CcrStore | 170 | LRU cache with SHA256 hashing + TTL |
-| ContextOptimizer | 150 | Orchestrator + env config |
 
-**Test Coverage**: 62 unit tests
+✅ **Content Detection** (Magika ML + regex)
-- Phase 1: 17 tests (router, log)
+- <1ms classification
-- Phase 2: 15 tests (json, diff)
+- Detects JSON, code, logs, diffs, config, text
-- Phase 3: 18 tests (cache align, CCR)
+- Thread-safe, ONNX local
-- Phase 4: 12 tests (text, config)
 
-**Commits**:
+✅ **5 Compressor Implementations**
-1. `bf13e3a` — Phase 1 complete (17 tests)
+- LogCompressor: 85-95% ratio (keep errors + stack traces)
-2. `a903a3f` — Phase 2 (15 tests)
+- JsonCrusher: 70-90% ratio (field variance)
-3. `edcc231` — Phase 3 (18 tests)
+- DiffCompressor: 60-80% ratio (change lines only)
-4. `8d8addc` — Phase 4a (12 tests) + PromptBuilder integration
+- TextCompressor: 30-50% ratio (token importance)
+- ConfigCompressor: passthrough (already compact)
 
-## Architecture
+✅ **Cache Alignment**
+- Detects dynamic patterns (timestamps, UUIDs, session IDs)
+- Drift metric (0.0-1.0)
+- Separates stable prefix from dynamic tail
 
-### Layer 1: Content Detection (Magika ML)
+✅ **Reversible Compression** (CCR Store)
-- Google Magika ONNX model (<1ms classification)
+- LRU cache with SHA256
-- Detects: JSON, code (Python/Rust/Go/JS/TS), logs, diffs, config, text
+- TTL-based expiry
-- Regex fallback when confidence < 0.7
+- Model can retrieve originals via hint injection
-- Thread-safe Mutex-wrapped Session
 
-### Layer 2: Compression (Per-Type)
+## What Needs Fixing
-- **LogCompressor** (85-95% ratio): error lines + stack traces only
-- **JsonCrusher** (70-90% ratio): statistical field analysis, boundary items
-- **DiffCompressor** (60-80% ratio): change lines only, drop context
-- **TextCompressor** (30-50% ratio): token importance, drop stop words
-- **ConfigCompressor** (passthrough): YAML/TOML already compact
 
-### Layer 3: Cache Alignment
+### Root Issue: Architecture Misunderstanding
-- Detects dynamic patterns: timestamps, UUIDs, session IDs, temp paths, hashes
-- Moves to tail, preserves stable prefix for LLM KV cache hits
-- Drift metrics (0.0-1.0 ratio) for monitoring
 
-### Layer 4: Reversible Compression (CCR)
+**Documented** (❌ Wrong):
-- LRU cache with IndexMap (insertion-order preserving)
+```
-- SHA256 hashing for content identification
+Ingest → pgvector + OpenSearch (full noise)
-- TTL-based expiry (default 1hr)
+  ↓
-- Thread-safe Mutex wrapper
+Query → M3.8 compression → LLM
-- Injection hint: `<!-- CCR:hash -->` for model retrieval
-
-## Integration
-
-**PromptBuilder::build_cache_aligned()**:
-```rust
-// After rendering, before prompt assembly:
-let optimizer = ContextOptimizer::from_env()?;
-let optimized = optimizer.optimize(&chunk_text)?;
-let chunk_text = optimized.compressed;  // Use optimized version
 ```
 
-**Environment Configuration**:
+**Should Be** (✅ Correct):
-```bash
+```
-# Enable/disable optimizer
+Ingest → M3.8 optimization → pgvector + OpenSearch (clean)
-MEM_CONTEXT_OPTIMIZER=on
+  ↓
-
+Query → retrieve clean results → LLM
-# Per-compressor controls (defaults: all on except CODE)
-MEM_COMPRESS_JSON=on
-MEM_COMPRESS_LOGS=on
-MEM_COMPRESS_CODE=off    # opt-in
-MEM_COMPRESS_DIFF=on
-MEM_COMPRESS_TEXT=on
-
-# Detection & caching
-MEM_MAGIKA_ENABLED=on
-MEM_MAGIKA_THRESHOLD=0.7
-MEM_CCR_ENABLED=on
 ```
 
-## Test Results
+**Why the correct way is better:**
+1. Cleaner text → better embeddings (pgvector)
+2. Signal-rich text → better BM25 ranking (OpenSearch)
+3. One-time processing at ingest, not per-query
+4. All users benefit from cleaner search results
+5. LLM already gets optimized chunks
 
-✅ **62 unit tests all passing**
+### Next Steps
-- 17 phase 1 (router, log)
-- 15 phase 2 (json, diff)
-- 18 phase 3 (cache align, CCR)
-- 12 phase 4 (text, config)
 
-✅ **114 total mem-core tests** (all passing)
+**M3.8.2**: Ingest Pipeline Integration (1 day)
-- 62 optimizer tests
+- Create OptimizerSink wrapper around ingest sources
-- 11 prompt tests (including PromptBuilder integration)
+- Wire into rebuild.rs
-- 10 query tests
+- Test with all source types
-- 10 symptom projection tests
+- Collect metrics
-- Plus integration scenarios
 
-## Quality & Safety
+**M3.8.3**: Metrics & Monitoring (1 day)
+- Track compression ratio per chunk
+- Aggregate per project/source/type
+- Emit to tracing/Prometheus
+- Dashboard visualization
 
-✅ **Graceful Degradation**:
+**M3.8.4**: Query Path Cleanup (0.5 days)
-- If Magika fails to load, falls back to regex
+- Remove PromptBuilder.build_cache_aligned() optimizer call
-- If optimizer unavailable, passes through unmodified
+- Keep cache_metrics() for observability (drift tracking)
-- Invalid content type → passthrough
+- Simplify PromptBuilder
 
-✅ **Thread Safety**:
+## Test Summary
-- Mutex-wrapped Magika Session
-- once_cell Lazy statics for patterns
-- IndexMap for LRU cache
 
-✅ **Zero Breaking Changes**:
+✅ **62 Unit Tests** (all passing)
-- All existing prompt tests pass
+- Phase 1: 17 (router, log)
-- Backward compatible env var defaults
+- Phase 2: 15 (json, diff)
-- Optional optimization (can disable globally)
+- Phase 3: 18 (cache align, CCR)
+- Phase 4: 12 (text, config)
 
-✅ **Compression Targets Met**:
+⏳ **20 New Tests Pending** (M3.8.2-3)
-- Logs: 85-95% compression
+- Ingest source optimization
-- JSON: 70-90% compression
+- Metrics collection
-- Diffs: 60-80% compression
+- End-to-end pipeline
-- Text: 30-50% compression
-- All < 10ms per chunk
 
-## Key Files
+## Files
 
-**Source**:
+**Implemented** (1,100 LOC):
-- `crates/mem-core/src/optimizer/mod.rs` (150 LOC)
+- `crates/mem-core/src/optimizer/mod.rs`
-- `crates/mem-core/src/optimizer/router.rs` (213 LOC)
+- `crates/mem-core/src/optimizer/router.rs`
-- `crates/mem-core/src/optimizer/log.rs` (220 LOC)
+- `crates/mem-core/src/optimizer/log.rs`
-- `crates/mem-core/src/optimizer/json.rs` (325 LOC)
+- `crates/mem-core/src/optimizer/json.rs`
-- `crates/mem-core/src/optimizer/diff.rs` (270 LOC)
+- `crates/mem-core/src/optimizer/diff.rs`
-- `crates/mem-core/src/optimizer/text.rs` (365 LOC)
+- `crates/mem-core/src/optimizer/text.rs`
-- `crates/mem-core/src/optimizer/cache_align.rs` (210 LOC)
+- `crates/mem-core/src/optimizer/cache_align.rs`
-- `crates/mem-core/src/optimizer/ccr.rs` (215 LOC)
+- `crates/mem-core/src/optimizer/ccr.rs`
 
-**Integration**:
+**Pending** (180 LOC):
-- `crates/mem-core/src/prompt.rs` (updated, +12 lines)
+- `crates/mem-ingest/src/optimizer_sink.rs` (M3.8.2)
-- `crates/mem-core/src/lib.rs` (updated, +2 exports)
+- `crates/mem-core/src/optimizer/metrics.rs` (M3.8.3)
-- `crates/mem-core/Cargo.toml` (added magika, ort, regex, once_cell, indexmap, lazy_static)
 
-**Tests**:
+## Commits
-- `crates/mem-core/src/optimizer/` (62 unit tests in modules)
-- All integration tests in PromptBuilder
 
-## Status
+1. `bf13e3a` — Phase 1: ContentRouter + LogCompressor
+2. `a903a3f` — Phase 2: JsonCrusher + DiffCompressor
+3. `edcc231` — Phase 3: CacheAligner + CcrStore
+4. `8d8addc` — Phase 4: TextCompressor + env config
 
-✅ **Complete and Production-Ready**
+## Lessons Learned
 
-- All phases implemented and tested
+1. **Ingest-time optimization > query-time**: Better for entire pipeline
-- Full PromptBuilder integration
+2. **Compression ratios vary widely**: Log 85-95% vs text 30-50%
-- Environment-configurable
+3. **Reversibility matters**: Model needs originals for detailed analysis
-- Comprehensive error handling
+4. **Metrics > assumption**: Need to measure actual improvement in search quality
-- Performance targets met (<10ms per chunk)
-- Zero breaking changes
 
-**Ready for**: M3.8.2 (CacheAligner header integration), M3.8.3 (benchmarking), M3.8.4 (gate)
+## Remediation
+
+See **`tasks/M3.8-CORRECTED-architecture.md`** for complete re-architecture plan.

tasks/M3.8.2-cache-aligner-headers.md

@@ -1,81 +1,126 @@
-# M3.8.2 — CacheAligner: HTTP Headers + Metrics
+# M3.8.2 — Ingest Pipeline Integration (OptimizerSink)
 
 | Field | Value |
 |---|---|
 | Phase | M3.8 — Context optimization |
 | Size | M — 1–2 days |
-| Status | ✅ COMPLETE |
+| Status | ⬜ Not started |
-| Depends | M3.8.1 (CacheAligner complete) |
+| Depends | M3.8.1 (core modules) |
 | Blocks | M3.8.3 |
 
 ## Goal
 
-Integrate CacheAligner output into HTTP response headers and observability metrics
+Wire M3.8 compressors into the ingest pipeline so that chunks are optimized
-so that:
+BEFORE embedding + indexing, resulting in:
-1. LLM provider can use cache hints
+- Better embeddings (clean text)
-2. Monitoring can track cache effectiveness
+- Better search ranking (signal-rich documents)
-3. Debugging can identify cache misses
+- Cleaner results for all queries
 
 ## Deliverables
 
-### 1. PromptBuilder::cache_metrics()
+### 1. OptimizerSink Wrapper (100 LOC)
 
-New method returning cache metadata:
+New module: `crates/mem-ingest/src/optimizer_sink.rs`
 
 ```rust
-pub struct CacheMetrics {
+pub struct OptimizerSink {
-    pub stable_prefix_bytes: usize,
+    inner: Box<dyn RecordSource>,
-    pub dynamic_tail_bytes: usize,
+    optimizer: ContextOptimizer,
-    pub drift_metric: f32,  // 0.0-1.0 ratio
+    config: OptimizerConfig,
-    pub cache_eligible: bool,  // true if drift < 0.3
+    metrics: MetricsCollector,
 }
 
-impl PromptBuilder {
+impl RecordSource for OptimizerSink {
-    pub fn cache_metrics(query: &Query, chunk: &Chunk) -> Result<CacheMetrics>
+    fn next_record(&mut self) -> Option<Record> {
+        let record = self.inner.next_record()?;
+        let optimized = self.optimizer.optimize(&record.content)?;
+        
+        // Track metrics
+        self.metrics.record(OptimizationMetrics {
+            input_bytes: record.content.len(),
+            output_bytes: optimized.compressed.len(),
+            compressor: optimized.compressor_used,
+            ..
+        });
+        
+        // Emit optimized chunk
+        Some(Record {
+            content: optimized.compressed,
+            ..record
+        })
+    }
 }
+
+pub fn optimize_source(
+    source: Box<dyn RecordSource>,
+    project: &str,
+) -> Result<OptimizerSink>
 ```
 
 Tests (3):
-- `test_cache_metrics_stable_query`
+- `test_optimizer_sink_preserves_structure`
-- `test_cache_metrics_high_drift`
+- `test_optimizer_sink_reduces_bytes`
-- `test_cache_metrics_zero_drift`
+- `test_optimizer_sink_handles_errors`
 
-### 2. HTTP Response Headers
+### 2. Rebuild Integration (30 LOC)
 
-Add to PromptBuilder output:
+Modify: `crates/mem-store/src/rebuild.rs`
-- `X-Cache-Stable-Bytes`: size of cacheable prefix
-- `X-Cache-Drift`: 0.0-1.0 ratio
-- `X-Cache-Eligible`: "true"/"false"
-- `X-Compression-Ratio`: original vs. compressed
 
-Tests (4):
-- `test_headers_present_in_response`
-- `test_headers_accurate_values`
-- `test_headers_skipped_when_disabled`
-- `test_headers_format_valid`
-
-### 3. Observability Hooks
-
-Integrate with logging:
 ```rust
-pub fn log_cache_metrics(metrics: &CacheMetrics) {
+let source = DocCorpusSource::new(vault_path)?;
-    tracing::info!(
+let optimized = optimize_source(Box::new(source), &project)?;  // ← NEW
-        stable_bytes = metrics.stable_prefix_bytes,
+
-        drift = metrics.drift_metric,
+for record in optimized {
-        eligible = metrics.cache_eligible,
+    let embedding = embed(&record.content)?;  // clean text
-        "cache_alignment"
+    insert_pgvector(embedding, &record)?;
-    );
+    insert_opensearch(&record)?;
 }
 ```
 
-Tests (2):
+Tests (4):
-- `test_metrics_logged_on_alignment`
+- `test_rebuild_with_optimizer_enabled`
-- `test_drift_high_triggers_warning`
+- `test_rebuild_with_optimizer_disabled`
+- `test_rebuild_compression_ratio`
+- `test_rebuild_pgvector_quality_improves`
+
+### 3. Source Integration Tests (150 LOC)
+
+New module: `tests/it_ingest_optimizer.rs`
+
+Test each ingest source with optimizer:
+- `test_pi_session_source_optimized` (Claude transcripts)
+- `test_doc_corpus_source_optimized` (markdown files)
+- `test_claude_transcript_source_optimized` (agent logs)
+- `test_optimizer_preserves_breadcrumb` (M3.6.1 paths)
+- `test_optimizer_respects_level` (L0/L1/L2)
+- `test_optimizer_disabled_via_env` (MEM_CONTEXT_OPTIMIZER=off)
+
+Tests (6):
+- Per-source integration tests
+
+### 4. Metrics Collection (NEW)
+
+Modified: `crates/mem-ingest/src/lib.rs`
+
+Export MetricsCollector from OptimizerSink:
+```rust
+pub struct OptimizerMetrics {
+    pub input_bytes: usize,
+    pub output_bytes: usize,
+    pub compression_ratio: f32,
+    pub compressor_used: String,
+    pub timestamp: i64,
+    pub project: String,
+}
+```
+
+No new tests (M3.8.3 handles metrics comprehensively)
 
 ## Acceptance
 
-- All 9 new tests passing
+✅ All 13 new tests passing
-- Existing 114 mem-core tests still pass
+✅ OptimizerSink integrated with rebuild.rs
-- Cache metrics accurately reflect alignment
+✅ All ingest sources work with optimizer
-- HTTP headers present and valid
+✅ Metrics collected (no performance regression <1ms per chunk)
-- Zero performance overhead (< 1ms additional)
+✅ Backward compatible (optimizer disableable via env)
+✅ Compression ratios match targets (log 85-95%, json 70-90%, etc.)