headroomlabs-ai
diff --git a/‎crates/headroom-core/src/ccr/mod.rs‎
Lines changed: 5 additions & 2 deletions b/‎crates/headroom-core/src/ccr/mod.rs‎
Lines changed: 5 additions & 2 deletions
diff --git a/‎crates/headroom-core/src/transforms/search_compressor.rs‎
Lines changed: 28 additions & 3 deletions b/‎crates/headroom-core/src/transforms/search_compressor.rs‎
Lines changed: 28 additions & 3 deletions
diff --git a/‎crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs‎
Lines changed: 10 additions & 0 deletions b/‎crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎crates/headroom-core/src/transforms/smart_crusher/config.rs‎
Lines changed: 37 additions & 3 deletions b/‎crates/headroom-core/src/transforms/smart_crusher/config.rs‎
Lines changed: 37 additions & 3 deletions
diff --git a/‎crates/headroom-core/src/transforms/smart_crusher/crusher.rs‎
Lines changed: 124 additions & 6 deletions b/‎crates/headroom-core/src/transforms/smart_crusher/crusher.rs‎
Lines changed: 124 additions & 6 deletions
diff --git a/‎crates/headroom-parity/src/lib.rs‎
Lines changed: 4 additions & 0 deletions b/‎crates/headroom-parity/src/lib.rs‎
Lines changed: 4 additions & 0 deletions
@@ -59,8 +59,11 @@ pub trait CcrStore: Send + Sync {
 /// Default capacity — matches Python's `CompressionStore` default.
 pub const DEFAULT_CAPACITY: usize = 1000;
 
-/// Default TTL — 5 minutes, matching Python.
-pub const DEFAULT_TTL: Duration = Duration::from_secs(300);
+/// Default TTL — 30 minutes, matching Python
+/// (`CCRConfig.store_ttl_seconds`). Session-scale: agentic sessions
+/// routinely outlive the old 5-minute default, and an expired entry
+/// silently converts "lossless with retrieval" into "lossy".
+pub const DEFAULT_TTL: Duration = Duration::from_secs(1800);
 
 /// Compute the canonical CCR key for `payload`. BLAKE3 → first 24 hex
 /// chars (96 bits — collision-resistant for the bounded LRU population
 
@@ -140,6 +140,14 @@ pub struct SearchCompressorConfig {
     /// to a config field here (Python had it inline) so a future
     /// pipeline can tune per-content-type.
     pub min_compression_ratio_for_ccr: f64,
+    /// Group output by file (`rg --heading` style): emit each file path
+    /// once as a header line, then `line:content` rows beneath it, with
+    /// a blank line between file groups. Eliminates per-match path
+    /// repetition — the dominant remaining token waste on large result
+    /// sets (a 70-char path repeated 15× is ~250 wasted tokens).
+    /// Default `false` (classic `file:line:content`) for parity; the
+    /// proxy enables it in token mode.
+    pub group_by_file: bool,
 }
 
 impl Default for SearchCompressorConfig {
@@ -155,6 +163,7 @@ impl Default for SearchCompressorConfig {
             enable_ccr: true,
             min_matches_for_ccr: 10,
             min_compression_ratio_for_ccr: 0.8,
+            group_by_file: false,
         }
     }
 }
