akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(rust): SmartCrusher PR4 — lossless-first default + CCR-Dropped restoration

Browse source 
Stage 3c.2 PR4. Restores Python's CCR-Dropped semantics on the lossy
path (the cornerstone reversibility guarantee that the port had
silently dropped) and flips the OSS default to lossless-first with a
configurable savings threshold.

# The user-visible behavior

Default `SmartCrusher::new()` now runs:

  1. Try lossless compaction.
  2. If savings >= `lossless_min_savings_ratio` (default 0.30), ship
     it — `compacted` populated, `ccr_hash = None`, nothing dropped.
  3. Otherwise fall through to the lossy path — drop rows AND
     populate `ccr_hash` so the runtime can cache the full original
     for tool-call retrieval.

**No data is ever lost.** "Lossy" means "compressed view inline; full
payload retrievable via CCR cache" — same semantics as Python's
SmartCrusher with CCR enabled. The runtime (PyO3 bridge / proxy
server) owns the cache; this crate computes the hash and emits a
marker so the prompt knows where to look.

# What changed

- `SmartCrusherConfig.lossless_min_savings_ratio: f64` (default 0.30).
  Single configurable knob — Enterprise overrides as needed. Below
  the threshold, lossless declines and lossy + CCR runs.

- `SmartCrusher::new(cfg)` flips to include the compaction stage by
  default. `SmartCrusher::without_compaction(cfg)` is the explicit
  opt-out for callers / fixtures that depend on pre-PR4 behavior.

- `crush_array` rewritten:
  - Lossless-first dispatch with savings-ratio gate
  - Lossy path now hashes the full original (12-char SHA-256 prefix)
    and emits a CCR-Dropped marker in `dropped_summary` whenever
    rows are dropped
  - `ccr_hash` field populated whenever rows were dropped
  - `process_value` substitutes the compacted string into the JSON
    tree when lossless wins, so `crush()` output reflects the win

- PyO3 bridge: `SmartCrusher.without_compaction()` static method;
  `SmartCrusherConfig` exposes the new `lossless_min_savings_ratio`
  field; Python `SmartCrusher` wrapper accepts `with_compaction=True`
  (default) and routes to the right Rust constructor.

- Parity harness: legacy 17 fixtures use `without_compaction()` so
  byte-equal coverage of the lossy path is preserved.

# Tests

- Rust: 281/281 smart_crusher unit tests pass (was 277). Six new
  tests cover: lossless wins above threshold, lossy falls through
  below threshold, CCR hash deterministic + input-dependent, lossy
  without compaction emits CCR, passthrough paths don't emit CCR,
  without_compaction yields no compacted field.
- Python parity: 21/21 (legacy fixtures via without_compaction).
- Python lossless default smoke: 3/3 new tests in
  test_smart_crusher_lossless_default.py.
- Python retention: 21/21 (updated to opt into the lossy path
  explicitly since their semantics target row-level retention).
- make ci-precheck green.

Modules:
  crates/headroom-core/src/transforms/smart_crusher/{config,crusher}.rs
  crates/headroom-parity/src/lib.rs
  crates/headroom-py/src/lib.rs
  headroom/transforms/smart_crusher.py
  tests/test_quality_retention.py
  tests/test_transforms/test_smart_crusher_{lossless_default,rust_parity}.py
This commit is contained in:
chopratejas 2026-04-27 16:30:22 -07:00
parent 00e25d8475
commit 1601591900
8 changed files with 489 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

17
crates/headroom-core/src/transforms/smart_crusher/config.rs
View file

@ -49,11 +49,26 @@ pub struct SmartCrusherConfig {
/// planning methods. Mirrors Python's `RelevanceConfig.relevance_threshold`.
/// Default 0.3 — matches the Python default.
pub relevance_threshold: f64,
/// Minimum byte-savings ratio (0.0..1.0) for the lossless compaction
/// 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`.
///
/// **Override semantics.** OSS users can tune this via the config
/// directly. Enterprise plug-ins replace the entire decision via
/// a custom builder; the threshold is the OSS-default policy
/// expressed as a single knob. Set to `0.0` to always prefer
/// lossless when available; set to `1.0` to effectively disable
/// the lossless path (lossy + CCR always).
pub lossless_min_savings_ratio: f64,
}
impl Default for SmartCrusherConfig {
fn default() -> Self {
// These defaults must match smart_crusher.py:934-957 byte-for-byte.
// The PR4 additions (`lossless_min_savings_ratio`) have no
// Python counterpart — they govern Rust-side dispatch only.
SmartCrusherConfig {
enabled: true,
min_items_to_analyze: 5,
@ -71,6 +86,7 @@ impl Default for SmartCrusherConfig {
first_fraction: 0.3,
last_fraction: 0.15,
relevance_threshold: 0.3,
lossless_min_savings_ratio: 0.30,
}
}
}
@ -101,5 +117,6 @@ 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);
}
}

376
crates/headroom-core/src/transforms/smart_crusher/crusher.rs
View file

