Merge pull request #325 from chopratejas/rust-stage-3g-pr1-pipeline-traits

feat(rust): pipeline traits + orchestrator skeleton (Phase 3g PR1)

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,

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,

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<TransformResult, TransformError> {
+        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<TransformResult, TransformError> {
+        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());
+    }
+}

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<dyn LineImportanceDetector>,
+    applies_to: Vec<ContentType>,
+    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<dyn LineImportanceDetector>) -> 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<ContentType>) -> 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<TransformResult, TransformError> {
+        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<usize> = 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::<Vec<_>>()
+            .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<String> = (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 🌍"));
+    }
+}

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,
+};

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<String>,
+    /// 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<String>,
+}
+
+/// 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<ContentType, Vec<Arc<dyn LosslessTransform>>>,
+    lossy_by_type: HashMap<ContentType, Vec<Arc<dyn LossyTransform>>>,
+    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<String> = Vec::new();
+        let mut reversible: Option<String> = 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(&current) {
+                    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(&current, 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<String>,
+        reversible: &mut Option<String>,
+        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<ContentType, Vec<Arc<dyn LosslessTransform>>>,
+    lossy_by_type: HashMap<ContentType, Vec<Arc<dyn LossyTransform>>>,
+    config: Option<PipelineConfig>,
+}
+
+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<T>(mut self, transform: T) -> Self
+    where
+        T: LosslessTransform + 'static,
+    {
+        let arc: Arc<dyn LosslessTransform> = Arc::new(transform);
+        let types: Vec<ContentType> = 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<T>(mut self, transform: T) -> Self
+    where
+        T: LossyTransform + 'static,
+    {
+        let arc: Arc<dyn LossyTransform> = Arc::new(transform);
+        let types: Vec<ContentType> = 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<TransformResult, TransformError> {
+            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<TransformResult, TransformError> {
+            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<TransformResult, TransformError> {
+                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::<Vec<_>>()
+            .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<TransformResult, TransformError> {
+                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::<Vec<_>>()
+            .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<TransformResult, TransformError> {
+                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<TransformResult, TransformError> {
+                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"]);
+    }
+}

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<String>) -> Self {
+        Self::InvalidInput {
+            transform,
+            message: message.into(),
+        }
+    }
+
+    pub fn skipped(transform: &'static str, message: impl Into<String>) -> Self {
+        Self::Skipped {
+            transform,
+            message: message.into(),
+        }
+    }
+
+    pub fn internal(transform: &'static str, message: impl Into<String>) -> 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<String>,
+}
+
+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<usize>,
+}
+
+impl CompressionContext {
+    pub fn with_query(query: impl Into<String>) -> 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<TransformResult, TransformError>;
+}
+
+/// 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<TransformResult, TransformError>;
+
+    /// 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<TransformResult, TransformError> {
+            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");
+    }
+}