@@ -527,15 +536,31 @@ impl SearchCompressor {
     ) -> (String, BTreeMap<String, String>) {
         let mut lines: Vec<String> = Vec::new();
         let mut summaries: BTreeMap<String, String> = BTreeMap::new();
+        let grouped = self.config.group_by_file;
 
         for (file, fm) in selected {
-            for m in &fm.matches {
-                lines.push(format!("{}:{}:{}", m.file, m.line_number, m.content));
+            if grouped {
+                // `rg --heading` style: path once, then line:content rows.
+                if !lines.is_empty() {
+                    lines.push(String::new());
+                }
+                lines.push(file.clone());
+                for m in &fm.matches {
+                    lines.push(format!("{}:{}", m.line_number, m.content));
+                }
+            } else {
+                for m in &fm.matches {
+                    lines.push(format!("{}:{}:{}", m.file, m.line_number, m.content));
+                }
             }
             if let Some(orig_fm) = original.get(file) {
                 if orig_fm.matches.len() > fm.matches.len() {
                     let omitted = orig_fm.matches.len() - fm.matches.len();
-                    let summary = format!("[... and {} more matches in {}]", omitted, file);
+                    let summary = if grouped {
+                        format!("[... and {} more matches]", omitted)
+                    } else {
+                        format!("[... and {} more matches in {}]", omitted, file)
+                    };
                     lines.push(summary.clone());
                     summaries.insert(file.clone(), summary);
                 }
 
@@ -59,6 +59,16 @@ impl CompactionStage {
         }
     }
 
+    /// CSV+schema formatter with an explicit config. Used by
+    /// `SmartCrusher::new` to honor the compaction heuristics carried
+    /// on `SmartCrusherConfig` instead of pinning `CompactConfig::default()`.
+    pub fn csv_schema(config: CompactConfig) -> Self {
+        Self {
+            config,
+            formatter: Box::new(CsvSchemaFormatter::new()),
+        }
+    }
+
     /// JSON formatter, default config — useful for debugging or for
     /// downstream consumers that want structured rather than CSV-shaped
     /// output.
 
@@ -53,7 +53,10 @@ pub struct SmartCrusherConfig {
     /// path to be chosen over lossy. Computed as
     /// `1 - len(rendered) / len(input)`. If lossless saves less than
     /// this fraction, `crush_array` falls through to the lossy path
-    /// (with CCR-Dropped retrieval markers). Default `0.30`.
+    /// (with CCR-Dropped retrieval markers). Default `0.15` — kept in
+    /// lockstep with the Python `SmartCrusherConfig` dataclass default
+    /// (lowered from 0.30 so cleanly tabular input takes the lossless
+    /// path more often; lossless needs no CCR retrieval round-trip).
     ///
     /// **Override semantics.** OSS users can tune this via the config
     /// directly. Enterprise plug-ins replace the entire decision via
@@ -77,6 +80,27 @@ pub struct SmartCrusherConfig {
     /// still emit always; they have no Python equivalent and no
     /// production caller has asked for them to be suppressed.
     pub enable_ccr_marker: bool,
+    /// Compaction heuristic: a field is "core" if it appears in at
+    /// least this fraction of rows. Mirrors
+    /// `CompactConfig::core_field_fraction`. Default 0.8.
+    pub compaction_core_field_fraction: f64,
+    /// Compaction heuristic: when fewer than this fraction of all
+    /// observed keys are core, treat the array as heterogeneous and
+    /// look for a discriminator. Mirrors
+    /// `CompactConfig::heterogeneous_core_ratio`. Default 0.6.
+    pub compaction_heterogeneous_core_ratio: f64,
+    /// Compaction heuristic: cap on inner-key count for
+    /// nested-uniform flattening. Mirrors
+    /// `CompactConfig::max_flatten_inner_keys`. Default 6.
+    pub compaction_max_flatten_inner_keys: usize,
+    /// Compaction heuristic: minimum bucket count before a candidate
+    /// discriminator is "useful". Mirrors `CompactConfig::min_buckets`.
+    /// Default 2.
+    pub compaction_min_buckets: usize,
+    /// Compaction heuristic: maximum bucket count — too many buckets
+    /// means the discriminator is too granular (e.g. an ID column).
+    /// Mirrors `CompactConfig::max_buckets`. Default 8.
+    pub compaction_max_buckets: usize,
 }
 
 impl Default for SmartCrusherConfig {
@@ -101,8 +125,13 @@ impl Default for SmartCrusherConfig {
             first_fraction: 0.3,
             last_fraction: 0.15,
             relevance_threshold: 0.3,
-            lossless_min_savings_ratio: 0.30,
+            lossless_min_savings_ratio: 0.15,
             enable_ccr_marker: true,
+            compaction_core_field_fraction: 0.8,
+            compaction_heterogeneous_core_ratio: 0.6,
+            compaction_max_flatten_inner_keys: 6,
+            compaction_min_buckets: 2,
+            compaction_max_buckets: 8,
         }
     }
 }
@@ -133,7 +162,12 @@ mod tests {
         assert_eq!(c.first_fraction, 0.3);
         assert_eq!(c.last_fraction, 0.15);
         assert_eq!(c.relevance_threshold, 0.3);
-        assert_eq!(c.lossless_min_savings_ratio, 0.30);
+        assert_eq!(c.lossless_min_savings_ratio, 0.15);
         assert!(c.enable_ccr_marker);
+        assert_eq!(c.compaction_core_field_fraction, 0.8);
+        assert_eq!(c.compaction_heterogeneous_core_ratio, 0.6);
+        assert_eq!(c.compaction_max_flatten_inner_keys, 6);
+        assert_eq!(c.compaction_min_buckets, 2);
+        assert_eq!(c.compaction_max_buckets, 8);
     }
 }
@@ -42,7 +42,7 @@ use super::builder::SmartCrusherBuilder;
 use super::classifier::{classify_array, ArrayType};
 use super::compaction::{
     classify_cell, emit_opaque_ccr_marker, try_parse_json_container, CellClass, ClassifyConfig,
-    Compaction, CompactionStage,
+    CompactConfig, Compaction, CompactionStage,
 };
 use super::config::SmartCrusherConfig;
 use super::crushers::{compute_k_split, crush_number_array, crush_object, crush_string_array};
@@ -154,9 +154,20 @@ impl SmartCrusher {
     /// CCR cache, not to nowhere — same semantics as Python's
     /// SmartCrusher with CCR enabled.
     pub fn new(config: SmartCrusherConfig) -> Self {
+        // Carry the compaction heuristics from the crusher config into
+        // the compaction stage; everything not exposed on
+        // SmartCrusherConfig keeps its CompactConfig default.
+        let compact_cfg = CompactConfig {
+            core_field_fraction: config.compaction_core_field_fraction,
+            heterogeneous_core_ratio: config.compaction_heterogeneous_core_ratio,
+            max_flatten_inner_keys: config.compaction_max_flatten_inner_keys,
+            min_buckets: config.compaction_min_buckets,
+            max_buckets: config.compaction_max_buckets,
+            ..CompactConfig::default()
+        };
         SmartCrusherBuilder::new(config)
             .with_default_oss_setup()
-            .with_default_compaction()
+            .with_compaction(CompactionStage::csv_schema(compact_cfg))
             .with_default_ccr_store()
             .build()
     }
@@ -263,16 +274,49 @@ impl SmartCrusher {
     /// kept-items list in original-array order. Mirrors Python's
     /// `_execute_plan` (line 3617-3633).
     ///
-    /// Schema-preserving: each kept item is cloned unchanged. No
-    /// summary objects, generated fields, or wrapper metadata.
+    /// Schema-preserving by default: each kept item is cloned unchanged.
+    /// No summary objects, generated fields, or wrapper metadata.
+    ///
+    /// When `factor_out_constants` is enabled (default off), fields the
+    /// analyzer found constant across ALL items are stripped from each
+    /// kept object and emitted once in a leading
+    /// `{"_constant_fields": {...}}` sentinel — same output-shape
+    /// convention as the `_ccr_dropped` sentinel. Stripping is
+    /// defensive: a key is only removed from an item when its value
+    /// equals the recorded constant, so a drifted item keeps its own
+    /// value. The CCR store always holds the full unfactored original.
     pub fn execute_plan(&self, plan: &CompressionPlan, items: &[Value]) -> Vec<Value> {
         let mut indices = plan.keep_indices.clone();
         indices.sort_unstable();
-        indices
+        let mut kept: Vec<Value> = indices
             .into_iter()
             .filter(|&idx| idx < items.len())
             .map(|idx| items[idx].clone())
-            .collect()
+            .collect();
+
+        if self.config.factor_out_constants && !plan.constant_fields.is_empty() && kept.len() >= 2 {
+            let mut any_stripped = false;
+            for item in kept.iter_mut() {
+                if let Value::Object(map) = item {
+                    for (key, constant) in &plan.constant_fields {
+                        if map.get(key) == Some(constant) {
+                            map.remove(key);
+                            any_stripped = true;
+                        }
+                    }
+                }
+            }
+            if any_stripped {
+                let mut sentinel = serde_json::Map::new();
+                sentinel.insert(
+                    "_constant_fields".to_string(),
+                    Value::Object(plan.constant_fields.clone().into_iter().collect()),
+                );
+                kept.insert(0, Value::Object(sentinel));
+            }
+        }
+
+        kept
     }
 
     /// Top-level entry point. Mirrors Python `SmartCrusher.crush`
@@ -1023,6 +1067,80 @@ mod tests {
         assert_eq!(result.len(), 2);
     }
 
+    #[test]
+    fn execute_plan_factors_constants_when_enabled() {
+        let cfg = SmartCrusherConfig {
+            factor_out_constants: true,
+            ..Default::default()
+        };
+        let c = SmartCrusher::new(cfg);
+        let items: Vec<Value> = (0..4)
+            .map(|i| json!({"id": i, "region": "us-west-2", "status": "ok"}))
+            .collect();
+        let mut constant_fields = std::collections::BTreeMap::new();
+        constant_fields.insert("region".to_string(), json!("us-west-2"));
+        constant_fields.insert("status".to_string(), json!("ok"));
+        let plan = CompressionPlan {
+            keep_indices: vec![0, 1, 2],
+            constant_fields,
+            ..CompressionPlan::default()
+        };
+        let result = c.execute_plan(&plan, &items);
+        // Sentinel first, then 3 slim items.
+        assert_eq!(result.len(), 4);
+        assert_eq!(result[0]["_constant_fields"]["region"], "us-west-2");
+        assert_eq!(result[0]["_constant_fields"]["status"], "ok");
+        for item in &result[1..] {
+            assert!(item.get("region").is_none());
+            assert!(item.get("status").is_none());
+            assert!(item.get("id").is_some());
+        }
+    }
+
+    #[test]
+    fn execute_plan_keeps_drifted_values_when_factoring() {
+        // Defensive strip: an item whose value differs from the recorded
+        // constant keeps its own value.
+        let cfg = SmartCrusherConfig {
+            factor_out_constants: true,
+            ..Default::default()
+        };
+        let c = SmartCrusher::new(cfg);
+        let items = vec![
+            json!({"id": 0, "status": "ok"}),
+            json!({"id": 1, "status": "FAILED"}),
+        ];
+        let mut constant_fields = std::collections::BTreeMap::new();
+        constant_fields.insert("status".to_string(), json!("ok"));
+        let plan = CompressionPlan {
+            keep_indices: vec![0, 1],
+            constant_fields,
+            ..CompressionPlan::default()
+        };
+        let result = c.execute_plan(&plan, &items);
+        assert_eq!(result.len(), 3);
+        assert!(result[1].get("status").is_none()); // matched → stripped
+        assert_eq!(result[2]["status"], "FAILED"); // drifted → kept
+    }
+
+    #[test]
+    fn execute_plan_default_off_leaves_items_unchanged() {
+        // factor_out_constants defaults to false: schema preserved even
+        // when the plan carries constant_fields.
+        let c = crusher();
+        let items: Vec<Value> = (0..3).map(|i| json!({"id": i, "k": "v"})).collect();
+        let mut constant_fields = std::collections::BTreeMap::new();
+        constant_fields.insert("k".to_string(), json!("v"));
+        let plan = CompressionPlan {
+            keep_indices: vec![0, 1, 2],
+            constant_fields,
+            ..CompressionPlan::default()
+        };
+        let result = c.execute_plan(&plan, &items);
+        assert_eq!(result.len(), 3);
+        assert_eq!(result[0]["k"], "v");
+    }
+
     // ---------- crush_array ----------
 
     #[test]
 
@@ -404,6 +404,10 @@ impl TransformComparator for SmartCrusherComparator {
                 .get("enable_ccr_marker")
                 .and_then(|v| v.as_bool())
                 .unwrap_or(defaults.enable_ccr_marker),
+            // Compaction heuristics are moot here: this comparator uses
+            // `without_compaction` (fixtures were recorded against the
+            // lossy-only path). Take the defaults wholesale.
+            ..defaults
         };
 
         // Use without_compaction so the legacy fixtures (recorded