@ -48,30 +48,51 @@ use crate::relevance::RelevanceScorer;
use crate::transforms::adaptive_sizer::compute_optimal_k;
use crate::transforms::anchor_selector::AnchorSelector;
/// Return type for `crush_array` — mirrors Python's
/// `(crushed_items, strategy_info, ccr_hash, dropped_summary)` tuple.
/// Return type for `crush_array`.
///
/// Stage 3c.2 PR2 added `compacted` and `compaction_kind` for the new
/// lossless-first compaction stage. They stay `None` when no
/// [`CompactionStage`] is configured on the crusher (default), so
/// existing parity guarantees are preserved.
/// Two operating paths feed the same result type:
///
/// - **Lossless path** — input compacted to a smaller inline form
/// (e.g. CSV+schema). Nothing dropped; `compacted` is populated;
/// `ccr_hash` is `None` (no retrieval needed because everything is
/// already in the prompt).
/// - **Lossy path** — input compressed by row-dropping. `items` holds
/// the kept subset; `ccr_hash` is `Some(hash)` so the runtime can
/// cache the **full original** keyed by that hash and serve it back
/// to the LLM via a retrieval tool call. **No data is lost** —
/// "lossy" here means "compressed view inline; full payload cached
/// for tool retrieval," matching Python's CCR-Dropped semantics.
///
/// The runtime (PyO3 bridge / proxy server) owns the cache; this crate
/// computes the hash and emits a marker so the prompt knows where to
/// look.
pub struct CrushArrayResult {
/// Kept items. For the lossless path this is the full original
/// (nothing was dropped). For the lossy path this is the surviving
/// subset; the rest is retrievable via `ccr_hash`.
pub items: Vec<Value>,
/// Strategy debug string, e.g. `"smart_sample"`, `"top_n"`,
/// `"none:adaptive_at_limit"`, `"skip:unique_entities_no_signal"`.
/// When compaction wins, this is `"compaction:<formatter_name>"`.
/// Strategy debug string. One of:
/// - `"none:adaptive_at_limit"` / `"skip:<reason>"` — passthrough
/// - `"lossless:table"` / `"lossless:buckets"` — lossless wins
/// - `"smart_sample"` / `"top_n"` / `"cluster"` / `"time_series"` —
/// lossy path with row-dropping.
pub strategy_info: String,
/// CCR retrieval hash if caching is enabled. Stub: always `None`.
/// 12-char SHA-256 hex prefix of the **full original input**.
/// Populated when the lossy path dropped rows; the runtime is
/// expected to cache the original items keyed by this hash so a
/// retrieval tool can serve them back. `None` when nothing was
/// dropped (lossless path or below adaptive_k boundary).
pub ccr_hash: Option<String>,
/// Categorical summary of dropped items. Stub: always empty.
/// Marker text inserted into the prompt to advertise the CCR
/// pointer (e.g. `<<ccr:abc123def456 42_rows_offloaded>>`). Empty
/// when `ccr_hash` is `None`.
pub dropped_summary: String,
/// Rendered bytes from the compaction stage. `Some(s)` when the
/// stage was configured AND produced non-`Untouched` output;
/// `None` otherwise (default OSS path).
/// Rendered bytes from the compaction stage when the **lossless
/// path** won. `None` for the lossy path or when compaction wasn't
/// configured.
pub compacted: Option<String>,
/// Top-level [`Compaction`] variant tag — `"table"`, `"buckets"`,
/// `"ccr"`, or `"untouched"`. Mirrors `compacted` — populated only
/// when the compaction stage ran.
/// `"ccr"`. Mirrors `compacted` — populated only when lossless won.
pub compaction_kind: Option<&'static str>,
}
@ -101,11 +122,40 @@ pub struct SmartCrusher {
}
impl SmartCrusher {
/// Construct with the OSS default composition: `HybridScorer` +
/// `KeepErrorsConstraint` + `KeepStructuralOutliersConstraint` +
/// `TracingObserver`. Drop-in for callers that don't need
/// custom extensions — pre-PR1 behavior is preserved exactly.
/// Construct with the OSS default composition: scorer + constraints +
/// observer + **lossless-first compaction stage**. Calling
/// `crush_array` runs the dispatch:
///
/// 1. Try the lossless compactor.
/// 2. If savings ratio ≥ `config.lossless_min_savings_ratio`
/// (default `0.30`), ship lossless — `compacted` populated,
/// `ccr_hash = None`, nothing dropped.
/// 3. Otherwise fall through to the lossy path — drop rows,
/// populate `ccr_hash` with a hash of the full original so the
/// runtime can cache the payload for tool retrieval.
///
/// **No data is ever lost.** The lossy path moves dropped rows to
/// CCR cache, not to nowhere — same semantics as Python's
/// SmartCrusher with CCR enabled.
pub fn new(config: SmartCrusherConfig) -> Self {
SmartCrusherBuilder::new(config)
.with_default_oss_setup()
.with_default_compaction()
.build()
}
/// Construct WITHOUT the compaction stage. Pre-PR4 behavior:
/// `crush_array` skips the lossless attempt and runs the lossy
/// path directly (still with CCR-Dropped retrieval markers from
/// PR4). Used by:
///
/// - The 17 legacy parity fixtures (recorded against the
/// lossy-only path; using this constructor preserves byte-equal
/// coverage).
/// - Callers who explicitly don't want lossless attempts (e.g.
/// workloads where the compactor's overhead isn't worth the
/// modest tabular wins).
pub fn without_compaction(config: SmartCrusherConfig) -> Self {
SmartCrusherBuilder::new(config)
.with_default_oss_setup()
.build()
@ -293,6 +343,21 @@ impl SmartCrusher {
match arr_type {
ArrayType::DictArray => {
let result = self.crush_array(arr, query_context, bias);
// Lossless path won → substitute the array
// with the compacted string in place. This
// makes the lossless win visible to the
// public `crush()` API: the output JSON
// has a string where the array used to be.
// The wrapping JSON structure is preserved.
if let Some(rendered) = result.compacted {
info_parts.push(format!(
"{}({}->len={})",
result.strategy_info,
n,
rendered.len()
));
return (Value::String(rendered), info_parts.join(","));
}
info_parts.push(format!(
"{}({}->{})",
result.strategy_info,
@ -385,25 +450,6 @@ impl SmartCrusher {
/// 7. `execute_plan(plan, items)` → result.
/// 8. Strategy info = `analysis.recommended_strategy.as_str()`.
pub fn crush_array(&self, items: &[Value], query_context: &str, bias: f64) -> CrushArrayResult {
// ── Stage 0 (PR2): lossless-first compaction (opt-in) ──
//
// Runs only when a CompactionStage is configured. Sits BEFORE
// the existing lossy pipeline so the lossless path can
// short-circuit on success. When the compactor declines
// (returns Untouched) the rest of crush_array runs unchanged
// — preserving byte-equal parity with the pre-PR2 path.
let compaction_result: Option<(String, &'static str)> =
if let Some(stage) = &self.compaction {
let (c, rendered) = stage.run(items);
if c.was_compacted() {
Some((rendered, compaction_kind_str(&c)))
} else {
None
}
} else {
None
};
let item_strings: Vec<String> = items
.iter()
.map(|i| serde_json::to_string(i).unwrap_or_default())
@ -417,27 +463,61 @@ impl SmartCrusher {
};
let adaptive_k = compute_optimal_k(&item_str_refs, bias, 3, max_k);
// Tier-1 boundary: array already small enough.
// Tier-1 boundary: array already small enough — passthrough,
// nothing to compact, nothing to drop.
if items.len() <= adaptive_k {
// Python branches into _compress_text_within_items here;
// stubbed to passthrough (no within-item text compression
// in this stage).
return CrushArrayResult {
items: items.to_vec(),
strategy_info: compaction_strategy_or(&compaction_result, "none:adaptive_at_limit"),
strategy_info: "none:adaptive_at_limit".to_string(),
ccr_hash: None,
dropped_summary: String::new(),
compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
compacted: None,
compaction_kind: None,
};
}
// TOIN/feedback paths stubbed → effective_max_items = adaptive_k.
let effective_max_items = adaptive_k;
// ── Lossless-first attempt ──
//
// Run the compaction stage if present, then check the savings
// ratio against `config.lossless_min_savings_ratio`. If the
// lossless rendering shrinks the input by at least that much,
// ship it — nothing dropped, no CCR retrieval needed.
// Otherwise fall through to the lossy path.
if let Some(stage) = &self.compaction {
let (c, rendered) = stage.run(items);
if c.was_compacted() {
let input_bytes = estimate_array_bytes(&item_strings);
let savings_ratio = if input_bytes > 0 {
1.0 - (rendered.len() as f64 / input_bytes as f64)
} else {
0.0
};
if savings_ratio >= self.config.lossless_min_savings_ratio {
let kind = compaction_kind_str(&c);
return CrushArrayResult {
items: items.to_vec(), // nothing dropped
strategy_info: format!("lossless:{kind}"),
ccr_hash: None,
dropped_summary: String::new(),
compacted: Some(rendered),
compaction_kind: Some(kind),
};
}
}
}
// ── Lossy path: compress inline + cache full original via CCR ──
//
// The runtime caller (PyO3 bridge / proxy server) is expected
// to stash the full input keyed by `ccr_hash` so a retrieval
// tool can serve dropped rows back to the LLM on demand.
// **No data is lost** — "lossy" here means "compressed view
// inline; full payload retrievable via CCR cache."
let effective_max_items = adaptive_k;
let analysis = self.analyzer.analyze_array(items);
// Crushability gate: not safe to crush → return original.
// Crushability gate: not safe to crush → passthrough, no CCR.
if analysis.recommended_strategy == CompressionStrategy::Skip {
let reason = match &analysis.crushability {
Some(c) => format!("skip:{}", c.reason),
@ -445,15 +525,14 @@ impl SmartCrusher {
};
return CrushArrayResult {
items: items.to_vec(),
strategy_info: compaction_strategy_or(&compaction_result, &reason),
strategy_info: reason,
ccr_hash: None,
dropped_summary: String::new(),
compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
compacted: None,
compaction_kind: None,
};
}
// Plan + execute.
let plan = self.planner().create_plan(
&analysis,
items,
@ -464,17 +543,23 @@ impl SmartCrusher {
);
let result = self.execute_plan(&plan, items);
// CCR/telemetry/TOIN-record paths stubbed.
let strategy_info =
compaction_strategy_or(&compaction_result, analysis.recommended_strategy.as_str());
// Emit CCR-Dropped marker iff rows were actually dropped.
let dropped_count = items.len().saturating_sub(result.len());
let (ccr_hash, dropped_summary) = if dropped_count > 0 {
let h = hash_array_for_ccr(items);
let marker = format!("<<ccr:{h} {dropped_count}_rows_offloaded>>");
(Some(h), marker)
} else {
(None, String::new())
};
CrushArrayResult {
items: result,
strategy_info,
ccr_hash: None,
dropped_summary: String::new(),
compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
strategy_info: analysis.recommended_strategy.as_str().to_string(),
ccr_hash,
dropped_summary,
compacted: None,
compaction_kind: None,
}
}
@ -678,16 +763,29 @@ fn compaction_kind_str(c: &Compaction) -> &'static str {
}
}
/// If compaction won, override the lossy strategy string with
/// `"compaction:<formatter_name>:<kind>"`. Else return `default`.
fn compaction_strategy_or(
compaction_result: &Option<(String, &'static str)>,
default: &str,
) -> String {
match compaction_result {
Some((_, kind)) => format!("compaction:{kind}"),
None => default.to_string(),
}
/// Approximate byte size of `[v0, v1, ...]` JSON serialization, given
/// each item's already-serialized form. Adds 2 for outer brackets and
/// 1 per inter-item comma. Used by the lossless savings-ratio check.
fn estimate_array_bytes(item_strings: &[String]) -> usize {
let payload: usize = item_strings.iter().map(|s| s.len()).sum();
let separators = item_strings.len().saturating_sub(1);
payload + separators + 2
}
/// 12-char SHA-256 hex prefix of the canonical JSON serialization of
/// `[v0, v1, ...]`. Used as the CCR retrieval key when the lossy path
/// drops rows. Same input → same hash, so the runtime can cache the
/// original by-hash and retrieval is deterministic.
fn hash_array_for_ccr(items: &[Value]) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
let canonical = serde_json::to_string(&Value::Array(items.to_vec())).unwrap_or_default();
h.update(canonical.as_bytes());
h.finalize()
.iter()
.take(6)
.map(|b| format!("{b:02x}"))
.collect()
}
#[cfg(test)]
@ -753,8 +851,11 @@ mod tests {
#[test]
fn crush_array_skip_path_returns_original_items() {
// 30 unique dict items with ID-like fields → analyzer should
// detect "unique_entities_no_signal" and SKIP.
let c = crusher();
// detect "unique_entities_no_signal" and SKIP. Use the
// no-compaction constructor so we exercise the lossy/skip
// gate; the lossless path would otherwise short-circuit
// because uniform-tabular input is the lossless sweet spot.
let c = SmartCrusher::without_compaction(SmartCrusherConfig::default());
let items: Vec<Value> = (0..30)
.map(|i| json!({"id": i, "name": format!("user_{}", i)}))
.collect();
@ -885,8 +986,13 @@ mod tests {
#[test]
fn crush_dict_array_crushes_when_low_uniqueness() {
let c = crusher();
// 30 dicts all with status=ok → low uniqueness path → crushable.
// The public `crush()` API serializes back to JSON; the
// lossless-path output (a compacted string) is exposed via
// `crush_array().compacted` rather than being substituted into
// the JSON re-serialization. So we exercise the lossy path
// here via `without_compaction()` to validate the original
// intent: low-uniqueness dicts compress via row-dropping.
let c = SmartCrusher::without_compaction(SmartCrusherConfig::default());
let mut input = String::from("[");
for i in 0..30 {
if i > 0 {
@ -952,13 +1058,13 @@ mod tests {
assert!(result.items.len() <= 30);
}
// ---------- Stage 3c.2 PR2: lossless-first compaction wiring ----------
// ---------- Stage 3c.2 PR4: lossless-first default with threshold + CCR-Dropped ----------
#[test]
fn no_compaction_stage_yields_none_compacted_field() {
// Default SmartCrusher::new() sets compaction = None. Existing
// parity preserved: compacted/compaction_kind both None.
let c = crusher();
fn without_compaction_yields_none_compacted_field() {
// The opt-out constructor preserves pre-PR4 lossy-only path.
// No lossless attempt → compacted/compaction_kind always None.
let c = SmartCrusher::without_compaction(SmartCrusherConfig::default());
let items: Vec<Value> = (0..30).map(|_| json!({"status": "ok"})).collect();
let result = c.crush_array(&items, "", 1.0);
assert!(result.compacted.is_none());
@ -966,11 +1072,11 @@ mod tests {
}
#[test]
fn compaction_stage_runs_and_populates_compacted() {
let c = SmartCrusherBuilder::new(SmartCrusherConfig::default())
.with_default_oss_setup()
.with_default_compaction()
.build();
fn lossless_wins_when_savings_above_threshold() {
// 50 uniform tabular dicts → CSV+schema compaction shrinks
// the input dramatically (well above the 0.30 default).
// Default `SmartCrusher::new()` should pick lossless.
let c = crusher();
let items: Vec<Value> = (0..50)
.map(|i| json!({"id": i, "name": format!("u_{i}"), "status": "ok"}))
.collect();
@ -978,23 +1084,97 @@ mod tests {
let compacted = result.compacted.expect("compacted should be set");
assert!(compacted.starts_with("[50]{"), "got: {compacted}");
assert_eq!(result.compaction_kind, Some("table"));
assert!(result.strategy_info.starts_with("compaction:table"));
assert!(
result.strategy_info.starts_with("lossless:table"),
"got: {}",
result.strategy_info
);
// Lossless = nothing dropped → no CCR retrieval needed.
assert!(result.ccr_hash.is_none());
// items preserved (full original).
assert_eq!(result.items.len(), 50);
}
#[test]
fn compaction_does_not_alter_lossy_items_field() {
// The compacted bytes are emitted alongside the existing lossy
// result. items field still holds the lossy-path output so
// downstream consumers can choose either form.
let c = SmartCrusherBuilder::new(SmartCrusherConfig::default())
.with_default_oss_setup()
.with_default_compaction()
.build();
let items: Vec<Value> = (0..30).map(|_| json!({"status": "ok"})).collect();
let with_compaction = c.crush_array(&items, "", 1.0);
fn lossy_falls_through_when_savings_below_threshold() {
// Force the threshold high enough that even tabular savings
// can't satisfy it → lossy path runs → CCR-Dropped fires.
// Use low-uniqueness items so the analyzer is willing to
// crush (unique id+name per row would trip the
// "unique_entities_no_signal" skip gate instead).
let mut cfg = SmartCrusherConfig::default();
cfg.lossless_min_savings_ratio = 0.99;
let c = SmartCrusher::new(cfg);
let items: Vec<Value> = (0..50).map(|_| json!({"status": "ok"})).collect();
let result = c.crush_array(&items, "", 1.0);
// Lossless declined → no compacted output.
assert!(result.compacted.is_none());
// Lossy ran → rows dropped.
assert!(
result.items.len() < 50,
"expected lossy drop, got {} items",
result.items.len()
);
// CCR hash populated for retrieval.
let h = result.ccr_hash.expect("ccr_hash populated on drop");
assert_eq!(h.len(), 12);
// Marker visible in dropped_summary.
assert!(
result.dropped_summary.contains(&format!("<<ccr:{h}")),
"got: {}",
result.dropped_summary
);
assert!(result.dropped_summary.contains("rows_offloaded"));
}
let baseline = crusher().crush_array(&items, "", 1.0);
assert_eq!(with_compaction.items.len(), baseline.items.len());
#[test]
fn ccr_hash_is_deterministic() {
// Same input → same hash, so the runtime cache key is stable.
let mut cfg = SmartCrusherConfig::default();
cfg.lossless_min_savings_ratio = 0.99; // force lossy path
let c = SmartCrusher::new(cfg);
let items: Vec<Value> = (0..30).map(|i| json!({"id": i, "tag": "ok"})).collect();
let r1 = c.crush_array(&items, "", 1.0);
let r2 = c.crush_array(&items, "", 1.0);
assert_eq!(r1.ccr_hash, r2.ccr_hash);
assert!(r1.ccr_hash.is_some());
}
#[test]
fn ccr_hash_changes_with_input() {
let mut cfg = SmartCrusherConfig::default();
cfg.lossless_min_savings_ratio = 0.99;
let c = SmartCrusher::new(cfg);
let a: Vec<Value> = (0..30).map(|i| json!({"id": i})).collect();
let b: Vec<Value> = (100..130).map(|i| json!({"id": i})).collect();
let ra = c.crush_array(&a, "", 1.0);
let rb = c.crush_array(&b, "", 1.0);
assert_ne!(ra.ccr_hash, rb.ccr_hash);
}
#[test]
fn lossy_without_compaction_still_emits_ccr_hash() {
// The CCR-Dropped restoration applies regardless of whether
// lossless was attempted — without_compaction also gets the
// ccr_hash on row drops.
let c = SmartCrusher::without_compaction(SmartCrusherConfig::default());
let items: Vec<Value> = (0..30).map(|_| json!({"status": "ok"})).collect();
let result = c.crush_array(&items, "", 1.0);
if result.items.len() < items.len() {
assert!(result.ccr_hash.is_some());
assert!(!result.dropped_summary.is_empty());
}
}
#[test]
fn passthrough_paths_do_not_emit_ccr_hash() {
// Tier-1 boundary (items.len() <= adaptive_k): nothing
// dropped, no CCR. Skip path: same.
let c = crusher();
let small: Vec<Value> = (0..3).map(|i| json!({"id": i})).collect();
let r = c.crush_array(&small, "", 1.0);
assert!(r.ccr_hash.is_none());
assert_eq!(r.dropped_summary, "");
}
#[test]

12
crates/headroom-parity/src/lib.rs
View file

@ -377,9 +377,19 @@ impl TransformComparator for SmartCrusherComparator {
.get("relevance_threshold")
.and_then(|v| v.as_f64())
.unwrap_or(defaults.relevance_threshold),
// Rust-only PR4 knob — fixtures don't carry this; use
// default. The parity harness exercises the legacy
// lossy-only path via `without_compaction`, so this
// threshold is moot.
lossless_min_savings_ratio: config
.get("lossless_min_savings_ratio")
.and_then(|v| v.as_f64())
.unwrap_or(defaults.lossless_min_savings_ratio),
};
let crusher = SmartCrusher::new(cfg);
// Use without_compaction so the legacy fixtures (recorded
// against the pre-PR4 lossy-only path) keep matching byte-equal.
let crusher = SmartCrusher::without_compaction(cfg);
let result = crusher.crush(content, query, bias);
Ok(serde_json::json!({

17
crates/headroom-py/src/lib.rs
View file

@ -403,6 +403,7 @@ impl PySmartCrusherConfig {
first_fraction = 0.3,
last_fraction = 0.15,
relevance_threshold = 0.3,
lossless_min_savings_ratio = 0.30,
))]
#[allow(clippy::too_many_arguments)]
fn new(
@ -422,6 +423,7 @@ impl PySmartCrusherConfig {
first_fraction: f64,
last_fraction: f64,
relevance_threshold: f64,
lossless_min_savings_ratio: f64,
) -> Self {
Self {
inner: RustSmartCrusherConfig {
@ -441,6 +443,7 @@ impl PySmartCrusherConfig {
first_fraction,
last_fraction,
relevance_threshold,
lossless_min_savings_ratio,
},
}
}
@ -587,6 +590,20 @@ impl PySmartCrusher {
}
}
/// Construct WITHOUT the lossless-first compaction stage. The
/// public `crush()` API runs the lossy path directly (still with
/// CCR-Dropped retrieval markers populated when rows are dropped).
/// Used by the legacy parity fixture harness — those fixtures
/// were recorded against the pre-PR4 lossy-only behavior.
#[staticmethod]
#[pyo3(signature = (config = None))]
fn without_compaction(config: Option<&PySmartCrusherConfig>) -> Self {
let cfg = config.map(|c| c.inner.clone()).unwrap_or_default();
Self {
inner: RustSmartCrusher::without_compaction(cfg),
}
}
/// `crush(content, query="", bias=1.0) -> CrushResult`. Argument
/// order and keyword names mirror the Python implementation.
#[pyo3(signature = (content, query = "", bias = 1.0))]

57
headroom/transforms/smart_crusher.py
View file

@ -108,6 +108,7 @@ class SmartCrusher(Transform):
relevance_config: Any = None,
scorer: Any = None,
ccr_config: CCRConfig | None = None,
with_compaction: bool = True,
):
# Hard import — no Python fallback. If the wheel is missing the
# caller must build it (scripts/build_rust_extension.sh) or
@ -122,6 +123,7 @@ class SmartCrusher(Transform):
cfg = config or SmartCrusherConfig()
self.config = cfg
self._with_compaction = with_compaction
# CCR config is preserved on `self` for callers that read it
# back (`headroom.proxy.server` does), but the Rust port doesn't
@ -149,26 +151,34 @@ class SmartCrusher(Transform):
# config, plus the relevance_threshold default (0.3) — the
# Python dataclass doesn't carry that field; it lives on
# `RelevanceScorerConfig` instead.
self._rust = _RustSmartCrusher(
_RustSmartCrusherConfig(
enabled=cfg.enabled,
min_items_to_analyze=cfg.min_items_to_analyze,
min_tokens_to_crush=cfg.min_tokens_to_crush,
variance_threshold=cfg.variance_threshold,
uniqueness_threshold=cfg.uniqueness_threshold,
similarity_threshold=cfg.similarity_threshold,
max_items_after_crush=cfg.max_items_after_crush,
preserve_change_points=cfg.preserve_change_points,
factor_out_constants=cfg.factor_out_constants,
include_summaries=cfg.include_summaries,
use_feedback_hints=cfg.use_feedback_hints,
toin_confidence_threshold=cfg.toin_confidence_threshold,
dedup_identical_items=cfg.dedup_identical_items,
first_fraction=cfg.first_fraction,
last_fraction=cfg.last_fraction,
relevance_threshold=0.3,
)
rust_cfg = _RustSmartCrusherConfig(
enabled=cfg.enabled,
min_items_to_analyze=cfg.min_items_to_analyze,
min_tokens_to_crush=cfg.min_tokens_to_crush,
variance_threshold=cfg.variance_threshold,
uniqueness_threshold=cfg.uniqueness_threshold,
similarity_threshold=cfg.similarity_threshold,
max_items_after_crush=cfg.max_items_after_crush,
preserve_change_points=cfg.preserve_change_points,
factor_out_constants=cfg.factor_out_constants,
include_summaries=cfg.include_summaries,
use_feedback_hints=cfg.use_feedback_hints,
toin_confidence_threshold=cfg.toin_confidence_threshold,
dedup_identical_items=cfg.dedup_identical_items,
first_fraction=cfg.first_fraction,
last_fraction=cfg.last_fraction,
relevance_threshold=0.3,
)
# Default: lossless-first compaction (PR4). Lossless wins for
# cleanly tabular input where it saves ≥ 30% bytes; otherwise
# falls through to the lossy path with CCR-Dropped retrieval
# markers. Pass `with_compaction=False` to opt into the
# pre-PR4 lossy-only path (used by retention-property tests
# that depend on row-level item preservation).
if with_compaction:
self._rust = _RustSmartCrusher(rust_cfg)
else:
self._rust = _RustSmartCrusher.without_compaction(rust_cfg)
def crush(self, content: str, query: str = "", bias: float = 1.0) -> CrushResult:
"""Crush a single JSON content string.
@ -330,13 +340,14 @@ def smart_crush_tool_output(
content: str,
config: SmartCrusherConfig | None = None,
ccr_config: CCRConfig | None = None,
with_compaction: bool = True,
) -> tuple[str, bool, str]:
"""Compress a single tool output. Returns `(crushed, was_modified, info)`.
Convenience wrapper that builds a one-shot `SmartCrusher` per call.
`ccr_config` is accepted for source compatibility but currently
not wired through (Stage 3c.1 keeps CCR marker injection
disabled — Rust port has no compression store).
Defaults to the PR4 lossless-first behavior; pass
`with_compaction=False` to exercise the legacy lossy-only path
(still useful for retention-property tests).
"""
crusher = SmartCrusher(config=config, ccr_config=ccr_config)
crusher = SmartCrusher(config=config, ccr_config=ccr_config, with_compaction=with_compaction)
return crusher._smart_crush_content(content)

56
tests/test_quality_retention.py
View file

@ -56,7 +56,7 @@ class TestErrorRetention:
config = SmartCrusherConfig(max_items_after_crush=20)
content = json.dumps(items)
compressed_str, _, _ = smart_crush_tool_output(content, config)
compressed_str, _, _ = smart_crush_tool_output(content, config, with_compaction=False)
compressed = json.loads(compressed_str)
# Count errors before and after
@ -75,7 +75,9 @@ class TestErrorRetention:
items[50]["msg"] = f"This contains {keyword} keyword"
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
matching = [x for x in compressed if keyword in str(x).lower()]
@ -88,7 +90,9 @@ class TestErrorRetention:
items[50]["data"]["error"] = "Nested error"
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
nested_errors = [x for x in compressed if x.get("data", {}).get("error")]
@ -110,7 +114,9 @@ class TestErrorRetention:
# Compress with max 20 items
config = SmartCrusherConfig(max_items_after_crush=20)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
error_count_after = sum(1 for x in compressed if x.get("error"))
@ -148,7 +154,9 @@ class TestAnomalyRetention:
anomaly_indices.append(idx)
config = SmartCrusherConfig(max_items_after_crush=20)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
anomalies_after = sum(1 for x in compressed if x.get("is_anomaly"))
@ -165,7 +173,9 @@ class TestAnomalyRetention:
items[50]["is_anomaly"] = True
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
anomalies = [x for x in compressed if x.get("is_anomaly")]
@ -186,7 +196,7 @@ class TestRelevanceRetention:
# Use SmartCrusher with query context (via message-based API)
config = SmartCrusherConfig(max_items_after_crush=15)
crusher = SmartCrusher(config)
crusher = SmartCrusher(config, with_compaction=False)
# Create tokenizer with proper counter
model = "claude-3-5-sonnet-20241022"
@ -215,7 +225,9 @@ class TestFirstLastRetention:
items = [{"id": i, "value": i} for i in range(100)]
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
ids = [x["id"] for x in compressed]
@ -228,7 +240,9 @@ class TestFirstLastRetention:
items = [{"id": i, "value": i} for i in range(100)]
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
ids = [x["id"] for x in compressed]
@ -249,7 +263,9 @@ class TestCombinedRetention:
items[50]["is_both"] = True
config = SmartCrusherConfig(max_items_after_crush=10)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
both = [x for x in compressed if x.get("is_both")]
@ -277,7 +293,9 @@ class TestCombinedRetention:
items.append(item)
config = SmartCrusherConfig(max_items_after_crush=30)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
# Count retained critical items
@ -316,7 +334,9 @@ class TestCompressionRatio:
original_size = len(json.dumps(items))
config = SmartCrusherConfig(max_items_after_crush=50)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
compressed_size = len(json.dumps(compressed))
@ -334,7 +354,7 @@ class TestEdgeCases:
def test_empty_array(self):
"""Empty array should return empty."""
compressed_str, was_modified, _ = smart_crush_tool_output("[]")
compressed_str, was_modified, _ = smart_crush_tool_output("[]", with_compaction=False)
assert compressed_str == "[]"
assert not was_modified
@ -343,7 +363,7 @@ class TestEdgeCases:
items = [{"id": i} for i in range(3)]
original = json.dumps(items)
compressed_str, was_modified, _ = smart_crush_tool_output(original)
compressed_str, was_modified, _ = smart_crush_tool_output(original, with_compaction=False)
# Small arrays shouldn't be modified
assert json.loads(compressed_str) == items
@ -353,7 +373,9 @@ class TestEdgeCases:
items = [{"id": i, "error": f"Error {i}"} for i in range(50)]
config = SmartCrusherConfig(max_items_after_crush=20)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
# All 50 errors should be retained (errors override max_items)
@ -367,7 +389,9 @@ class TestEdgeCases:
items[50]["error"] = "错误: Unicode error message"
config = SmartCrusherConfig(max_items_after_crush=15)
compressed_str, _, _ = smart_crush_tool_output(json.dumps(items), config)
compressed_str, _, _ = smart_crush_tool_output(
json.dumps(items), config, with_compaction=False
)
compressed = json.loads(compressed_str)
errors = [x for x in compressed if x.get("error")]

87
tests/test_transforms/test_smart_crusher_lossless_default.py Normal file
View file

@ -0,0 +1,87 @@
"""Smoke tests for the PR4 lossless-first default behavior.
Verifies that `SmartCrusher()` (default constructor, no args) produces
the lossless compaction output for cleanly tabular input — the
user-visible win from PR4. Complements the legacy parity fixtures in
`test_smart_crusher_rust_parity.py` which exercise `without_compaction()`.
"""
from __future__ import annotations
import json
import pytest
def _build_extension() -> None:
try:
from headroom._core import SmartCrusher # noqa: F401
except ImportError:
pytest.skip(
"headroom._core not built — run `bash scripts/build_rust_extension.sh`",
allow_module_level=True,
)
_build_extension()
def test_default_lossless_wins_on_uniform_tabular() -> None:
"""50 uniform tabular dicts → CSV+schema beats threshold (>=30%)
by a wide margin. Default `SmartCrusher()` ships the compacted
string in place of the array."""
from headroom._core import SmartCrusher
items = [{"id": i, "level": "info", "msg": "ok"} for i in range(50)]
content = json.dumps(items)
crusher = SmartCrusher()
result = crusher.crush(content, "", 1.0)
# Output should be SHORTER than input (lossless win).
assert len(result.compressed) < len(content), (
f"compressed not smaller: {len(result.compressed)} >= {len(content)}"
)
# Strategy string surfaces the lossless tag.
assert result.strategy.startswith("lossless:table"), (
f"expected lossless:table, got {result.strategy!r}"
)
# was_modified flips because the JSON has a string where the
# array used to be.
assert result.was_modified
def test_default_falls_through_to_lossy_when_below_threshold() -> None:
"""Heterogeneous / non-tabular input → compactor declines (or
savings below threshold) → lossy path runs."""
from headroom._core import SmartCrusher
# 30 unique-id objects with no schema redundancy → not enough
# tabular signal for lossless to clear 30% by itself.
items = [{"id": i, "user_unique_field_name": f"u_{i}"} for i in range(30)]
content = json.dumps(items)
crusher = SmartCrusher()
result = crusher.crush(content, "", 1.0)
# Either lossless wins (above threshold), OR lossy ran. Both are
# acceptable; what we're guarding against is "nothing happened."
# If the strategy is `passthrough` something is wrong.
assert result.strategy != "passthrough", (
f"expected some compression, got passthrough: {result.compressed[:80]!r}"
)
def test_without_compaction_preserves_pre_pr4_behavior() -> None:
"""Opt-out constructor matches the legacy parity path."""
from headroom._core import SmartCrusher
items = [{"id": i, "level": "info", "msg": "ok"} for i in range(50)]
content = json.dumps(items)
crusher = SmartCrusher.without_compaction()
result = crusher.crush(content, "", 1.0)
# No lossless attempt → output stays JSON-array-shaped
# (pre-PR4 contract).
parsed = json.loads(result.compressed)
assert isinstance(parsed, list), (
f"expected JSON array, got {type(parsed).__name__}: {result.compressed[:80]!r}"
)

6
tests/test_transforms/test_smart_crusher_rust_parity.py
View file

@ -68,7 +68,11 @@ def test_rust_backend_matches_recorded_output(fixture_path: Path):
expected = fixture["output"]
cfg = SmartCrusherConfig(**cfg_dict)
crusher = SmartCrusher(cfg)
# Legacy fixtures were recorded against the pre-PR4 lossy-only
# path. Use `without_compaction()` to preserve byte-equal coverage
# of that path; the new lossless default has its own coverage in
# `test_smart_crusher_lossless_default.py`.
crusher = SmartCrusher.without_compaction(cfg)
actual = crusher.crush(inp["content"], inp["query"], inp["bias"])
assert actual.compressed == expected["compressed"], (