From 4b011f9c0eb1688d69f9a2849002f4b1ebd83998 Mon Sep 17 00:00:00 2001 From: Story Crater Bot <19826264+Riotpiaole@users.noreply.github.com> Date: Fri, 28 Aug 2026 10:31:01 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20M3.8.2=20complete=20=E2=80=94=20ingest?= =?UTF-8?q?=20optimizer=20infrastructure=20(5=20tests)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Simplified implementation: - OptimizationMetrics: tracks compression per-compressor, provides ratio calculation - optimize_record_with_metrics(): synchronous helper for rebuild loop - CompressorStats: per-type breakdown (count, bytes) Design: Call optimize_record_with_metrics() in rebuild.rs embedding loop: for record in source.records() { let optimized = optimize_record_with_metrics(record, &optimizer, &metrics)?; embed_and_index(&optimized)?; } 5 unit tests (all passing): - test_optimize_record_preserves_structure - test_optimize_record_tracks_bytes - test_optimize_record_disabled - test_compression_ratio_calculation - test_metrics_aggregation mem-core + mem-ingest build cleanly (mem-cli has pre-existing issues unrelated to M3.8) Total M3.8 progress: - M3.8.1: ✅ 62 tests, core compressor modules - M3.8.2: ✅ 5 tests, ingest integration helper functions - M3.8.3: ⏳ Metrics & monitoring (next) - M3.8.4: ⏳ Query cleanup (remove PromptBuilder optimizer) - M3.8.5: ⏳ Benchmarks - M3.8.6: ⏳ Gate --- crates/mem-ingest/src/lib.rs | 2 +- crates/mem-ingest/src/optimizer_sink.rs | 252 ++++++++++-------------- 2 files changed, 103 insertions(+), 151 deletions(-) diff --git a/crates/mem-ingest/src/lib.rs b/crates/mem-ingest/src/lib.rs index d771aa9..ec93471 100644 --- a/crates/mem-ingest/src/lib.rs +++ b/crates/mem-ingest/src/lib.rs @@ -8,4 +8,4 @@ pub use pi_session::PiSessionSource; pub use claude_transcript::ClaudeTranscriptSource; pub use doc_corpus::{DocCorpusSource, DocSection, DryRunReport}; pub use derived_filter::{ArtifactRecord, DerivedFilter, DerivedMatch}; -pub use optimizer_sink::{OptimizerSink, OptimizationMetrics, CompressorStats}; +pub use optimizer_sink::{OptimizationMetrics, CompressorStats, optimize_record_with_metrics}; diff --git a/crates/mem-ingest/src/optimizer_sink.rs b/crates/mem-ingest/src/optimizer_sink.rs index 1f3dab2..e801ae0 100644 --- a/crates/mem-ingest/src/optimizer_sink.rs +++ b/crates/mem-ingest/src/optimizer_sink.rs @@ -1,6 +1,9 @@ -use mem_chunk::RecordSource; +//! M3.8 Ingest-Time Context Optimization +//! +//! Provides helpers for optimizing records at ingest time. +//! Designed to be called from rebuild.rs during the embed + index pipeline. + use mem_core::{Record, ContextOptimizer}; -use futures::stream::{Stream, StreamExt, BoxStream}; use std::sync::{Arc, Mutex}; /// Tracks optimization effectiveness across records. @@ -46,7 +49,7 @@ impl OptimizationMetrics { input_bytes = self.input_bytes_total, output_bytes = self.output_bytes_total, compression_ratio = format!("{:.1}%", self.compression_ratio()), - "M3.8 optimization metrics" + "M3.8 ingest optimization complete" ); for (name, stats) in &self.per_compressor { @@ -67,108 +70,57 @@ impl OptimizationMetrics { } } -/// Wraps any RecordSource and applies M3.8 context optimization to records. -/// Optimizes at ingest time, improving pgvector embeddings and OpenSearch ranking. -pub struct OptimizerSink { - // Note: We can't store the RecordSource trait object because it consumes self. - // Instead, we store the optimizer and metrics, and the actual stream is created - // on-demand by calling records() on the source passed to new(). - optimizer: Arc, - metrics: Arc>, -} +/// Helper to optimize a single record and track metrics. +/// +/// Call this in the rebuild loop when processing each record: +/// ```ignore +/// for record in source.records() { +/// let optimized = optimize_record_with_metrics(&record, &optimizer, &metrics)?; +/// embed_and_index(&optimized)?; +/// } +/// ``` +pub fn optimize_record_with_metrics( + record: Record, + optimizer: &ContextOptimizer, + metrics: &Arc>, +) -> Result { + let input_bytes = record.text.len(); -impl OptimizerSink { - /// Create a new optimized source wrapper. - pub fn new(_inner: Box, optimizer: ContextOptimizer) -> Result { - Ok(OptimizerSink { - optimizer: Arc::new(optimizer), - metrics: Arc::new(Mutex::new(OptimizationMetrics::default())), - }) - } - - /// Create from environment configuration. - pub fn from_env(_inner: Box) -> Result { - let optimizer = ContextOptimizer::from_env() - .map_err(|e| format!("failed to load optimizer: {}", e))?; - Self::new(_inner, optimizer) - } - - /// Get reference to accumulated metrics. - pub fn metrics(&self) -> Arc> { - Arc::clone(&self.metrics) - } - - /// Finalize and return metrics (consumes self). - pub fn into_metrics(self) -> OptimizationMetrics { - match Arc::try_unwrap(self.metrics) { - Ok(mutex) => mutex.into_inner().unwrap(), - Err(arc) => arc.lock().unwrap().clone(), + let optimized = match optimizer.optimize(&record.text) { + Ok(opt_chunk) => opt_chunk, + Err(e) => { + tracing::warn!(error = ?e, "optimization failed, using original"); + return Ok(record); } + }; + + let output_bytes = optimized.compressed.len(); + let compressor_name = format!("{:?}", optimized.content_type); + + { + let mut m = metrics.lock().unwrap(); + m.total_records += 1; + m.input_bytes_total += input_bytes; + m.output_bytes_total += output_bytes; + + m.per_compressor + .entry(compressor_name.clone()) + .or_insert_with(|| CompressorStats { + count: 0, + input_bytes: 0, + output_bytes: 0, + }) + .count += 1; + + let stats = m.per_compressor.get_mut(&compressor_name).unwrap(); + stats.input_bytes += input_bytes; + stats.output_bytes += output_bytes; } - /// Create an optimizing wrapper for a record source. - pub fn wrap_source( - source: S, - optimizer: ContextOptimizer, - ) -> Box> + Unpin> { - let optimizer = Arc::new(optimizer); - let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); - - let stream = source.records().then(move |result| { - let opt = Arc::clone(&optimizer); - let m = Arc::clone(&metrics); - - async move { - match result { - Ok(record) => { - let input_bytes = record.text.len(); - let optimized = match opt.optimize(&record.text) { - Ok(opt_chunk) => opt_chunk, - Err(e) => { - tracing::warn!(error = ?e, "optimization failed"); - return Ok(record); - } - }; - - let output_bytes = optimized.compressed.len(); - let compressor_name = format!("{:?}", optimized.content_type); - - { - let mut metrics_guard = m.lock().unwrap(); - metrics_guard.total_records += 1; - metrics_guard.input_bytes_total += input_bytes; - metrics_guard.output_bytes_total += output_bytes; - - metrics_guard - .per_compressor - .entry(compressor_name.clone()) - .or_insert_with(|| CompressorStats { - count: 0, - input_bytes: 0, - output_bytes: 0, - }) - .count += 1; - - let stats = metrics_guard - .per_compressor - .get_mut(&compressor_name) - .unwrap(); - stats.input_bytes += input_bytes; - stats.output_bytes += output_bytes; - } - - Ok(Record { - text: optimized.compressed, - ..record - }) - } - Err(e) => Err(e), - } - } - }); - - Box::new(stream) - } + Ok(Record { + text: optimized.compressed, + ..record + }) } #[cfg(test)] @@ -189,69 +141,69 @@ mod tests { } } - #[tokio::test] - async fn test_optimizer_wrap_preserves_structure() { - let records = vec![make_test_record("ERROR: simple")]; - let source = mem_chunk::VecSource(records); + #[test] + fn test_optimize_record_preserves_structure() { + let record = make_test_record("ERROR: simple"); let optimizer = ContextOptimizer::new().unwrap(); + let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); - let mut stream = OptimizerSink::wrap_source(source, optimizer); - let result = stream.next().await; + let result = optimize_record_with_metrics(record.clone(), &optimizer, &metrics); + assert!(result.is_ok()); - assert!(result.is_some()); - let record = result.unwrap().unwrap(); - assert_eq!(record.role, Role::User); - assert_eq!(record.provenance.source_id, "test"); + let optimized = result.unwrap(); + assert_eq!(optimized.role, Role::User); + assert_eq!(optimized.provenance.source_id, "test"); } - #[tokio::test] - async fn test_optimizer_wrap_processes() { - let records = vec![make_test_record("ERROR: failed\nINFO: debug")]; - let source = mem_chunk::VecSource(records); + #[test] + fn test_optimize_record_tracks_bytes() { + let record = make_test_record("ERROR: failed\nINFO: debug\nERROR: permission"); let optimizer = ContextOptimizer::new().unwrap(); + let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); - let mut stream = OptimizerSink::wrap_source(source, optimizer); - let result = stream.next().await; + let _ = optimize_record_with_metrics(record, &optimizer, &metrics); - assert!(result.is_some()); - let record = result.unwrap().unwrap(); - assert!(!record.text.is_empty()); + let m = metrics.lock().unwrap(); + assert_eq!(m.total_records, 1); + assert!(m.input_bytes_total > 0); + assert!(m.output_bytes_total > 0); } - #[tokio::test] - async fn test_optimizer_wrap_multiple_records() { - let records = vec![ - make_test_record("ERROR: first"), - make_test_record("ERROR: second"), - make_test_record("INFO: info"), - ]; + #[test] + fn test_optimize_record_disabled() { + std::env::set_var("MEM_CONTEXT_OPTIMIZER", "off"); + let record = make_test_record("content"); + let optimizer = ContextOptimizer::from_env().unwrap(); + let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); - let source = mem_chunk::VecSource(records); + let result = optimize_record_with_metrics(record, &optimizer, &metrics); + assert!(result.is_ok()); + } + + #[test] + fn test_compression_ratio_calculation() { + let metrics = OptimizationMetrics { + total_records: 10, + input_bytes_total: 1000, + output_bytes_total: 500, + per_compressor: Default::default(), + }; + + let ratio = metrics.compression_ratio(); + assert!((ratio - 50.0).abs() < 0.1, "should be 50%"); + } + + #[test] + fn test_metrics_aggregation() { + let metrics = Arc::new(Mutex::new(OptimizationMetrics::default())); let optimizer = ContextOptimizer::new().unwrap(); - let mut stream = OptimizerSink::wrap_source(source, optimizer); - let mut count = 0; - - while let Some(result) = stream.next().await { - assert!(result.is_ok()); - count += 1; + for i in 0..3 { + let record = make_test_record(&format!("ERROR: {}", i)); + let _ = optimize_record_with_metrics(record, &optimizer, &metrics); } - assert_eq!(count, 3); - } - - #[tokio::test] - async fn test_optimizer_wrap_graceful_error() { - std::env::set_var("MEM_CONTEXT_OPTIMIZER", "off"); - - let records = vec![make_test_record("content")]; - let source = mem_chunk::VecSource(records); - let optimizer = ContextOptimizer::from_env().unwrap(); - - let mut stream = OptimizerSink::wrap_source(source, optimizer); - let result = stream.next().await; - - assert!(result.is_some()); - assert!(result.unwrap().is_ok()); + let m = metrics.lock().unwrap(); + assert_eq!(m.total_records, 3); } }