From 9e0778553fc505edb2c5bc949b7277f9ffdf3bda Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Sat, 18 Jul 2026 12:53:44 -0400 Subject: [PATCH] feat(rust): add structured prose offload plumbing (#334) (#2378) ## Description Structured payloads still leave long prose leaves without a dedicated prose compressor. The Rust pipeline already handles top-level log, diff, search, and JSON-array shapes, and the existing structured recursion rewrites stringified JSON and opaque blobs, but a plain prose string leaf inside structured content still falls back to generic opaque long-string handling instead of query-aware extractive compression. That wastes prompt budget on fields like `summary`, `description`, and `analysis` even though `headroom-core` already ships the deterministic, query-aware `TextCrusher`. This PR adds a bounded prose-field path for structured leaves. It introduces a reusable `ProseFieldOffload` backed by `TextCrusher`, then wires that offload into `JsonOffload`'s structured recursion with conservative byte and segment thresholds. Only detector-confirmed `PlainText` leaves are eligible. When a leaf clears those gates and the marker-inclusive output still saves bytes, the exact original leaf is written to CCR and the inline output carries a prose marker keyed to that store entry. Short prose, low-segment prose, diff-shaped strings, stringified JSON, and opaque base64 or HTML keep their existing behavior. This stays inside the Rust transform stack. It does not add a PyO3 shim, ONNX runtime, live-zone prose handling, or any new Python dependency. It also keeps the existing wrapper-level `JsonOffload` CCR entry, so the full structured payload remains recoverable as before. ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - Add `ProseFieldOffload` as a `ContentType::PlainText` pipeline offload backed by `TextCrusher`, with conservative byte, segment, and target-ratio thresholds. - Thread the prose offload into the structured `JsonOffload` recursion so nested prose leaves can compress and recover through the orchestrator store. - Add a pipeline-aware `JsonOffload::from_pipeline` constructor so `offload.prose_field` overrides actually reach the live prose hook instead of falling back to embedded defaults. - Preserve current behavior for short prose, low-segment prose, diff-shaped leaves, stringified JSON containers, and opaque base64 or HTML leaves. - Add focused config, routing, determinism, and CCR roundtrip coverage for the new prose path. - Leave changelog generation to the repo's conventional-commit release flow rather than editing `CHANGELOG.md` directly. ## Testing - [x] Unit tests pass (`cargo test -p headroom-core --lib transforms::pipeline::offloads::prose_field::tests`) - [x] Linting passes (`cargo clippy -p headroom-core -- -D warnings`) - [ ] Type checking passes (`uv run mypy headroom`) - [x] New tests added for new functionality when applicable - [ ] Manual testing performed ### Test Output ```text cargo fmt --all -- --check cargo clippy -p headroom-core -- -D warnings cargo test -p headroom-core --lib transforms::pipeline::offloads::prose_field::tests test result: ok. 6 passed; 0 failed cargo test -p headroom-core --lib transforms::pipeline::offloads::json_offload::tests test result: ok. 17 passed; 0 failed cargo test -p headroom-core --lib transforms::smart_crusher::crusher::tests::default_crush_ignores_opt_in_prose_hook -- --exact test result: ok. 1 passed; 0 failed cargo test -p headroom-core --lib transforms::smart_crusher::crusher::tests::prose_hook_preserves_html_opaque_routing -- --exact test result: ok. 1 passed; 0 failed cargo test -p headroom-core --lib transforms::smart_crusher::crusher::tests::prose_hook_runs_for_dict_array_rows -- --exact test result: ok. 1 passed; 0 failed cargo test -p headroom-core --lib transforms::smart_crusher::crusher::tests::unchanged_stringified_json_container_skips_prose_hook -- --exact test result: ok. 1 passed; 0 failed cargo test -p headroom-core --test ccr_roundtrip nested_structured_prose_leaf_uses_ccr -- --exact test result: ok. 1 passed; 0 failed git diff --check ``` ## Real Behavior Proof - Environment: Windows 11, stable Rust toolchain, in-memory CCR store, no live provider - Exact command / steps: run the focused nested CCR roundtrip test through `CompressionPipeline::run` on a five-row structured payload containing a long prose leaf, then resolve the emitted prose marker key from the same orchestrator store - Observed result: the generic `CompressionPipeline` plus `JsonOffload` path applies, the nested prose leaf becomes shorter on the wire, and that prose key retrieves the byte-identical original leaf from the orchestrator store while HTML-shaped and diff-shaped leaves stay on their opaque marker routes - Not tested: live provider run ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [ ] I have commented my code, particularly in hard-to-understand areas - [x] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - The upstream issue body originally parked PR3b behind a PyO3 shim or a later ONNX port. This PR takes the narrower Rust-native path instead by reusing the existing `TextCrusher` already in `headroom-core`. - This PR advances the pipeline-side PR3b slice from #334. It does not close #334, and it does not wire live-zone or PyO3 SmartCrusher callers to this path. - `CHANGELOG.md` is intentionally untouched because this repo's release pipeline generates changelog entries from conventional commits, and `repos/headroom/config.md` marks manual changelog edits as out of policy. - Python lint, type checking, and pytest are not part of the focused local proof for this slice because the change stays inside `crates/headroom-core`. --- crates/headroom-core/config/pipeline.toml | 10 + crates/headroom-core/src/transforms/mod.rs | 3 +- .../src/transforms/pipeline/config.rs | 69 ++++++ .../src/transforms/pipeline/mod.rs | 9 +- .../pipeline/offloads/json_offload.rs | 131 ++++++++++- .../src/transforms/pipeline/offloads/mod.rs | 2 + .../pipeline/offloads/prose_field.rs | 170 ++++++++++++++ .../src/transforms/pipeline/orchestrator.rs | 4 +- .../src/transforms/smart_crusher/crusher.rs | 209 +++++++++++++++++- crates/headroom-core/tests/ccr_roundtrip.rs | 40 ++++ 10 files changed, 626 insertions(+), 21 deletions(-) create mode 100644 crates/headroom-core/src/transforms/pipeline/offloads/prose_field.rs diff --git a/crates/headroom-core/config/pipeline.toml b/crates/headroom-core/config/pipeline.toml index 811123778..687ef7309 100644 --- a/crates/headroom-core/config/pipeline.toml +++ b/crates/headroom-core/config/pipeline.toml @@ -149,3 +149,13 @@ lockfile_suffixes = [ # show up in formatter / linter commits and carry no signal the LLM # needs to reason about. drop_whitespace_only_hunks = true + +# ─── Structured prose-field offload config ───────────────────────── +# +# Only detector-confirmed PlainText leaves above both floors are candidates. +# The final marker-inclusive output must still be shorter than the leaf. + +[offload.prose_field] +min_bytes = 256 +min_segments = 6 +target_ratio = 0.5 diff --git a/crates/headroom-core/src/transforms/mod.rs b/crates/headroom-core/src/transforms/mod.rs index d65c1deca..9bb11c3f9 100644 --- a/crates/headroom-core/src/transforms/mod.rs +++ b/crates/headroom-core/src/transforms/mod.rs @@ -55,7 +55,8 @@ pub use magika_detector::{magika_detect, map_magika_label, MagikaDetectorError}; pub use pipeline::{ CompressionContext, CompressionPipeline, CompressionPipelineBuilder, DiffNoise, DiffOffload, JsonMinifier, JsonOffload, LogOffload, LogTemplate, OffloadOutput, OffloadTransform, - PipelineConfig, PipelineResult, ReformatOutput, ReformatTransform, TransformError, + PipelineConfig, PipelineResult, ProseFieldOffload, ReformatOutput, ReformatTransform, + TransformError, }; pub use recommendations::{Recommendation, RecommendationStore, RECOMMENDATIONS_PATH_ENV_VAR}; pub use safety::{tool_pair_indices, ToolPair}; diff --git a/crates/headroom-core/src/transforms/pipeline/config.rs b/crates/headroom-core/src/transforms/pipeline/config.rs index a44a56a7d..dc7fc4ed3 100644 --- a/crates/headroom-core/src/transforms/pipeline/config.rs +++ b/crates/headroom-core/src/transforms/pipeline/config.rs @@ -184,6 +184,7 @@ pub struct LogTemplateConfig { pub struct OffloadConfigs { pub json: JsonOffloadConfig, pub diff_noise: DiffNoiseConfig, + pub prose_field: ProseFieldConfig, } /// Knobs for the [`crate::transforms::pipeline::offloads::JsonOffload`] @@ -196,6 +197,15 @@ pub struct JsonOffloadConfig { pub saturation_rows: usize, } +/// Knobs for the [`crate::transforms::pipeline::offloads::ProseFieldOffload`] +/// structured string-leaf compressor. +#[derive(Debug, Clone, Copy, Deserialize, PartialEq)] +pub struct ProseFieldConfig { + pub min_bytes: usize, + pub min_segments: usize, + pub target_ratio: f64, +} + /// Knobs for the [`crate::transforms::pipeline::offloads::DiffNoise`] /// offload. Lockfile suffixes are matched against the new-file path /// at the end of each `diff --git` header. @@ -283,6 +293,11 @@ mod tests { min_lines = 20 lockfile_suffixes = ["custom.lock"] drop_whitespace_only_hunks = false + + [offload.prose_field] + min_bytes = 300 + min_segments = 5 + target_ratio = 0.4 "#; let cfg = PipelineConfig::from_toml_str(toml).expect("override parses"); assert_eq!(cfg.pipeline.reformat_target_ratio, 0.3); @@ -293,6 +308,9 @@ mod tests { cfg.offload.diff_noise.lockfile_suffixes, vec!["custom.lock"] ); + assert_eq!(cfg.offload.prose_field.min_bytes, 300); + assert_eq!(cfg.offload.prose_field.min_segments, 5); + assert_eq!(cfg.offload.prose_field.target_ratio, 0.4); } #[test] @@ -302,6 +320,9 @@ mod tests { assert_eq!(cfg.reformat.log_template.min_run, 3); assert_eq!(cfg.offload.json.min_array_rows, 5); assert_eq!(cfg.offload.json.saturation_rows, 50); + assert_eq!(cfg.offload.prose_field.min_bytes, 256); + assert_eq!(cfg.offload.prose_field.min_segments, 6); + assert_eq!(cfg.offload.prose_field.target_ratio, 0.5); assert!(!cfg.offload.diff_noise.lockfile_suffixes.is_empty()); assert!(cfg .offload @@ -311,6 +332,54 @@ mod tests { .any(|s| s == "Cargo.lock")); } + #[test] + fn prose_field_defaults_and_override() { + let defaults = PipelineConfig::default().offload.prose_field; + let override_cfg = PipelineConfig::from_toml_str( + r#" + [pipeline] + reformat_target_ratio = 0.5 + bloat_threshold = 0.5 + offload_fallback_ratio = 0.85 + [bloat.log] + min_lines = 50 + sample_size = 100 + high_priority_threshold = 0.4 + uniqueness_weight = 0.5 + priority_dilution_weight = 0.5 + [bloat.diff] + min_lines = 50 + normal_context_ratio = 0.6 + [bloat.search] + min_matches = 10 + cluster_threshold = 10.0 + [reformat.log_template] + min_lines = 20 + min_run = 3 + similarity_threshold = 0.8 + min_constant_tokens = 2 + [offload.json] + min_array_rows = 5 + saturation_rows = 50 + [offload.diff_noise] + min_lines = 30 + lockfile_suffixes = ["Cargo.lock"] + drop_whitespace_only_hunks = true + [offload.prose_field] + min_bytes = 300 + min_segments = 5 + target_ratio = 0.4 + "#, + ) + .unwrap() + .offload + .prose_field; + assert_eq!(defaults.min_bytes, 256); + assert_eq!(override_cfg.min_bytes, 300); + assert_eq!(override_cfg.min_segments, 5); + assert_eq!(override_cfg.target_ratio, 0.4); + } + #[test] fn malformed_toml_returns_error() { let r = PipelineConfig::from_toml_str("this is not toml = [unterminated"); diff --git a/crates/headroom-core/src/transforms/pipeline/mod.rs b/crates/headroom-core/src/transforms/pipeline/mod.rs index 54fe4a667..c6bde6073 100644 --- a/crates/headroom-core/src/transforms/pipeline/mod.rs +++ b/crates/headroom-core/src/transforms/pipeline/mod.rs @@ -73,9 +73,6 @@ //! but is NOT in the default re-exports — modern agents use scoped //! `rg`/`grep`, the marginal value didn't justify default registration. //! -//! Deferred to later PRs: -//! - **ProseFieldCompressor** — Phase 3g PR3. Compresses prose-shaped -//! string fields inside structured payloads. //! //! [`estimate_bloat`]: traits::OffloadTransform::estimate_bloat @@ -87,14 +84,14 @@ pub mod traits; pub use config::{ BloatConfigs, ConfigError, DiffBloatConfig, DiffNoiseConfig, JsonOffloadConfig, LogBloatConfig, - LogTemplateConfig, OffloadConfigs, OrchestratorConfig, PipelineConfig, ReformatConfigs, - SearchBloatConfig, + LogTemplateConfig, OffloadConfigs, OrchestratorConfig, PipelineConfig, ProseFieldConfig, + ReformatConfigs, SearchBloatConfig, }; // `SearchOffload` is intentionally NOT in the top-level re-export // (deprecated from default pipeline; reach via the explicit module // path if you want to opt in). See `offloads::search_offload` head // docs for rationale. -pub use offloads::{DiffNoise, DiffOffload, JsonOffload, LogOffload}; +pub use offloads::{DiffNoise, DiffOffload, JsonOffload, LogOffload, ProseFieldOffload}; pub use orchestrator::{CompressionPipeline, CompressionPipelineBuilder, PipelineResult}; pub use reformats::{JsonMinifier, LogTemplate}; pub use traits::{ diff --git a/crates/headroom-core/src/transforms/pipeline/offloads/json_offload.rs b/crates/headroom-core/src/transforms/pipeline/offloads/json_offload.rs index eae284835..54bb62ded 100644 --- a/crates/headroom-core/src/transforms/pipeline/offloads/json_offload.rs +++ b/crates/headroom-core/src/transforms/pipeline/offloads/json_offload.rs @@ -48,7 +48,8 @@ use md5::{Digest, Md5}; use crate::ccr::CcrStore; -use crate::transforms::pipeline::config::JsonOffloadConfig; +use crate::transforms::pipeline::config::{JsonOffloadConfig, PipelineConfig, ProseFieldConfig}; +use crate::transforms::pipeline::offloads::prose_field::ProseFieldOffload; use crate::transforms::pipeline::traits::{ CompressionContext, OffloadOutput, OffloadTransform, TransformError, }; @@ -63,6 +64,7 @@ const CONFIDENCE: f32 = 0.85; pub struct JsonOffload { crusher: SmartCrusher, config: JsonOffloadConfig, + prose: ProseFieldOffload, } impl JsonOffload { @@ -75,13 +77,38 @@ impl JsonOffload { Self { crusher: SmartCrusher::new(SmartCrusherConfig::default()), config, + prose: ProseFieldOffload::new(PipelineConfig::default().offload.prose_field), + } + } + + pub fn from_pipeline(config: &PipelineConfig) -> Self { + Self { + crusher: SmartCrusher::new(SmartCrusherConfig::default()), + config: config.offload.json, + prose: ProseFieldOffload::new(config.offload.prose_field), } } /// Custom constructor — used by tests that want a stubbed crusher /// or a custom SmartCrusher config. pub fn with_crusher(crusher: SmartCrusher, config: JsonOffloadConfig) -> Self { - Self { crusher, config } + Self { + crusher, + config, + prose: ProseFieldOffload::new(PipelineConfig::default().offload.prose_field), + } + } + + pub fn with_crusher_and_prose( + crusher: SmartCrusher, + config: JsonOffloadConfig, + prose_config: ProseFieldConfig, + ) -> Self { + Self { + crusher, + config, + prose: ProseFieldOffload::new(prose_config), + } } } @@ -121,7 +148,17 @@ impl OffloadTransform for JsonOffload { ctx: &CompressionContext, store: &dyn CcrStore, ) -> Result { - let result = self.crusher.crush(content, &ctx.query, 0.0); + let prose = &self.prose; + let prose_hook = |leaf: &str, query: &str| { + let leaf_ctx = CompressionContext::with_query(query); + prose + .apply(leaf, &leaf_ctx, store) + .ok() + .map(|output| (output.output, output.cache_key)) + }; + let result = self + .crusher + .crush_with_prose_hook(content, &ctx.query, 0.0, &prose_hook); if !result.was_modified { return Err(TransformError::skipped( NAME, @@ -198,6 +235,94 @@ mod tests { JsonOffload::new(cfg()) } + #[test] + fn from_pipeline_uses_overrideable_prose_config() { + let config = PipelineConfig::from_toml_str( + r#" + [pipeline] + reformat_target_ratio = 0.5 + bloat_threshold = 0.5 + offload_fallback_ratio = 0.85 + + [bloat.log] + min_lines = 50 + sample_size = 100 + high_priority_threshold = 0.4 + uniqueness_weight = 0.5 + priority_dilution_weight = 0.5 + + [bloat.diff] + min_lines = 50 + normal_context_ratio = 0.6 + + [bloat.search] + min_matches = 10 + cluster_threshold = 10.0 + + [reformat.log_template] + min_lines = 20 + min_run = 3 + similarity_threshold = 0.4 + min_constant_tokens = 2 + + [offload.json] + min_array_rows = 5 + saturation_rows = 50 + + [offload.prose_field] + min_bytes = 300 + min_segments = 5 + target_ratio = 0.4 + + [offload.diff_noise] + min_lines = 30 + lockfile_suffixes = ["Cargo.lock"] + drop_whitespace_only_hunks = true + "#, + ) + .expect("override parses"); + + let offload = JsonOffload::from_pipeline(&config); + assert_eq!(offload.prose.config(), config.offload.prose_field); + } + + #[test] + fn prose_hook_ignores_diff_shaped_leaf() { + let diff = format!( + "diff --git a/foo.py b/foo.py\n--- a/foo.py\n+++ b/foo.py\n@@ -1,20 +1,20 @@\n{}", + (0..20) + .map(|i| format!("-old line {i}\n+new line {i}")) + .collect::>() + .join("\n") + ); + let input = serde_json::json!([{"description": diff}]).to_string(); + let store = InMemoryCcrStore::new(); + let output = offload() + .apply(&input, &CompressionContext::default(), &store) + .expect("diff-shaped leaf should still process through the direct offload path"); + let marker = output + .output + .split("<>").next()) + .expect("direct offload should still emit a leaf marker"); + assert!( + marker.contains(",string,"), + "diff leaf should stay on the opaque string route, got {marker}" + ); + } + + #[test] + fn prose_hook_preserves_opaque_html_leaf() { + let html = "

".to_string() + &"x".repeat(300) + "

"; + let input = serde_json::json!([{"summary": html}]).to_string(); + let store = InMemoryCcrStore::new(); + let result = offload() + .apply(&input, &CompressionContext::with_query("recovery"), &store) + .expect("structured html should still process"); + assert!(result.output.contains(",html,")); + } + /// Build a JSON array of N similar dicts with id + name + value. /// Compact JSON (no extra whitespace) so byte counts are predictable. fn build_tabular_array(n: usize) -> String { diff --git a/crates/headroom-core/src/transforms/pipeline/offloads/mod.rs b/crates/headroom-core/src/transforms/pipeline/offloads/mod.rs index cf01bb1cd..05fe24be1 100644 --- a/crates/headroom-core/src/transforms/pipeline/offloads/mod.rs +++ b/crates/headroom-core/src/transforms/pipeline/offloads/mod.rs @@ -21,12 +21,14 @@ pub mod diff_noise; pub mod diff_offload; pub mod json_offload; pub mod log_offload; +pub mod prose_field; pub mod search_offload; pub use diff_noise::DiffNoise; pub use diff_offload::DiffOffload; pub use json_offload::JsonOffload; pub use log_offload::LogOffload; +pub use prose_field::ProseFieldOffload; // `SearchOffload` is intentionally NOT re-exported here. The // orchestrator-default registration omits it; keep the type accessible // via the explicit module path for opt-in callers, but discourage new diff --git a/crates/headroom-core/src/transforms/pipeline/offloads/prose_field.rs b/crates/headroom-core/src/transforms/pipeline/offloads/prose_field.rs new file mode 100644 index 000000000..6494f8cd1 --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/offloads/prose_field.rs @@ -0,0 +1,170 @@ +//! CCR-backed extractive compression for prose leaves in structured payloads. + +use crate::ccr::{compute_key, marker_for, CcrStore}; +use crate::transforms::content_detector::detect_content_type; +use crate::transforms::pipeline::config::ProseFieldConfig; +use crate::transforms::pipeline::traits::{ + CompressionContext, OffloadOutput, OffloadTransform, TransformError, +}; +use crate::transforms::text_crusher::TextCrusher; +use crate::transforms::ContentType; + +const NAME: &str = "prose_field_offload"; +const CONFIDENCE: f32 = 0.8; + +pub struct ProseFieldOffload { + crusher: TextCrusher, + config: ProseFieldConfig, +} + +impl ProseFieldOffload { + pub fn new(config: ProseFieldConfig) -> Self { + Self { + crusher: TextCrusher::default(), + config, + } + } + + pub fn config(&self) -> ProseFieldConfig { + self.config + } + + fn eligible(&self, content: &str) -> bool { + content.len() >= self.config.min_bytes + && detect_content_type(content).content_type == ContentType::PlainText + } + + fn compress(&self, content: &str, query: &str) -> Option<(String, String)> { + if !self.eligible(content) { + return None; + } + let result = self + .crusher + .compress(content, query, Some(self.config.target_ratio)); + if result.total_segments < self.config.min_segments || result.compressed == content { + return None; + } + + let key = compute_key(content.as_bytes()); + let output = format!("{}\n{}", result.compressed, marker_for(&key)); + (output.len() < content.len()).then_some((output, key)) + } +} + +impl OffloadTransform for ProseFieldOffload { + fn name(&self) -> &'static str { + NAME + } + + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + + fn estimate_bloat(&self, content: &str) -> f32 { + if !self.eligible(content) { + return 0.0; + } + let segments = content + .split(['.', '!', '?', '\n']) + .filter(|segment| !segment.trim().is_empty()) + .count(); + if segments < self.config.min_segments { + 0.0 + } else { + 1.0 + } + } + + fn apply( + &self, + content: &str, + ctx: &CompressionContext, + store: &dyn CcrStore, + ) -> Result { + let Some((output, key)) = self.compress(content, &ctx.query) else { + return Err(TransformError::skipped( + NAME, + "prose compression not worth it", + )); + }; + store.put(&key, content); + Ok(OffloadOutput::from_lengths(content.len(), output, key)) + } + + fn confidence(&self) -> f32 { + CONFIDENCE + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ccr::InMemoryCcrStore; + use crate::transforms::pipeline::config::PipelineConfig; + + fn offload() -> ProseFieldOffload { + ProseFieldOffload::new(PipelineConfig::default().offload.prose_field) + } + + fn prose() -> String { + (0..12) + .map(|i| { + if i % 3 == 0 { + format!("Segment {i} documents recovery safeguards for this field.") + } else { + format!("Segment {i} explains general context without the key term present.") + } + }) + .collect::>() + .join(" ") + } + + #[test] + fn short_plain_text_is_byte_identical() { + let input = "A short note."; + let store = InMemoryCcrStore::new(); + let result = offload().apply(input, &CompressionContext::default(), &store); + assert!(result.is_err()); + assert!(store.is_empty()); + } + + #[test] + fn long_low_segment_text_is_byte_identical() { + let input = "A".repeat(300); + let store = InMemoryCcrStore::new(); + let result = offload().apply(&input, &CompressionContext::default(), &store); + assert!(result.is_err()); + assert!(store.is_empty()); + } + + #[test] + fn estimate_bloat_is_zero_for_short_or_non_plain_text() { + assert_eq!(offload().estimate_bloat("A short note."), 0.0); + let html = "

".to_string() + &"x".repeat(300) + "

"; + assert_eq!(offload().estimate_bloat(&html), 0.0); + } + + #[test] + fn estimate_bloat_is_one_for_eligible_plain_text() { + assert_eq!(offload().estimate_bloat(&prose()), 1.0); + } + + #[test] + fn estimate_bloat_is_zero_for_long_low_segment_plain_text() { + let input = "A".repeat(300); + assert_eq!(offload().estimate_bloat(&input), 0.0); + } + + #[test] + fn query_changes_selection_deterministically() { + let input = prose(); + let crusher = offload(); + let a = crusher.compress(&input, "recovery").unwrap(); + let b = crusher.compress(&input, "recovery").unwrap(); + assert_eq!(a, b); + assert!(a.0.contains("recovery")); + assert!(!a + .0 + .contains("Segment 1 explains general context without the key term present.")); + } +} diff --git a/crates/headroom-core/src/transforms/pipeline/orchestrator.rs b/crates/headroom-core/src/transforms/pipeline/orchestrator.rs index 69df00f4b..66592e09f 100644 --- a/crates/headroom-core/src/transforms/pipeline/orchestrator.rs +++ b/crates/headroom-core/src/transforms/pipeline/orchestrator.rs @@ -776,7 +776,7 @@ mod tests { let cfg = PipelineConfig::default(); let p = CompressionPipeline::builder() .with_reformat(JsonMinifier) - .with_offload(JsonOffload::new(cfg.offload.json)) + .with_offload(JsonOffload::from_pipeline(&cfg)) .with_config(cfg) .build(); let s = store(); @@ -809,7 +809,7 @@ mod tests { fn end_to_end_json_offload_skipped_for_small_array() { let cfg = PipelineConfig::default(); let p = CompressionPipeline::builder() - .with_offload(JsonOffload::new(cfg.offload.json)) + .with_offload(JsonOffload::from_pipeline(&cfg)) .with_config(cfg) .build(); let s = store(); diff --git a/crates/headroom-core/src/transforms/smart_crusher/crusher.rs b/crates/headroom-core/src/transforms/smart_crusher/crusher.rs index 0c45f163a..24eb97337 100644 --- a/crates/headroom-core/src/transforms/smart_crusher/crusher.rs +++ b/crates/headroom-core/src/transforms/smart_crusher/crusher.rs @@ -54,6 +54,8 @@ use crate::relevance::RelevanceScorer; use crate::transforms::adaptive_sizer::compute_optimal_k; use crate::transforms::anchor_selector::AnchorSelector; +type ProseHook<'a> = dyn Fn(&str, &str) -> Option<(String, String)> + 'a; + /// Return type for `crush_array`. /// /// Two operating paths feed the same result type: @@ -138,6 +140,43 @@ pub struct SmartCrusher { } impl SmartCrusher { + /// Opt-in variant used by structured pipeline owners that want to + /// transform plain string leaves while preserving default callers. + pub fn crush_with_prose_hook( + &self, + content: &str, + query: &str, + bias: f64, + hook: &ProseHook<'_>, + ) -> CrushResult { + let start = std::time::Instant::now(); + let (compressed, was_modified, info) = + self.smart_crush_content_with_hook(content, query, bias, Some(hook)); + let strategy = if info.is_empty() { + "passthrough".to_string() + } else { + info + }; + if !self.observers.is_empty() { + let event = CrushEvent { + strategy: strategy.clone(), + input_bytes: content.len(), + output_bytes: compressed.len(), + elapsed_ns: start.elapsed().as_nanos() as u64, + was_modified, + }; + for observer in &self.observers { + observer.on_event(&event); + } + } + CrushResult { + compressed, + original: content.to_string(), + was_modified, + strategy, + } + } + /// Construct with the OSS default composition: scorer + constraints + /// observer + **lossless-first compaction stage**. Calling /// `crush_array` runs the dispatch: @@ -390,12 +429,22 @@ impl SmartCrusher { query_context: &str, bias: f64, ) -> (String, bool, String) { - // Parse — non-JSON content passes through unchanged. + self.smart_crush_content_with_hook(content, query_context, bias, None) + } + + fn smart_crush_content_with_hook( + &self, + content: &str, + query_context: &str, + bias: f64, + prose_hook: Option<&ProseHook<'_>>, + ) -> (String, bool, String) { let Ok(parsed) = serde_json::from_str::(content) else { return (content.to_string(), false, String::new()); }; - let (crushed, info) = self.process_value(&parsed, 0, query_context, bias); + let (crushed, info) = + self.process_value_with_hook(&parsed, 0, query_context, bias, prose_hook); // Re-serialize with Python `safe_json_dumps` formatting: // compact `(",", ":")` separators + `ensure_ascii=False`, @@ -422,6 +471,17 @@ impl SmartCrusher { depth: usize, query_context: &str, bias: f64, + ) -> (Value, String) { + self.process_value_with_hook(value, depth, query_context, bias, None) + } + + fn process_value_with_hook( + &self, + value: &Value, + depth: usize, + query_context: &str, + bias: f64, + prose_hook: Option<&ProseHook<'_>>, ) -> (Value, String) { if depth >= Self::MAX_PROCESS_DEPTH { return (value.clone(), String::new()); @@ -436,7 +496,34 @@ impl SmartCrusher { let arr_type = classify_array(arr); match arr_type { ArrayType::DictArray => { - let result = self.crush_array(arr, query_context, bias); + let mut rows: Vec = Vec::with_capacity(n); + if let Some(hook) = prose_hook { + for item in arr { + if let Value::Object(map) = item { + let mut processed = serde_json::Map::new(); + for (k, v) in map { + let (p_val, p_info) = self.process_value_with_hook( + v, + depth + 1, + query_context, + bias, + Some(hook), + ); + processed.insert(k.clone(), p_val); + if !p_info.is_empty() { + info_parts.push(p_info); + } + } + rows.push(Value::Object(processed)); + } else { + rows.push(item.clone()); + } + } + } else { + rows.extend(arr.iter().cloned()); + } + + let result = self.crush_array(&rows, query_context, bias); // Lossless path won → substitute the array // with the compacted string in place. This // makes the lossless win visible to the @@ -512,7 +599,13 @@ impl SmartCrusher { // Below threshold or not crushable → recurse into items. let mut processed: Vec = Vec::with_capacity(n); for item in arr { - let (p_item, p_info) = self.process_value(item, depth + 1, query_context, bias); + let (p_item, p_info) = self.process_value_with_hook( + item, + depth + 1, + query_context, + bias, + prose_hook, + ); processed.push(p_item); if !p_info.is_empty() { info_parts.push(p_info); @@ -524,7 +617,8 @@ impl SmartCrusher { // First pass: recurse into values to compress nested arrays. let mut processed = serde_json::Map::new(); for (k, v) in map { - let (p_val, p_info) = self.process_value(v, depth + 1, query_context, bias); + let (p_val, p_info) = + self.process_value_with_hook(v, depth + 1, query_context, bias, prose_hook); processed.insert(k.clone(), p_val); if !p_info.is_empty() { info_parts.push(p_info); @@ -547,7 +641,9 @@ impl SmartCrusher { // `process_string` which parses stringified-JSON containers // (recursing through `process_value`) and CCR-substitutes // opaque blobs (with store-write so retrieval works). - Value::String(s) => self.process_string(s, depth, query_context, bias), + Value::String(s) => { + self.process_string_with_hook(s, depth, query_context, bias, prose_hook) + } // Other scalars — passthrough. _ => (value.clone(), String::new()), } @@ -569,16 +665,19 @@ impl SmartCrusher { /// format as `compaction::walker::format_ccr_marker` so /// downstream consumers can pattern-match markers regardless /// of which path emitted them. - fn process_string( + fn process_string_with_hook( &self, s: &str, depth: usize, query_context: &str, bias: f64, + prose_hook: Option<&ProseHook<'_>>, ) -> (Value, String) { + let mut parsed_container_unchanged = false; // 1. Stringified-JSON: parse, recurse, re-render. if let Some(parsed) = try_parse_json_container(s) { - let (processed, sub_info) = self.process_value(&parsed, depth + 1, query_context, bias); + let (processed, sub_info) = + self.process_value_with_hook(&parsed, depth + 1, query_context, bias, prose_hook); // If recursion produced something different, re-emit. // Special case: if the recursion returned a `Value::String` // (lossless compaction substituted the array with a @@ -598,6 +697,7 @@ impl SmartCrusher { }; return (Value::String(rendered), info); } + parsed_container_unchanged = true; } // 2. Opaque blob: substitute with CCR marker AND stash the @@ -609,13 +709,33 @@ impl SmartCrusher { emit_opaque_markers: self.config.opaque_markers_enabled(), ..ClassifyConfig::default() }; + if let CellClass::Opaque(kind) = classify_cell(&Value::String(s.to_string()), &cfg) { + match kind { + super::compaction::OpaqueKind::Base64Blob + | super::compaction::OpaqueKind::HtmlChunk => { + let marker = emit_opaque_ccr_marker(s, &kind, self.ccr_store.as_ref()); + let kind_label = opaque_kind_label(&kind); + return (Value::String(marker), format!("string_ccr:{kind_label}")); + } + super::compaction::OpaqueKind::LongString + | super::compaction::OpaqueKind::Other(_) => {} + } + } + if !parsed_container_unchanged { + if let Some(hook) = prose_hook { + if let Some((compressed, key)) = hook(s, query_context) { + return (Value::String(compressed), format!("string_prose:{key}")); + } + } + } + if let CellClass::Opaque(kind) = classify_cell(&Value::String(s.to_string()), &cfg) { let marker = emit_opaque_ccr_marker(s, &kind, self.ccr_store.as_ref()); let kind_label = opaque_kind_label(&kind); return (Value::String(marker), format!("string_ccr:{kind_label}")); } - // 3. Plain string — passthrough. + // 4. Plain string — passthrough. (Value::String(s.to_string()), String::new()) } @@ -1074,6 +1194,20 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn default_crush_ignores_opt_in_prose_hook() { + let crusher = SmartCrusher::new(SmartCrusherConfig::default()); + let input = serde_json::json!({ + "summary": (0..8) + .map(|i| format!("Segment {i} keeps the default route stable.")) + .collect::>() + .join(" ") + }) + .to_string(); + let result = crusher.crush(&input, "default route", 0.0); + assert!(!result.strategy.contains("string_prose:")); + } + fn crusher() -> SmartCrusher { SmartCrusher::new(SmartCrusherConfig::default()) } @@ -1601,6 +1735,63 @@ mod tests { assert!(blob.contains(",base64,")); } + #[test] + fn prose_hook_preserves_html_opaque_routing() { + let c = SmartCrusher::new(SmartCrusherConfig::default()); + let html = "

".to_string() + &"x".repeat(300) + "

"; + let hook = |_leaf: &str, _query: &str| -> Option<(String, String)> { + Some(("compressed prose".to_string(), "prose-key".to_string())) + }; + let (out, info) = c.process_string_with_hook(&html, 0, "recovery", 1.0, Some(&hook)); + let routed = out.as_str().expect("html stays string"); + assert!(routed.contains(",html,")); + assert_eq!(info, "string_ccr:html"); + } + + #[test] + fn prose_hook_runs_for_dict_array_rows() { + let c = SmartCrusher::new(SmartCrusherConfig::default()); + let prose = (0..12) + .map(|i| format!("Segment {i} documents recovery safeguards for this field.")) + .collect::>() + .join(" "); + let rows = (0..5) + .map(|i| serde_json::json!({"id": i, "summary": prose})) + .collect::>(); + let input = Value::Array(rows); + let hook = |leaf: &str, _query: &str| -> Option<(String, String)> { + if leaf.contains("recovery safeguards") { + Some(("<>".to_string(), "prose-key".to_string())) + } else { + None + } + }; + + let (out, info) = c.process_value_with_hook(&input, 0, "recovery", 1.0, Some(&hook)); + let rendered = out.to_string(); + assert!(rendered.contains("<>")); + assert!(info.contains("string_prose:prose-key")); + } + + #[test] + fn unchanged_stringified_json_container_skips_prose_hook() { + let c = SmartCrusher::new(SmartCrusherConfig::default()); + let payload = serde_json::json!({ + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa": "short", + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb": "still short", + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc": "tiny", + "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd": "small" + }) + .to_string(); + let hook = |leaf: &str, _query: &str| -> Option<(String, String)> { + (leaf.len() > 256).then_some(("<>".to_string(), "prose-key".to_string())) + }; + let (out, info) = c.process_string_with_hook(&payload, 0, "recovery", 1.0, Some(&hook)); + let routed = out.as_str().expect("stringified JSON stays string"); + assert_eq!(routed, payload); + assert!(info.is_empty()); + } + #[test] fn process_string_top_level_string_processed() { // crush() takes a string; if it doesn't parse as JSON, today's diff --git a/crates/headroom-core/tests/ccr_roundtrip.rs b/crates/headroom-core/tests/ccr_roundtrip.rs index 28d35dd8a..7cbc21eef 100644 --- a/crates/headroom-core/tests/ccr_roundtrip.rs +++ b/crates/headroom-core/tests/ccr_roundtrip.rs @@ -361,6 +361,46 @@ fn document_walker_with_store_roundtrips_opaque_blob() { assert_eq!(store.get(&h).unwrap(), big); } +#[test] +fn nested_structured_prose_leaf_uses_ccr() { + use headroom_core::transforms::pipeline::config::PipelineConfig; + use headroom_core::transforms::pipeline::offloads::JsonOffload; + use headroom_core::transforms::pipeline::orchestrator::CompressionPipeline; + use headroom_core::transforms::pipeline::traits::CompressionContext; + use headroom_core::transforms::ContentType; + + let prose = (0..12) + .map(|i| format!("Segment {i} explains the durable recovery behavior for this field.")) + .collect::>() + .join(" "); + let input = serde_json::json!((0..5) + .map(|i| serde_json::json!({"id": i, "summary": prose})) + .collect::>()) + .to_string(); + let store = InMemoryCcrStore::new(); + let config = PipelineConfig::default(); + let pipeline = CompressionPipeline::builder() + .with_offload(JsonOffload::from_pipeline(&config)) + .with_config(config) + .build(); + let result = pipeline.run( + &input, + ContentType::JsonArray, + &CompressionContext::with_query("recovery"), + &store, + ); + assert!(result.bytes_saved > 0); + assert!(result + .steps_applied + .iter() + .any(|step| step == "json_offload")); + assert!(result.output.contains("<>").next().unwrap(); + assert_eq!(store.get(leaf_key).as_deref(), Some(prose.as_str())); + assert_eq!(store.len(), 2); +} + // ─── helpers ────────────────────────────────────────────────────── /// Pull the hash out of a `<>` row marker.