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): add structured prose offload plumbing (#334) (#2378)

Browse source 
## 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`.
This commit is contained in:
Rod Boev 2026-07-18 12:53:44 -04:00 • committed by GitHub
parent 8906d3a676
commit 9e0778553f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 626 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

10
crates/headroom-core/config/pipeline.toml
View file

@ -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

3
crates/headroom-core/src/transforms/mod.rs
View file

@ -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};

69
crates/headroom-core/src/transforms/pipeline/config.rs
View file

@ -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");

9
crates/headroom-core/src/transforms/pipeline/mod.rs
View file

@ -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::{

131
crates/headroom-core/src/transforms/pipeline/offloads/json_offload.rs
View file

@ -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<OffloadOutput, TransformError> {
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::<Vec<_>>()
.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("<<ccr:")
.nth(1)
.and_then(|tail| tail.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 = "<html><body><p>".to_string() + &"x".repeat(300) + "</p></body></html>";
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 {

2
crates/headroom-core/src/transforms/pipeline/offloads/mod.rs
View file

@ -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

170
crates/headroom-core/src/transforms/pipeline/offloads/prose_field.rs Normal file
View file

@ -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<OffloadOutput, TransformError> {
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::<Vec<_>>()
.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 = "<html><body><p>".to_string() + &"x".repeat(300) + "</p></body></html>";
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."));
}
}

4
crates/headroom-core/src/transforms/pipeline/orchestrator.rs
View file

@ -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();

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

@ -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::<Value>(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<Value> = 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<Value> = 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::<Vec<_>>()
.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 = "<html><body><p>".to_string() + &"x".repeat(300) + "</p></body></html>";
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::<Vec<_>>()
.join(" ");
let rows = (0..5)
.map(|i| serde_json::json!({"id": i, "summary": prose}))
.collect::<Vec<_>>();
let input = Value::Array(rows);
let hook = |leaf: &str, _query: &str| -> Option<(String, String)> {
if leaf.contains("recovery safeguards") {
Some(("<<ccr:prose-key>>".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("<<ccr:prose-key>>"));
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(("<<ccr:prose-key>>".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

40
crates/headroom-core/tests/ccr_roundtrip.rs
View file

@ -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::<Vec<_>>()
.join(" ");
let input = serde_json::json!((0..5)
.map(|i| serde_json::json!({"id": i, "summary": prose}))
.collect::<Vec<_>>())
.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("<<ccr:"));
let marker_start = result.output.find("<<ccr:").unwrap() + 6;
let leaf_key = result.output[marker_start..].split(">>").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 `<<ccr:HASH N_rows_offloaded>>` row marker.