From caf8fefdb368af9871fb908c1c3720213ce5e869 Mon Sep 17 00:00:00 2001 From: chopratejas Date: Wed, 29 Apr 2026 23:42:56 -0700 Subject: [PATCH] feat(rust): pipeline traits + orchestrator skeleton (Phase 3g PR1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces `CompressionPipeline` — the formal lossless-then-lossy orchestrator described in issue #315. Before this PR each compressor carried its own ad-hoc decision tree (SmartCrusher's 3c.2 refactor made it explicit for one transform; the rest still decide privately). This PR replaces that scaffolding with two traits, one orchestrator, and one real impl per trait. # Surface * `LosslessTransform` — `name()`, `applies_to()`, `apply(content)`. Preserves all information; orchestrator runs these first and stops early if the cumulative savings hit `lossless_target_ratio`. * `LossyTransform` — same shape plus `apply(content, ctx)` taking a `CompressionContext` (query + token budget) and a `confidence()` calibration score for telemetry. * `TransformResult { output, bytes_saved, structure_preserved, reversible_via }` — common return shape. * `TransformError { InvalidInput, Skipped, Internal }` — orchestrator treats all three as skip-this-transform; never panics. * `CompressionPipeline` + `CompressionPipelineBuilder` — sequential dispatch keyed on `ContentType`. Acceptance gate is `min_savings_ratio` (default 5%); lossless stop gate is `lossless_target_ratio` (default 50% of original). Per-step `name()` reporting feeds the strategy-stats JSONB nest from 3e.0. # Concrete impls (one real impl per trait — no speculative extraction) * `JsonMinifier` (lossless) — `serde_json::Value` round-trip. Pretty-printed JSON shrinks 25-35%; already-compact returns `bytes_saved == 0` and the orchestrator rejects it. * `LineImportanceFilter` (lossy) — consumes the existing `signals::LineImportanceDetector` trait. Walks `str::lines()`, scores each, drops below threshold, anchor-preserves first/last, collapses gaps into `[... N lines omitted ...]` markers. # No regex By project convention, nothing in this module uses the `regex` crate. JsonMinifier is pure `serde_json`. LineImportanceFilter walks lines and consumes the signals trait (aho-corasick + ASCII word-boundary post-filter, also no regex). # Touches outside the new module * `ContentType` gains `Hash` derive so the orchestrator can key transforms by content type. Trivially safe — the enum is already `Eq + PartialEq`. * `transforms/mod.rs` re-exports the new public surface. # Test plan (all green) * 41 unit tests across the four submodules: - 9 trait + error-handling tests - 9 JsonMinifier tests (pretty/compact/empty/malformed/Unicode/ deeply-nested/structure-preserved/reversible-via/applies-to) - 11 LineImportanceFilter tests (drop/keep/anchor windows/single- line/empty/gap-counting/confidence/structure-preserved/Unicode/ overlapping anchors/applies-to narrowing) - 12 orchestrator tests (empty pipeline/no-applicable/lossless- runs/rejection-below-min-ratio/error-recovery/lossless-target- stop/lossy-runs/lossy-compounds/structure-preserved-flag/ is-acceptable-zero-cases/min-savings-ratio/builder-dispatch/ builder-order) * `make ci-precheck` clean * `cargo fmt --all` + `cargo clippy` happy # Out of scope (lands in later PRs) * PR2: wrap existing structural transforms (Diff/Log/Search/Tag) in trait shape. Begins parallel-execution candidate via subagents. * PR3: SmartCrusher refactor to use the orchestrator + retire Python glue. * PR4: `ProseFieldCompressor` (parser/model boundary primitive, blocked on labeled corpus). * PR5: migrate text/search/log compressors and delete Python ContentRouter strategy dispatch. --- .../src/transforms/content_detector.rs | 2 +- crates/headroom-core/src/transforms/mod.rs | 6 + .../src/transforms/pipeline/json_minifier.rs | 179 +++++ .../pipeline/line_importance_filter.rs | 359 +++++++++ .../src/transforms/pipeline/mod.rs | 66 ++ .../src/transforms/pipeline/orchestrator.rs | 690 ++++++++++++++++++ .../src/transforms/pipeline/traits.rs | 268 +++++++ 7 files changed, 1569 insertions(+), 1 deletion(-) create mode 100644 crates/headroom-core/src/transforms/pipeline/json_minifier.rs create mode 100644 crates/headroom-core/src/transforms/pipeline/line_importance_filter.rs create mode 100644 crates/headroom-core/src/transforms/pipeline/mod.rs create mode 100644 crates/headroom-core/src/transforms/pipeline/orchestrator.rs create mode 100644 crates/headroom-core/src/transforms/pipeline/traits.rs diff --git a/crates/headroom-core/src/transforms/content_detector.rs b/crates/headroom-core/src/transforms/content_detector.rs index 57b46cd14..61aacd425 100644 --- a/crates/headroom-core/src/transforms/content_detector.rs +++ b/crates/headroom-core/src/transforms/content_detector.rs @@ -29,7 +29,7 @@ use serde_json::{json, Map, Value}; /// Content types recognized by the detector. String tags match Python's /// `ContentType` enum values 1:1. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum ContentType { JsonArray, SourceCode, diff --git a/crates/headroom-core/src/transforms/mod.rs b/crates/headroom-core/src/transforms/mod.rs index cbe60190b..aa465c7cc 100644 --- a/crates/headroom-core/src/transforms/mod.rs +++ b/crates/headroom-core/src/transforms/mod.rs @@ -22,6 +22,7 @@ pub mod detection; pub mod diff_compressor; pub mod log_compressor; pub mod magika_detector; +pub mod pipeline; pub mod search_compressor; pub mod smart_crusher; pub mod unidiff_detector; @@ -38,6 +39,11 @@ pub use log_compressor::{ LogLevel, LogLine, }; pub use magika_detector::{magika_detect, map_magika_label, MagikaDetectorError}; +pub use pipeline::{ + CompressionContext, CompressionPipeline, CompressionPipelineBuilder, JsonMinifier, + LineImportanceFilter, LineImportanceFilterConfig, LosslessTransform, LossyTransform, + PipelineConfig, PipelineResult, TransformError, TransformResult, +}; pub use search_compressor::{ FileMatches, SearchCompressionResult, SearchCompressor, SearchCompressorConfig, SearchCompressorStats, SearchMatch, diff --git a/crates/headroom-core/src/transforms/pipeline/json_minifier.rs b/crates/headroom-core/src/transforms/pipeline/json_minifier.rs new file mode 100644 index 000000000..d7a27b60d --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/json_minifier.rs @@ -0,0 +1,179 @@ +//! `JsonMinifier` — round-trip whitespace removal for JSON content. +//! +//! Smallest useful lossless transform: parse with `serde_json`, write +//! with `serde_json::to_string` (which omits all decorative +//! whitespace). The bytes saved depends on how indented the source is +//! — pretty-printed JSON typically shrinks by ~25–35%, already- +//! compact JSON saves ~0%. The orchestrator's +//! [`min_savings_ratio`] gate rejects no-savings runs automatically. +//! +//! [`min_savings_ratio`]: super::orchestrator::PipelineConfig::min_savings_ratio +//! +//! # Why not strip whitespace by hand +//! +//! Hand-rolled minification has to handle string literals, escape +//! sequences, and Unicode separator characters correctly. `serde_json` +//! already does that — and a parse-then-serialize roundtrip +//! validates the input as a side effect, which means downstream +//! consumers can trust the output is well-formed. The cost is one +//! `Value` allocation per call, which is dwarfed by the I/O the proxy +//! is already doing on the same payload. +//! +//! # No regex +//! +//! Pure `serde_json`. No regex, no hand-rolled scanner. + +use crate::transforms::pipeline::traits::{LosslessTransform, TransformError, TransformResult}; +use crate::transforms::ContentType; + +/// Minify any JSON document by parse-then-serialize round-trip. +#[derive(Debug, Default, Clone, Copy)] +pub struct JsonMinifier; + +impl JsonMinifier { + pub const NAME: &'static str = "json_minifier"; + + /// The content types this transform claims. Static slice — the + /// applicability is fixed by the implementation, not configured + /// per instance. + const APPLIES_TO: &'static [ContentType] = &[ContentType::JsonArray]; +} + +impl LosslessTransform for JsonMinifier { + fn name(&self) -> &'static str { + Self::NAME + } + + fn applies_to(&self) -> &[ContentType] { + Self::APPLIES_TO + } + + fn apply(&self, content: &str) -> Result { + if content.is_empty() { + return Err(TransformError::skipped(Self::NAME, "empty input")); + } + // `from_str` validates the input as a side effect of building + // the in-memory tree. Any structural error becomes + // InvalidInput so the orchestrator can move on. + let value: serde_json::Value = serde_json::from_str(content).map_err(|e| { + TransformError::invalid_input(Self::NAME, format!("not valid json: {e}")) + })?; + let minified = serde_json::to_string(&value) + .map_err(|e| TransformError::internal(Self::NAME, format!("serialize failed: {e}")))?; + Ok(TransformResult::from_lengths(content.len(), minified, true)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn run(input: &str) -> Result { + JsonMinifier.apply(input) + } + + #[test] + fn name_matches_telemetry_convention() { + // Lowercase snake_case so the strategy-stats JSONB nest stays + // clean. + assert_eq!(JsonMinifier.name(), "json_minifier"); + } + + #[test] + fn applies_to_json_array() { + assert_eq!(JsonMinifier.applies_to(), &[ContentType::JsonArray]); + } + + #[test] + fn pretty_printed_json_shrinks() { + let pretty = r#"{ + "name": "Alice", + "age": 30, + "tags": [ + "engineer", + "manager" + ] +}"#; + let result = run(pretty).expect("valid json should compress"); + assert!(result.bytes_saved > 0); + assert!(result.structure_preserved); + assert!(!result.output.contains('\n')); + assert!(!result.output.contains(" ")); + // Round-trips back to the same logical value. + let v_in: serde_json::Value = serde_json::from_str(pretty).unwrap(); + let v_out: serde_json::Value = serde_json::from_str(&result.output).unwrap(); + assert_eq!(v_in, v_out); + } + + #[test] + fn already_compact_json_returns_zero_savings() { + let compact = r#"{"a":1,"b":[1,2,3]}"#; + let result = run(compact).expect("compact json is still valid"); + // serde_json's canonical form may match the input byte-for-byte + // here; the important contract is bytes_saved == 0 (orchestrator + // will reject this in is_acceptable). + assert_eq!(result.bytes_saved, 0); + } + + #[test] + fn empty_string_is_skipped_not_invalid() { + let err = run("").expect_err("empty input is a skip, not an error"); + assert!(matches!(err, TransformError::Skipped { .. })); + } + + #[test] + fn malformed_json_is_invalid_input() { + let err = run("{not json").expect_err("garbage is invalid input"); + assert!(matches!(err, TransformError::InvalidInput { .. })); + } + + #[test] + fn nested_arrays_round_trip() { + let nested = r#"{ + "users": [ + {"id": 1, "name": "a"}, + {"id": 2, "name": "b"} + ] +}"#; + let result = run(nested).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&result.output).unwrap(); + assert_eq!(parsed["users"][0]["id"], 1); + assert_eq!(parsed["users"][1]["name"], "b"); + assert!(result.bytes_saved > 0); + } + + #[test] + fn unicode_strings_preserved() { + let input = r#"{ "greeting": "héllo wörld 🌍" }"#; + let result = run(input).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&result.output).unwrap(); + assert_eq!(parsed["greeting"], "héllo wörld 🌍"); + } + + #[test] + fn structure_preserved_flag_is_always_true() { + // Lossless by construction. + let result = run(r#"{"a": 1}"#).unwrap(); + assert!(result.structure_preserved); + assert!(result.reversible_via.is_none()); + } + + #[test] + fn deeply_nested_does_not_blow_up() { + // 50 levels of nesting — well under any reasonable parse limit + // but enough to catch a recursion bug if we ever introduce one. + let mut s = String::new(); + for _ in 0..50 { + s.push('['); + } + s.push('1'); + for _ in 0..50 { + s.push(']'); + } + let result = run(&s).unwrap(); + // Parsing+serializing a deeply nested array doesn't add or + // remove much — important is that we don't panic or fail. + let parsed: serde_json::Value = serde_json::from_str(&result.output).unwrap(); + assert!(parsed.is_array()); + } +} diff --git a/crates/headroom-core/src/transforms/pipeline/line_importance_filter.rs b/crates/headroom-core/src/transforms/pipeline/line_importance_filter.rs new file mode 100644 index 000000000..6c9caa3d6 --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/line_importance_filter.rs @@ -0,0 +1,359 @@ +//! `LineImportanceFilter` — drop low-priority lines using the signals +//! trait from Phase 3e.1. +//! +//! Smallest useful lossy transform: walk lines, score each via a +//! [`LineImportanceDetector`] (the trait shipped in Phase 3e.1), keep +//! lines whose `priority` is at or above a threshold, plus unconditional +//! anchors at the head and tail. Gaps between kept lines collapse into +//! `[... N lines omitted ...]` markers so the downstream LLM has a +//! visible signal that bytes were dropped. +//! +//! # Why this is lossy +//! +//! Dropped lines are gone — the filter does not emit a CCR retrieval +//! handle in PR1 (CCR offload moves into the pipeline in PR3). The +//! omission markers carry coarse "something was here" signal but not +//! the original bytes. Hence `structure_preserved = false` and +//! `confidence = 0.7` (calibrated from the detector's own confidence +//! aggregate; tuned later in PR4 once we have a labeled corpus). +//! +//! # No regex +//! +//! Walks `str::lines()`. The detector itself is aho-corasick + ASCII +//! word-boundary post-filter, also no regex. + +use crate::signals::{ImportanceContext, LineImportanceDetector}; +use crate::transforms::pipeline::traits::{ + CompressionContext, LossyTransform, TransformError, TransformResult, +}; +use crate::transforms::ContentType; + +/// Configuration knobs for [`LineImportanceFilter`]. +#[derive(Debug, Clone)] +pub struct LineImportanceFilterConfig { + /// Minimum priority for a line to survive the filter. Lines with + /// `priority < min_priority` get dropped (unless they're inside + /// the head/tail anchor windows). Default `0.4` — picked to keep + /// warning/importance lines (priority 0.5) and drop neutral lines + /// (priority 0.0) without aggressively trimming borderline cases. + pub min_priority: f32, + /// Always keep the first N lines. Header context, banner, summary + /// — losing these breaks the LLM's ability to interpret what + /// follows. + pub keep_first: usize, + /// Always keep the last N lines. Trailing summaries, exit codes, + /// totals — same logic as the head anchor. + pub keep_last: usize, + /// Importance context to pass to the detector. Different contexts + /// fire different keyword sets (markdown structure matters in + /// `Text`, doesn't in `Diff`). + pub context: ImportanceContext, +} + +impl Default for LineImportanceFilterConfig { + fn default() -> Self { + Self { + min_priority: 0.4, + keep_first: 5, + keep_last: 5, + context: ImportanceContext::Text, + } + } +} + +/// Drop low-priority lines, keep anchors at head/tail. +pub struct LineImportanceFilter { + detector: Box, + applies_to: Vec, + config: LineImportanceFilterConfig, +} + +impl LineImportanceFilter { + pub const NAME: &'static str = "line_importance_filter"; + + /// Build a filter with the given detector and the default config. + /// `applies_to` defaults to the broad set of line-based content + /// types — caller can narrow via [`with_applies_to`]. + /// + /// [`with_applies_to`]: Self::with_applies_to + pub fn new(detector: Box) -> Self { + Self { + detector, + applies_to: vec![ + ContentType::PlainText, + ContentType::BuildOutput, + ContentType::SearchResults, + ], + config: LineImportanceFilterConfig::default(), + } + } + + pub fn with_config(mut self, config: LineImportanceFilterConfig) -> Self { + self.config = config; + self + } + + pub fn with_applies_to(mut self, types: Vec) -> Self { + self.applies_to = types; + self + } +} + +impl LossyTransform for LineImportanceFilter { + fn name(&self) -> &'static str { + Self::NAME + } + + fn applies_to(&self) -> &[ContentType] { + &self.applies_to + } + + fn apply( + &self, + content: &str, + _ctx: &CompressionContext, + ) -> Result { + if content.is_empty() { + return Err(TransformError::skipped(Self::NAME, "empty input")); + } + + // `str::lines` iterates without retaining line terminators; we + // collect once because the keep-decision per line depends on + // the total count (head/tail anchors). + let lines: Vec<&str> = content.lines().collect(); + let n = lines.len(); + if n == 0 { + return Err(TransformError::skipped(Self::NAME, "no lines")); + } + + // Score each line and decide keep-or-drop. + let mut keep = vec![false; n]; + let head_end = self.config.keep_first.min(n); + let tail_start = n.saturating_sub(self.config.keep_last); + for (i, line) in lines.iter().enumerate() { + if i < head_end || i >= tail_start { + keep[i] = true; + continue; + } + let signal = self.detector.score(line, self.config.context); + if signal.priority >= self.config.min_priority { + keep[i] = true; + } + } + + // Build output, collapsing dropped runs into omission markers. + let mut out = String::with_capacity(content.len()); + let mut last_kept: Option = None; + for (i, line) in lines.iter().enumerate() { + if !keep[i] { + continue; + } + if let Some(prev) = last_kept { + let gap = i - prev - 1; + if gap > 0 { + if !out.is_empty() { + out.push('\n'); + } + out.push_str(&format!("[... {gap} lines omitted ...]")); + } + } + if !out.is_empty() { + out.push('\n'); + } + out.push_str(line); + last_kept = Some(i); + } + if let Some(prev) = last_kept { + let gap = n - prev - 1; + if gap > 0 { + out.push('\n'); + out.push_str(&format!("[... {gap} lines omitted ...]")); + } + } + + // Bytes-saved is computed against the original content, not + // the line-iterated reconstruction (which can differ by the + // trailing newline). We account for that by using the + // input length directly. + let bytes_saved = content.len().saturating_sub(out.len()); + Ok(TransformResult { + output: out, + bytes_saved, + structure_preserved: false, + reversible_via: None, + }) + } + + fn confidence(&self) -> f32 { + // Calibrated 0.7: the detector is reliable on keyword signals + // (priority 0.5+) but has no view into semantic relevance to + // the user's query — that's what PR4 ProseFieldCompressor + // adds. 0.7 is "decent extractive baseline." + 0.7 + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::signals::{ImportanceCategory, ImportanceSignal}; + + /// Test detector with deterministic priorities — keyed on a + /// substring match. Avoids spinning up the real keyword detector + /// for unit tests. + struct StubDetector { + priority_keyword: &'static str, + priority: f32, + } + impl LineImportanceDetector for StubDetector { + fn score(&self, line: &str, _ctx: ImportanceContext) -> ImportanceSignal { + if line.contains(self.priority_keyword) { + ImportanceSignal::matched(ImportanceCategory::Importance, self.priority, 0.9) + } else { + ImportanceSignal::neutral() + } + } + } + + fn run(filter: &LineImportanceFilter, content: &str) -> TransformResult { + filter + .apply(content, &CompressionContext::default()) + .expect("test inputs are well-formed") + } + + fn filter_keep_keyword() -> LineImportanceFilter { + LineImportanceFilter::new(Box::new(StubDetector { + priority_keyword: "KEEP", + priority: 0.9, + })) + .with_config(LineImportanceFilterConfig { + min_priority: 0.5, + keep_first: 1, + keep_last: 1, + context: ImportanceContext::Text, + }) + } + + #[test] + fn name_matches_telemetry_convention() { + let f = filter_keep_keyword(); + assert_eq!(f.name(), "line_importance_filter"); + } + + #[test] + fn applies_to_default_set_includes_plain_text_logs_and_search() { + let f = filter_keep_keyword(); + let types = f.applies_to(); + assert!(types.contains(&ContentType::PlainText)); + assert!(types.contains(&ContentType::BuildOutput)); + assert!(types.contains(&ContentType::SearchResults)); + } + + #[test] + fn drops_low_priority_lines_keeps_high_priority() { + let f = filter_keep_keyword(); + let input = (0..20) + .map(|i| { + if i == 10 { + "KEEP this line".into() + } else { + format!("noise {i}") + } + }) + .collect::>() + .join("\n"); + let result = run(&f, &input); + assert!(result.output.contains("KEEP this line")); + assert!(result.output.contains("[... ") && result.output.contains("lines omitted ...]")); + assert!(result.bytes_saved > 0); + assert!(!result.structure_preserved); + } + + #[test] + fn always_keeps_first_and_last_anchor_windows() { + let f = filter_keep_keyword(); + let lines: Vec = (0..20).map(|i| format!("plain line {i}")).collect(); + let input = lines.join("\n"); + let result = run(&f, &input); + // First line and last line must survive even when nothing + // matches the keyword. + assert!(result.output.contains("plain line 0")); + assert!(result.output.contains("plain line 19")); + } + + #[test] + fn empty_input_returns_skipped_error() { + let f = filter_keep_keyword(); + let err = f + .apply("", &CompressionContext::default()) + .expect_err("empty input is a skip"); + assert!(matches!(err, TransformError::Skipped { .. })); + } + + #[test] + fn single_line_input_passes_through() { + let f = filter_keep_keyword(); + let result = run(&f, "only one line"); + // Head anchor keeps it; no gap markers possible. + assert_eq!(result.output, "only one line"); + assert_eq!(result.bytes_saved, 0); + } + + #[test] + fn omission_markers_count_dropped_lines_correctly() { + // 5 lines: keep 1st (head anchor), drop middle 3 (no keyword), + // keep last (tail anchor). Expect ONE marker reporting 3 + // omitted lines. + let f = filter_keep_keyword(); + let input = ["first", "drop a", "drop b", "drop c", "last"].join("\n"); + let result = run(&f, &input); + assert!(result.output.starts_with("first")); + assert!(result.output.ends_with("last")); + assert!(result.output.contains("[... 3 lines omitted ...]")); + } + + #[test] + fn confidence_is_calibrated_constant_in_pr1() { + let f = filter_keep_keyword(); + assert!((f.confidence() - 0.7).abs() < f32::EPSILON); + } + + #[test] + fn structure_preserved_flag_is_always_false() { + let f = filter_keep_keyword(); + let result = run(&f, "first\nKEEP me\nlast"); + assert!(!result.structure_preserved); + } + + #[test] + fn anchor_windows_overlap_safely_on_short_input() { + // 3 lines, keep_first=5, keep_last=5 — windows fully overlap, + // every line should survive without panicking on the index + // arithmetic. + let f = LineImportanceFilter::new(Box::new(StubDetector { + priority_keyword: "_", + priority: 0.9, + })) + .with_config(LineImportanceFilterConfig { + min_priority: 0.5, + keep_first: 5, + keep_last: 5, + context: ImportanceContext::Text, + }); + let result = run(&f, "a\nb\nc"); + assert_eq!(result.output, "a\nb\nc"); + } + + #[test] + fn applies_to_can_be_narrowed() { + let f = filter_keep_keyword().with_applies_to(vec![ContentType::SearchResults]); + assert_eq!(f.applies_to(), &[ContentType::SearchResults]); + } + + #[test] + fn unicode_lines_handled_safely() { + let f = filter_keep_keyword(); + let result = run(&f, "first\nKEEP héllo wörld 🌍\nlast"); + assert!(result.output.contains("héllo wörld 🌍")); + } +} diff --git a/crates/headroom-core/src/transforms/pipeline/mod.rs b/crates/headroom-core/src/transforms/pipeline/mod.rs new file mode 100644 index 000000000..eae75b220 --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/mod.rs @@ -0,0 +1,66 @@ +//! Compression pipeline — formal orchestrator for lossless → lossy → CCR. +//! +//! # Why this module exists +//! +//! Before Phase 3g each compressor (SmartCrusher, DiffCompressor, +//! LogCompressor, SearchCompressor, TagProtector) carried its own +//! ad-hoc decision tree: parse, score, drop, optionally emit a CCR +//! marker. SmartCrusher's 3c.2 refactor made the decision explicit +//! inside that one transform; the rest still decide privately. That +//! makes the system hard to reason about (which transforms ran in +//! what order, how much each saved, why CCR fired or didn't) and +//! impossible to extend without copying the same scaffolding. +//! +//! This module replaces that scaffolding with two traits and an +//! orchestrator: +//! +//! * [`LosslessTransform`] — preserves all information; can run +//! first and stop early if structural compression suffices. +//! * [`LossyTransform`] — drops bytes deliberately. Runs only after +//! the lossless pass. Reports a calibrated `confidence` so callers +//! and telemetry can tell which transforms were trusted vs taken +//! on faith. +//! * [`CompressionPipeline`] — content-type-keyed dispatch over the +//! two transform sets, with budget gates between phases. +//! +//! # The pipeline contract +//! +//! ```text +//! input → DetectContentType (Magika, already shipped) +//! → LosslessTransforms[content_type] stop if structural compression suffices +//! → BudgetCheck +//! → LossyTransforms[content_type] includes ProseFieldCompressor (Phase 3g PR4) +//! → BudgetCheck +//! → CCR for everything beyond budget (Phase 3g PR3+) +//! ``` +//! +//! PR1 ships traits + orchestrator + one real impl per trait +//! ([`JsonMinifier`] for lossless, [`LineImportanceFilter`] for lossy). +//! PR2 wraps existing structural transforms (Diff/Log/Search/Tag) in +//! the trait shape. PR3 refactors SmartCrusher to use the orchestrator +//! and starts retiring Python orchestration glue. PR4 adds +//! ProseFieldCompressor (the parser/model boundary primitive). PR5 +//! migrates the remaining compressors and lets us delete the Python +//! `ContentRouter` strategy dispatch. +//! +//! # No regex +//! +//! By project convention (and feedback memory), nothing in this module +//! uses the `regex` crate. JsonMinifier is `serde_json` round-trip; +//! LineImportanceFilter walks `str::lines()` and consumes the existing +//! `signals::LineImportanceDetector` trait (which uses aho-corasick +//! plus an ASCII word-boundary post-filter, also no regex). + +pub mod json_minifier; +pub mod line_importance_filter; +pub mod orchestrator; +pub mod traits; + +pub use json_minifier::JsonMinifier; +pub use line_importance_filter::{LineImportanceFilter, LineImportanceFilterConfig}; +pub use orchestrator::{ + CompressionPipeline, CompressionPipelineBuilder, PipelineConfig, PipelineResult, +}; +pub use traits::{ + CompressionContext, LosslessTransform, LossyTransform, TransformError, TransformResult, +}; diff --git a/crates/headroom-core/src/transforms/pipeline/orchestrator.rs b/crates/headroom-core/src/transforms/pipeline/orchestrator.rs new file mode 100644 index 000000000..03ddd9fd7 --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/orchestrator.rs @@ -0,0 +1,690 @@ +//! `CompressionPipeline` — content-type-keyed dispatch over lossless +//! and lossy transforms. +//! +//! # Decision flow +//! +//! ```text +//! input + content_type +//! │ +//! ▼ +//! ┌────────────────────────────────────┐ +//! │ Lossless transforms for this type │ ──┐ +//! │ for each in registration order: │ │ stop early if +//! │ try apply │ │ current_len/orig_len +//! │ accept iff is_acceptable() │ │ falls below +//! │ │ │ lossless_target_ratio +//! └────────────────────────────────────┘ ──┘ +//! │ +//! ▼ +//! ┌────────────────────────────────────┐ +//! │ Lossy transforms for this type │ +//! │ for each in registration order: │ +//! │ try apply │ +//! │ accept iff is_acceptable() │ +//! │ flag structure_preserved=false │ +//! └────────────────────────────────────┘ +//! │ +//! ▼ steps_applied[], bytes_saved, structure_preserved, reversible_via +//! ``` +//! +//! Acceptance gate (`is_acceptable`): +//! * Reject if output isn't strictly shorter than the input it just +//! processed (a transform that grew the content gets discarded). +//! * Reject if `bytes_saved / current_len < min_savings_ratio`. The +//! default is 5% — transforms that move bytes by less than that +//! aren't worth the runtime overhead they impose. +//! +//! Skip-further-lossless gate (`should_keep_running_lossless`): +//! * If the running output is already at or below +//! `lossless_target_ratio * original_len`, stop running additional +//! lossless transforms and move straight to lossy. Default 0.5 — +//! "we've cut input in half losslessly, no need to keep +//! structural-pass churning." +//! +//! Per-transform errors are logged at TRACE level and treated as +//! skips. The pipeline never panics on a transform failure; the +//! callers can rely on getting *some* output back, even if it's the +//! original input verbatim (all transforms skipped). + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::transforms::pipeline::traits::{ + CompressionContext, LosslessTransform, LossyTransform, TransformResult, +}; +use crate::transforms::ContentType; + +/// Runtime knobs for the orchestrator. Defaults chosen empirically; +/// callers can override per-pipeline. +#[derive(Debug, Clone, Copy)] +pub struct PipelineConfig { + /// Minimum savings ratio (saved / current_len) for a transform's + /// output to be accepted. Default 0.05 — transforms that move + /// fewer than 5% of bytes don't earn the risk of having moved + /// the wrong ones. + pub min_savings_ratio: f64, + /// Stop the lossless pass once the running output drops below + /// this fraction of the original input length. Default 0.5 — + /// once we've cut by half losslessly, downstream lossy passes + /// can do the rest cheaper than another structural transform. + pub lossless_target_ratio: f64, +} + +impl Default for PipelineConfig { + fn default() -> Self { + Self { + min_savings_ratio: 0.05, + lossless_target_ratio: 0.5, + } + } +} + +/// Result returned by [`CompressionPipeline::run`]. +#[derive(Debug, Clone, Default)] +pub struct PipelineResult { + /// Final compressed output. Equal to the input if every transform + /// skipped or was rejected. + pub output: String, + /// Total bytes removed across all accepted transforms. + pub bytes_saved: usize, + /// True iff every accepted transform preserved structure. A + /// single accepted lossy transform flips this to false. + pub structure_preserved: bool, + /// Names of accepted transforms in execution order. Maps 1:1 to + /// the per-strategy stats nest from Phase 3e.0. + pub steps_applied: Vec, + /// Most recent CCR cache key emitted by an accepted transform. + /// PR1 transforms never set this; the field exists for PR2/PR3 + /// integrations. + pub reversible_via: Option, +} + +/// Sequential lossless-then-lossy pipeline keyed on `ContentType`. +/// +/// Construct via [`builder`](Self::builder), then call +/// [`run`](Self::run) on each input. +pub struct CompressionPipeline { + lossless_by_type: HashMap>>, + lossy_by_type: HashMap>>, + config: PipelineConfig, +} + +impl CompressionPipeline { + pub fn builder() -> CompressionPipelineBuilder { + CompressionPipelineBuilder::default() + } + + /// Run the configured pipeline against `content`. Always returns a + /// [`PipelineResult`] — failures inside individual transforms are + /// recorded in tracing and turn into skips, never panics. + pub fn run( + &self, + content: &str, + content_type: ContentType, + ctx: &CompressionContext, + ) -> PipelineResult { + let original_len = content.len(); + let mut current = content.to_string(); + let mut total_saved: usize = 0; + let mut structure_preserved = true; + let mut steps: Vec = Vec::new(); + let mut reversible: Option = None; + + // Phase 1 — lossless. Run in registration order, stop if we've + // hit the lossless target ratio. + if let Some(lossless) = self.lossless_by_type.get(&content_type) { + for transform in lossless { + if !self.should_keep_running_lossless(current.len(), original_len) { + tracing::trace!( + target: "headroom::pipeline", + transform = transform.name(), + current_len = current.len(), + original_len, + "lossless target reached, stopping lossless phase" + ); + break; + } + match transform.apply(¤t) { + Ok(result) => { + if !self.is_acceptable(result.bytes_saved, current.len()) { + tracing::trace!( + target: "headroom::pipeline", + transform = transform.name(), + bytes_saved = result.bytes_saved, + "lossless transform rejected: insufficient savings" + ); + continue; + } + Self::accept( + &mut current, + &mut total_saved, + &mut steps, + &mut reversible, + &mut structure_preserved, + transform.name(), + result, + ); + } + Err(e) => { + tracing::trace!( + target: "headroom::pipeline", + transform = transform.name(), + error = %e, + "lossless transform errored" + ); + } + } + } + } + + // Phase 2 — lossy. Always runs (PR3 will gate this on token + // budget once the tokenizer hookup lands). + if let Some(lossy) = self.lossy_by_type.get(&content_type) { + for transform in lossy { + match transform.apply(¤t, ctx) { + Ok(result) => { + if !self.is_acceptable(result.bytes_saved, current.len()) { + tracing::trace!( + target: "headroom::pipeline", + transform = transform.name(), + bytes_saved = result.bytes_saved, + "lossy transform rejected: insufficient savings" + ); + continue; + } + Self::accept( + &mut current, + &mut total_saved, + &mut steps, + &mut reversible, + &mut structure_preserved, + transform.name(), + result, + ); + } + Err(e) => { + tracing::trace!( + target: "headroom::pipeline", + transform = transform.name(), + error = %e, + "lossy transform errored" + ); + } + } + } + } + + PipelineResult { + output: current, + bytes_saved: total_saved, + structure_preserved, + steps_applied: steps, + reversible_via: reversible, + } + } + + /// Centralized accept logic. Updates the running output, savings, + /// step list, reversibility handle, and structure flag as one + /// step so the lossless and lossy phases share semantics. + fn accept( + current: &mut String, + total_saved: &mut usize, + steps: &mut Vec, + reversible: &mut Option, + structure_preserved: &mut bool, + name: &'static str, + result: TransformResult, + ) { + *total_saved = total_saved.saturating_add(result.bytes_saved); + if !result.structure_preserved { + *structure_preserved = false; + } + if let Some(handle) = result.reversible_via { + *reversible = Some(handle); + } + *current = result.output; + steps.push(name.to_string()); + } + + /// Acceptance gate — whether this transform's savings on the + /// current state of the buffer are worth keeping. + pub(crate) fn is_acceptable(&self, bytes_saved: usize, current_len: usize) -> bool { + if bytes_saved == 0 || current_len == 0 { + return false; + } + let ratio = bytes_saved as f64 / current_len as f64; + ratio >= self.config.min_savings_ratio + } + + /// Whether the lossless phase should keep running. False once + /// `current_len / original_len <= lossless_target_ratio`. + pub(crate) fn should_keep_running_lossless( + &self, + current_len: usize, + original_len: usize, + ) -> bool { + if original_len == 0 { + return false; + } + let ratio = current_len as f64 / original_len as f64; + ratio > self.config.lossless_target_ratio + } +} + +/// Fluent builder for [`CompressionPipeline`]. The example in +/// `issue #315` is the spec for this surface. +#[derive(Default)] +pub struct CompressionPipelineBuilder { + lossless_by_type: HashMap>>, + lossy_by_type: HashMap>>, + config: Option, +} + +impl CompressionPipelineBuilder { + /// Register a lossless transform. The transform is added to the + /// pipeline for every `ContentType` it self-declares via + /// `applies_to()`. Order of registration is order of execution. + pub fn with_lossless(mut self, transform: T) -> Self + where + T: LosslessTransform + 'static, + { + let arc: Arc = Arc::new(transform); + let types: Vec = arc.applies_to().to_vec(); + for ct in types { + self.lossless_by_type + .entry(ct) + .or_default() + .push(arc.clone()); + } + self + } + + pub fn with_lossy(mut self, transform: T) -> Self + where + T: LossyTransform + 'static, + { + let arc: Arc = Arc::new(transform); + let types: Vec = arc.applies_to().to_vec(); + for ct in types { + self.lossy_by_type.entry(ct).or_default().push(arc.clone()); + } + self + } + + pub fn with_config(mut self, config: PipelineConfig) -> Self { + self.config = Some(config); + self + } + + pub fn build(self) -> CompressionPipeline { + CompressionPipeline { + lossless_by_type: self.lossless_by_type, + lossy_by_type: self.lossy_by_type, + config: self.config.unwrap_or_default(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::signals::{ + ImportanceCategory, ImportanceContext, ImportanceSignal, LineImportanceDetector, + }; + use crate::transforms::pipeline::traits::{ + CompressionContext, LosslessTransform, LossyTransform, TransformError, TransformResult, + }; + use crate::transforms::pipeline::{ + JsonMinifier, LineImportanceFilter, LineImportanceFilterConfig, + }; + + // ── Test helpers ───────────────────────────────────────────────── + + /// Trivial lossless transform that always saves exactly 1 byte + /// (drops a trailing newline if present, otherwise reports + /// bytes_saved == 0). + struct TrivialLossless; + impl LosslessTransform for TrivialLossless { + fn name(&self) -> &'static str { + "trivial_lossless" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, content: &str) -> Result { + let trimmed = content.trim_end_matches('\n').to_string(); + Ok(TransformResult::from_lengths(content.len(), trimmed, true)) + } + } + + /// Lossless transform that always errors. Used to verify the + /// orchestrator continues past failures. + struct AlwaysErrors; + impl LosslessTransform for AlwaysErrors { + fn name(&self) -> &'static str { + "always_errors" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, _content: &str) -> Result { + Err(TransformError::invalid_input("always_errors", "by design")) + } + } + + /// Stub detector for the LineImportanceFilter integration tests. + struct StubDetector; + impl LineImportanceDetector for StubDetector { + fn score(&self, line: &str, _ctx: ImportanceContext) -> ImportanceSignal { + if line.contains("KEEP") { + ImportanceSignal::matched(ImportanceCategory::Importance, 0.9, 0.9) + } else { + ImportanceSignal::neutral() + } + } + } + + fn ctx() -> CompressionContext { + CompressionContext::default() + } + + // ── Empty-pipeline tests ───────────────────────────────────────── + + #[test] + fn empty_pipeline_passes_input_through_unchanged() { + let p = CompressionPipeline::builder().build(); + let input = "hello world"; + let r = p.run(input, ContentType::PlainText, &ctx()); + assert_eq!(r.output, input); + assert_eq!(r.bytes_saved, 0); + assert!(r.steps_applied.is_empty()); + assert!(r.structure_preserved); + assert!(r.reversible_via.is_none()); + } + + #[test] + fn pipeline_with_no_applicable_transforms_passes_through() { + // Register a JSON-only transform; run on plain text. + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .build(); + let r = p.run("not json", ContentType::PlainText, &ctx()); + assert_eq!(r.output, "not json"); + assert!(r.steps_applied.is_empty()); + } + + // ── Lossless phase ─────────────────────────────────────────────── + + #[test] + fn lossless_runs_when_applicable() { + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .build(); + let pretty = r#"{ + "a": 1, + "b": [1, 2, 3] +}"#; + let r = p.run(pretty, ContentType::JsonArray, &ctx()); + assert!(r.bytes_saved > 0); + assert_eq!(r.steps_applied, vec!["json_minifier".to_string()]); + assert!(r.structure_preserved); + assert!(r.output.len() < pretty.len()); + } + + #[test] + fn lossless_rejects_below_min_savings_ratio() { + // Compact JSON minifies to itself → bytes_saved == 0 → rejected. + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .build(); + let r = p.run(r#"{"a":1}"#, ContentType::JsonArray, &ctx()); + assert_eq!(r.bytes_saved, 0); + assert!(r.steps_applied.is_empty()); + } + + #[test] + fn lossless_continues_past_error() { + let p = CompressionPipeline::builder() + .with_lossless(AlwaysErrors) + .with_lossless(TrivialLossless) + .with_config(PipelineConfig { + min_savings_ratio: 0.0, // accept any savings + lossless_target_ratio: 0.0, + }) + .build(); + let r = p.run("hello\n", ContentType::PlainText, &ctx()); + // Both transforms run; first errors and is recorded as skip, + // second saves 1 byte. Without the per-error-recovery loop the + // pipeline would short-circuit. + assert_eq!(r.steps_applied, vec!["trivial_lossless".to_string()]); + assert_eq!(r.bytes_saved, 1); + } + + #[test] + fn lossless_stops_once_target_ratio_reached() { + struct HalvesContent; + impl LosslessTransform for HalvesContent { + fn name(&self) -> &'static str { + "halver" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, content: &str) -> Result { + let half = &content[..content.len() / 2]; + Ok(TransformResult::from_lengths( + content.len(), + half.to_string(), + true, + )) + } + } + // First halver: 100 → 50 bytes (ratio 0.5 — at the gate). + // Second halver: should NOT run because ratio is now <= 0.5. + let p = CompressionPipeline::builder() + .with_lossless(HalvesContent) + .with_lossless(HalvesContent) + .build(); + let input = "x".repeat(100); + let r = p.run(&input, ContentType::PlainText, &ctx()); + assert_eq!(r.steps_applied.len(), 1); + assert_eq!(r.output.len(), 50); + } + + // ── Lossy phase ────────────────────────────────────────────────── + + #[test] + fn lossy_runs_after_lossless() { + let lif = LineImportanceFilter::new(Box::new(StubDetector)).with_config( + LineImportanceFilterConfig { + min_priority: 0.5, + keep_first: 1, + keep_last: 1, + context: ImportanceContext::Text, + }, + ); + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) // doesn't apply to plain text + .with_lossy(lif) + .build(); + let input = (0..20) + .map(|i| { + if i == 10 { + "KEEP me".into() + } else { + format!("noise {i}") + } + }) + .collect::>() + .join("\n"); + let r = p.run(&input, ContentType::PlainText, &ctx()); + assert!(r.bytes_saved > 0); + assert_eq!(r.steps_applied, vec!["line_importance_filter".to_string()]); + assert!(!r.structure_preserved); + assert!(r.output.contains("KEEP me")); + } + + #[test] + fn lossy_after_lossless_compounds_savings() { + // Run JsonMinifier on JsonArray (saves whitespace), then a + // hypothetical lossy on top. We just verify the orchestrator + // accumulates bytes_saved across phases. + struct JsonAsTextLossy; + impl LossyTransform for JsonAsTextLossy { + fn name(&self) -> &'static str { + "json_truncator" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::JsonArray] + } + fn apply( + &self, + content: &str, + _ctx: &CompressionContext, + ) -> Result { + let cut = content.len() / 2; + Ok(TransformResult { + output: format!("{}…", &content[..cut]), + bytes_saved: content.len().saturating_sub(cut + 3), + structure_preserved: false, + reversible_via: None, + }) + } + fn confidence(&self) -> f32 { + 0.4 + } + } + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .with_lossy(JsonAsTextLossy) + .build(); + let pretty = r#"{ + "users": [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}] +}"#; + let r = p.run(pretty, ContentType::JsonArray, &ctx()); + assert_eq!( + r.steps_applied, + vec!["json_minifier".to_string(), "json_truncator".to_string()] + ); + assert!(!r.structure_preserved); + assert!(r.bytes_saved > 0); + } + + // ── Structure-preservation flag ───────────────────────────────── + + #[test] + fn structure_preserved_stays_true_when_only_lossless_runs() { + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .build(); + let r = p.run(r#"{ "a": 1, "b": 2 }"#, ContentType::JsonArray, &ctx()); + assert!(r.structure_preserved); + } + + #[test] + fn structure_preserved_flips_when_lossy_runs() { + let lif = LineImportanceFilter::new(Box::new(StubDetector)); + let p = CompressionPipeline::builder() + .with_lossy(lif) + .with_config(PipelineConfig { + min_savings_ratio: 0.0, + lossless_target_ratio: 0.5, + }) + .build(); + let many = (0..30) + .map(|i| format!("plain line {i}")) + .collect::>() + .join("\n"); + let r = p.run(&many, ContentType::PlainText, &ctx()); + if !r.steps_applied.is_empty() { + assert!(!r.structure_preserved); + } + } + + // ── Acceptance gate edge cases ─────────────────────────────────── + + #[test] + fn is_acceptable_handles_zero_lengths() { + let p = CompressionPipeline::builder().build(); + assert!(!p.is_acceptable(0, 100), "no savings → reject"); + assert!(!p.is_acceptable(50, 0), "current_len 0 → reject"); + } + + #[test] + fn is_acceptable_uses_min_savings_ratio() { + let p = CompressionPipeline::builder() + .with_config(PipelineConfig { + min_savings_ratio: 0.10, + lossless_target_ratio: 0.5, + }) + .build(); + assert!(!p.is_acceptable(5, 100), "5% < 10% threshold → reject"); + assert!(p.is_acceptable(15, 100), "15% > 10% threshold → accept"); + } + + #[test] + fn should_keep_running_lossless_handles_zero_original() { + let p = CompressionPipeline::builder().build(); + assert!(!p.should_keep_running_lossless(0, 0)); + } + + // ── Builder-side properties ────────────────────────────────────── + + #[test] + fn builder_dispatches_by_applies_to() { + let p = CompressionPipeline::builder() + .with_lossless(JsonMinifier) + .with_lossless(TrivialLossless) + .build(); + // JsonMinifier registered for JsonArray; TrivialLossless for + // PlainText. Each should be the only one available for its + // own type. + assert_eq!(p.lossless_by_type[&ContentType::JsonArray].len(), 1); + assert_eq!(p.lossless_by_type[&ContentType::PlainText].len(), 1); + } + + #[test] + fn builder_preserves_registration_order() { + struct A; + impl LosslessTransform for A { + fn name(&self) -> &'static str { + "a" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, content: &str) -> Result { + Ok(TransformResult::from_lengths( + content.len(), + content.into(), + true, + )) + } + } + struct B; + impl LosslessTransform for B { + fn name(&self) -> &'static str { + "b" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, content: &str) -> Result { + Ok(TransformResult::from_lengths( + content.len(), + content.into(), + true, + )) + } + } + let p = CompressionPipeline::builder() + .with_lossless(A) + .with_lossless(B) + .build(); + let order: Vec<&str> = p.lossless_by_type[&ContentType::PlainText] + .iter() + .map(|t| t.name()) + .collect(); + assert_eq!(order, vec!["a", "b"]); + } +} diff --git a/crates/headroom-core/src/transforms/pipeline/traits.rs b/crates/headroom-core/src/transforms/pipeline/traits.rs new file mode 100644 index 000000000..3ea33f89a --- /dev/null +++ b/crates/headroom-core/src/transforms/pipeline/traits.rs @@ -0,0 +1,268 @@ +//! Compression pipeline traits + supporting types. +//! +//! Two traits, one shared result type, one per-call context object, +//! one error type. That's the entire surface — every piece earns its +//! place by being needed by either an impl in this PR (JsonMinifier, +//! LineImportanceFilter) or the orchestrator that runs them. + +use crate::transforms::ContentType; + +/// Errors a transform can return. +/// +/// `InvalidInput` is for malformed input the transform can't parse +/// (e.g. JsonMinifier on non-JSON). `Skipped` is for "I ran cleanly +/// but found nothing to do" — used when an early exit is the right +/// call but isn't actually an error. `Internal` is for serializer / +/// allocator / logic-bug failures the orchestrator should record but +/// not crash on. +#[derive(Debug, thiserror::Error)] +pub enum TransformError { + /// The transform couldn't parse the input. Orchestrator skips and + /// tries the next transform. + #[error("invalid input for {transform}: {message}")] + InvalidInput { + transform: &'static str, + message: String, + }, + /// The transform decided not to run (e.g. empty input, content + /// already minimal). Orchestrator skips silently. + #[error("{transform} skipped: {message}")] + Skipped { + transform: &'static str, + message: String, + }, + /// Internal failure (serializer, allocator, logic bug). Surfaces + /// to logs but does not abort the pipeline. + #[error("{transform} internal error: {message}")] + Internal { + transform: &'static str, + message: String, + }, +} + +impl TransformError { + pub fn invalid_input(transform: &'static str, message: impl Into) -> Self { + Self::InvalidInput { + transform, + message: message.into(), + } + } + + pub fn skipped(transform: &'static str, message: impl Into) -> Self { + Self::Skipped { + transform, + message: message.into(), + } + } + + pub fn internal(transform: &'static str, message: impl Into) -> Self { + Self::Internal { + transform, + message: message.into(), + } + } +} + +/// Result of a single transform invocation. +/// +/// Every successful run produces an owned `output` string and reports +/// how much was saved. The orchestrator decides whether the savings +/// pass its acceptance threshold; transforms that didn't actually +/// shrink anything return `bytes_saved == 0` and the orchestrator +/// rejects the result. +#[derive(Debug, Clone)] +pub struct TransformResult { + /// Compressed (or pass-through) output. Always owned because + /// transforms typically produce new strings; for the rare + /// pass-through case the allocation is unavoidable but tiny. + pub output: String, + /// `original.len() - output.len()`, clamped to 0. Negative + /// "savings" (output longer than input) are normalized to 0 here + /// and the orchestrator's [`is_acceptable`] check rejects them. + /// + /// [`is_acceptable`]: super::orchestrator::CompressionPipeline::is_acceptable + pub bytes_saved: usize, + /// True iff the transform preserves structural invariants. JSON + /// minification preserves shape → true. A code-comment stripper + /// preserves the token stream's semantic meaning → true. Filtering + /// out lines a relevance scorer ranked low → false (lines are + /// gone, the LLM can't reconstruct them without CCR). + /// + /// Used by the orchestrator (and downstream PR3+ CCR-emission + /// logic) to decide whether to attach a retrieval marker. + pub structure_preserved: bool, + /// CCR cache key when the transform deliberately offloaded + /// content. `None` means the output is self-contained — no + /// retrieval handle needed. PR1 transforms never set this; the + /// field exists for PR2/PR3 wrappers around DiffCompressor / + /// LogCompressor / SmartCrusher which do emit CCR markers. + pub reversible_via: Option, +} + +impl TransformResult { + /// Construct a result whose `bytes_saved` is computed from the + /// difference between input and output lengths. + pub fn from_lengths(input_len: usize, output: String, structure_preserved: bool) -> Self { + let bytes_saved = input_len.saturating_sub(output.len()); + Self { + output, + bytes_saved, + structure_preserved, + reversible_via: None, + } + } +} + +/// Per-call context the orchestrator hands to each transform. +/// +/// Holds anything a transform might need that *isn't* the input +/// content — query for relevance scoring, target token budget for +/// aggressiveness tuning. Transforms borrow the context for the +/// duration of their `apply` call. +#[derive(Debug, Default, Clone)] +pub struct CompressionContext { + /// The user's question for the LLM. Empty string means no query + /// available; transforms must treat this as a generic compression + /// pass (no relevance bias). + pub query: String, + /// Token budget the orchestrator is targeting, in tokens (not + /// bytes). `None` means "compress as much as is safe — caller + /// will accept whatever comes out." + pub token_budget: Option, +} + +impl CompressionContext { + pub fn with_query(query: impl Into) -> Self { + Self { + query: query.into(), + token_budget: None, + } + } + + pub fn with_budget(token_budget: usize) -> Self { + Self { + query: String::new(), + token_budget: Some(token_budget), + } + } +} + +/// Compression that preserves all information (the LLM gets the same +/// bytes back, just packed denser). +/// +/// Examples (in this PR or PR2+): JSON minification, code-comment +/// stripping, log RLE, HTML extraction. These run first — if the +/// result fits the budget, no lossy work is needed. +pub trait LosslessTransform: Send + Sync { + /// Stable telemetry name, e.g. `"json_minifier"`. Used as the + /// strategy key in the per-strategy stats nest landed in + /// Phase 3e.0. Must match `^[a-z][a-z0-9_]*$` by convention so + /// it composes cleanly into JSONB keys. + fn name(&self) -> &'static str; + + /// Content types this transform accepts. The orchestrator skips + /// the transform when the input's detected type isn't in this + /// slice. Returning a `&[ContentType]` borrowed from `&self` + /// lets impls store either a static literal or an owned config- + /// driven list. + fn applies_to(&self) -> &[ContentType]; + + /// Run the transform. Returns `Err(TransformError)` for malformed + /// input or internal failures (orchestrator skips); returns + /// `Ok(TransformResult)` for successful runs even when nothing + /// was saved (the orchestrator decides what to do based on + /// `bytes_saved`). + fn apply(&self, content: &str) -> Result; +} + +/// Compression that deliberately drops content to fit a budget. +/// +/// Examples (in this PR or PR2+): line-importance filtering, prose- +/// field compression via a small model, extractive summarization. +/// These run after lossless and gate further passes on byte savings. +pub trait LossyTransform: Send + Sync { + /// Stable telemetry name (same convention as + /// [`LosslessTransform::name`]). + fn name(&self) -> &'static str; + + /// Content types this transform accepts. + fn applies_to(&self) -> &[ContentType]; + + /// Run the transform. Same error semantics as + /// [`LosslessTransform::apply`]. The `ctx` argument lets the + /// transform read the query (for relevance) and budget (for + /// aggressiveness) without taking on tokenizer or proxy state. + fn apply( + &self, + content: &str, + ctx: &CompressionContext, + ) -> Result; + + /// Calibrated 0.0–1.0 quality score. Higher = more confident the + /// output preserves enough information for the LLM to answer the + /// query. The orchestrator records this in telemetry; selection + /// between competing lossy transforms (PR4 ProseFieldCompressor + /// vs LineImportanceFilter on the same content type) becomes a + /// future extension. For now it's purely observational. + fn confidence(&self) -> f32; +} + +#[cfg(test)] +mod tests { + use super::*; + + /// One throwaway impl per trait, used by orchestrator-level tests + /// when they need a deterministic transform. Real impls live in + /// sibling modules. + pub struct AlwaysSavesNothing; + impl LosslessTransform for AlwaysSavesNothing { + fn name(&self) -> &'static str { + "test_no_op" + } + fn applies_to(&self) -> &[ContentType] { + &[ContentType::PlainText] + } + fn apply(&self, content: &str) -> Result { + Ok(TransformResult { + output: content.to_string(), + bytes_saved: 0, + structure_preserved: true, + reversible_via: None, + }) + } + } + + #[test] + fn from_lengths_clamps_negative_savings_to_zero() { + let r = TransformResult::from_lengths(10, "this is longer than 10 bytes".into(), true); + assert_eq!(r.bytes_saved, 0); + } + + #[test] + fn transform_error_messages_round_trip() { + let e = TransformError::invalid_input("json_minifier", "bad token at line 3"); + let msg = e.to_string(); + assert!(msg.contains("json_minifier")); + assert!(msg.contains("bad token at line 3")); + } + + #[test] + fn compression_context_constructors_are_clean() { + let q = CompressionContext::with_query("find errors"); + assert_eq!(q.query, "find errors"); + assert_eq!(q.token_budget, None); + + let b = CompressionContext::with_budget(2048); + assert_eq!(b.query, ""); + assert_eq!(b.token_budget, Some(2048)); + } + + #[test] + fn no_op_lossless_transform_is_a_real_impl() { + let t = AlwaysSavesNothing; + let r = t.apply("hello").expect("no-op never errors"); + assert_eq!(r.bytes_saved, 0); + assert_eq!(t.applies_to(), &[ContentType::PlainText]); + assert_eq!(t.name(), "test_no_op"); + } +}