feat(rust): SmartCrusher PR2 — lossless-first tabular compaction

Stage 3c.2 PR2. Adds an opt-in compaction stage that runs BEFORE the
existing lossy pipeline. When configured, it tries to losslessly
re-shape arrays of objects into a recursive Compaction IR and renders
that to bytes via a pluggable Formatter trait. When not configured
(default OSS), behavior is byte-equal with the pre-PR2 path — all 17
SmartCrusher parity fixtures stay green.

# What lands

- Recursive Compaction IR (`compaction/ir.rs`): Table / Buckets /
  OpaqueRef / Untouched. CellValue can hold a nested Compaction so
  multi-level cases (stringified-JSON inside cells, heterogeneous
  arrays bucketed by discriminator, opaque blobs CCR-substituted)
  share one tree shape.

- Cell classifier (`compaction/classifier.rs`): per-cell decision —
  Scalar / JsonObject / JsonArray / StringifiedJson(parsed) /
  Opaque(kind). Conservative: in doubt, return Scalar.

- TabularCompactor (`compaction/compactor.rs`): array → IR. Handles
  uniform-nested flattening into dotted columns ("meta.region",
  "meta.tier"), stringified-JSON parsing + recursion, opaque-blob
  CCR-substitution (12-char SHA-256 prefix), and heterogeneous
  bucketing by discriminator. Falls through to a sparse Table when
  no clean discriminator exists, so we always do better than the
  lossy path for object arrays.

- Formatter trait (`compaction/formatter.rs`) + two impls:
  - JsonFormatter: structured JSON for debugging / programmatic use.
  - CsvSchemaFormatter: [N]{col:type,col:type} declaration + CSV
    rows. Steals TOON's row-count-and-shape declaration without
    adopting TOON's bespoke escaping. CSV is the format LLMs are
    strongest at — every model has seen millions of examples in
    training. >30% smaller than raw JSON serialization on tabular
    fixtures.

- Wiring (`crusher.rs`, `builder.rs`): SmartCrusher gains an optional
  compaction stage. Builder methods with_compaction(stage) and
  with_default_compaction() opt in. CrushArrayResult gets two new
  fields (compacted, compaction_kind) populated only when the stage
  runs. strategy_info becomes compaction kind when compaction won.

# Why this design

- Three-trait extension surface preserved. PR1 added Constraint /
  Observer / Scorer; PR2 adds Formatter as the fourth pluggable
  seam. Enterprise plug-ins land cleanly without forking core.

- Empty default builder rule held. SmartCrusherBuilder::new() still
  produces a no-compaction crusher. with_default_compaction() is
  the explicit OSS preset. No silent fallbacks.

- Recursive IR was the unlock. A flat table-of-scalars IR would have
  collapsed the moment a cell held nested JSON. Making
  CellValue::Nested hold another Compaction made stringified-JSON
  parsing + heterogeneous bucketing + opaque substitution all share
  one renderer pass.

- CCR substitution for opaque cells. Strings classified as
  base64/HTML/long-opaque become structured markers keyed by 12-char
  SHA-256 prefix. The full bytes round-trip via the CCR store (PyO3
  bridge owns actual storage; this PR emits the marker and computes
  the hash).

# Tests

- 60 new unit tests across IR / classifier / compactor / formatter /
  wiring (448 total in headroom-core, was 388).
- 17/17 SmartCrusher parity fixtures byte-equal — default-config
  path completely unchanged.
- 21/21 Python parity tests pass via PyO3 bridge.
- make ci-precheck green: ruff, mypy, cargo fmt/clippy/test
  (1.95.0), commitlint.

# Deferred to follow-up PRs

- ToonFormatter (small; ship after eval harness compares formats)
- Diff/code detection in cells → routes to DiffCompressor /
  CodeCompressor (coupled to ContentRouter Phase 4)
- Budget-aware row dropping (Constraint-respecting) when rendered
  size exceeds budget
- Format A/B eval harness
- ContentRouter unification (Phase 4)

Modules: crates/headroom-core/src/transforms/smart_crusher/compaction/*, builder.rs, crusher.rs, mod.rs

.claude-plugin/marketplace.json

@@ -5,14 +5,14 @@
   },
   "metadata": {
     "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
-    "version": "0.10.18"
+    "version": "0.11.0"
   },
   "plugins": [
     {
       "name": "headroom",
       "source": "./plugins/headroom-agent-hooks",
       "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
-      "version": "0.10.18",
+      "version": "0.11.0",
       "author": {
         "name": "Headroom Contributors",
         "url": "https://github.com/chopratejas/headroom"

.github/plugin/marketplace.json

@@ -5,14 +5,14 @@
   },
   "metadata": {
     "description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
-    "version": "0.10.18"
+    "version": "0.11.0"
   },
   "plugins": [
     {
       "name": "headroom",
       "source": "./plugins/headroom-agent-hooks",
       "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
-      "version": "0.10.18",
+      "version": "0.11.0",
       "author": {
         "name": "Headroom Contributors",
         "url": "https://github.com/chopratejas/headroom"

crates/headroom-core/src/transforms/smart_crusher/builder.rs

@@ -34,6 +34,7 @@ use crate::relevance::{HybridScorer, RelevanceScorer};
 use crate::transforms::anchor_selector::{AnchorConfig, AnchorSelector};
 
 use super::analyzer::SmartAnalyzer;
+use super::compaction::CompactionStage;
 use super::config::SmartCrusherConfig;
 use super::constraints::default_oss_constraints;
 use super::crusher::SmartCrusher;
@@ -47,10 +48,12 @@ pub struct SmartCrusherBuilder {
     scorer: Option<Box<dyn RelevanceScorer + Send + Sync>>,
     constraints: Vec<Box<dyn Constraint>>,
     observers: Vec<Box<dyn Observer>>,
+    compaction: Option<CompactionStage>,
 }
 
 impl SmartCrusherBuilder {
-    /// Empty builder — no scorer, no constraints, no observers.
+    /// Empty builder — no scorer, no constraints, no observers, no
+    /// compaction stage.
     pub fn new(config: SmartCrusherConfig) -> Self {
         SmartCrusherBuilder {
             config,
@@ -58,6 +61,7 @@ impl SmartCrusherBuilder {
             scorer: None,
             constraints: Vec::new(),
             observers: Vec::new(),
+            compaction: None,
         }
     }
 
@@ -117,6 +121,25 @@ impl SmartCrusherBuilder {
             .add_observer(Box::new(TracingObserver))
     }
 
+    /// Plug in a compaction stage. When set, `crush_array` runs the
+    /// stage before the lossy pipeline; if it produces a non-`Untouched`
+    /// compaction the rendered bytes are returned via
+    /// [`CrushArrayResult::compacted`]. The lossy result still fills
+    /// `items` so callers can choose either output.
+    ///
+    /// [`CrushArrayResult::compacted`]: super::crusher::CrushArrayResult::compacted
+    pub fn with_compaction(mut self, stage: CompactionStage) -> Self {
+        self.compaction = Some(stage);
+        self
+    }
+
+    /// Convenience: enable the OSS compaction preset (CSV+schema
+    /// formatter, default `CompactConfig`). Equivalent to
+    /// `with_compaction(CompactionStage::default_csv_schema())`.
+    pub fn with_default_compaction(self) -> Self {
+        self.with_compaction(CompactionStage::default_csv_schema())
+    }
+
     /// Construct the `SmartCrusher`. If `with_scorer` was not called,
     /// falls back to `HybridScorer::default()` so a builder with no
     /// other customization still produces a working crusher.
@@ -133,6 +156,7 @@ impl SmartCrusherBuilder {
             analyzer,
             self.constraints,
             self.observers,
+            self.compaction,
         )
     }
 }

crates/headroom-core/src/transforms/smart_crusher/compaction/classifier.rs

@@ -0,0 +1,283 @@
+//! Per-cell classification for the compaction pipeline.
+//!
+//! Given a JSON value, decide what kind of compaction treatment it needs.
+//! The classifier is intentionally conservative — when in doubt, return
+//! [`CellClass::Scalar`] so the cell is rendered verbatim.
+//!
+//! # Detection priorities
+//!
+//! 1. **Object / array** — pass through to caller, who decides whether to
+//!    flatten (uniform-nested) or recurse ([`CellClass::JsonObject`],
+//!    [`CellClass::JsonArray`]).
+//! 2. **Stringified-JSON** — strings that parse to a JSON object/array.
+//!    Common in tool-output payloads where one field is a serialized
+//!    sub-structure ([`CellClass::StringifiedJson`]).
+//! 3. **Opaque blob** — strings above a length threshold the classifier
+//!    couldn't otherwise place. Sub-classified into base64 / HTML /
+//!    plain long-string for telemetry ([`CellClass::Opaque`]).
+//! 4. **Scalar** — everything else, rendered verbatim.
+
+use serde_json::Value;
+
+use super::ir::OpaqueKind;
+
+/// Per-cell classification result.
+#[derive(Debug, Clone, PartialEq)]
+pub enum CellClass {
+    /// Number, bool, null, short string — render verbatim.
+    Scalar,
+    /// Cell is a JSON object. Caller decides flatten-vs-recurse based
+    /// on schema uniformity across rows.
+    JsonObject,
+    /// Cell is a JSON array. Caller may recurse with TabularCompactor.
+    JsonArray,
+    /// String that parses to a JSON object/array. The parsed value is
+    /// returned so the caller doesn't re-parse.
+    StringifiedJson(Value),
+    /// Long string the classifier judged opaque. Sub-classified for
+    /// telemetry only — all variants get CCR-substituted.
+    Opaque(OpaqueKind),
+}
+
+/// Config controlling classification thresholds.
+///
+/// Defaults are tuned for typical tool-output payloads. Override via
+/// builder if a workload has different characteristics (e.g. an API
+/// that always emits 500-char status descriptions shouldn't have those
+/// CCR-substituted).
+#[derive(Debug, Clone)]
+pub struct ClassifyConfig {
+    /// Strings strictly longer than this become candidates for opaque
+    /// classification. Default: 256 bytes.
+    pub opaque_min_bytes: usize,
+    /// Base64-alphabet ratio threshold. Strings whose chars are at
+    /// least this fraction in `[A-Za-z0-9+/=_-]` and longer than 64
+    /// bytes are tagged base64. Default: 0.95.
+    pub base64_alphabet_ratio: f64,
+    /// `<` count above which a long string is considered HTML-ish.
+    /// Default: 3.
+    pub html_min_open_brackets: usize,
+}
+
+impl Default for ClassifyConfig {
+    fn default() -> Self {
+        Self {
+            opaque_min_bytes: 256,
+            base64_alphabet_ratio: 0.95,
+            html_min_open_brackets: 3,
+        }
+    }
+}
+
+/// Classify a single cell value.
+pub fn classify_cell(value: &Value, cfg: &ClassifyConfig) -> CellClass {
+    match value {
+        Value::Object(_) => CellClass::JsonObject,
+        Value::Array(_) => CellClass::JsonArray,
+        Value::String(s) => classify_string(s, cfg),
+        _ => CellClass::Scalar,
+    }
+}
+
+fn classify_string(s: &str, cfg: &ClassifyConfig) -> CellClass {
+    // Stringified-JSON check first. Cheap fast-path: must start with
+    // `{` or `[` (after optional whitespace) — skip strings that
+    // can't possibly be JSON containers. Parsing `"123"` would
+    // technically succeed as JSON-the-number, but that's a scalar,
+    // not a recursion target.
+    let trimmed = s.trim_start();
+    if matches!(trimmed.chars().next(), Some('{') | Some('[')) {
+        if let Ok(parsed) = serde_json::from_str::<Value>(s) {
+            if matches!(parsed, Value::Object(_) | Value::Array(_)) {
+                return CellClass::StringifiedJson(parsed);
+            }
+        }
+    }
+
+    // Opaque-blob check — only for strings above the byte threshold.
+    if s.len() <= cfg.opaque_min_bytes {
+        return CellClass::Scalar;
+    }
+
+    if looks_like_base64(s, cfg.base64_alphabet_ratio) {
+        return CellClass::Opaque(OpaqueKind::Base64Blob);
+    }
+
+    if looks_like_html(s, cfg.html_min_open_brackets) {
+        return CellClass::Opaque(OpaqueKind::HtmlChunk);
+    }
+
+    CellClass::Opaque(OpaqueKind::LongString)
+}
+
+fn looks_like_base64(s: &str, ratio_threshold: f64) -> bool {
+    if s.len() < 64 {
+        return false;
+    }
+    // Disqualifying signals — these instantly rule out base64.
+    if s.contains('<') || s.contains('>') {
+        return false;
+    }
+    if s.chars().any(|c| c.is_whitespace()) {
+        return false;
+    }
+
+    let total = s.len();
+    let alphabet = s
+        .chars()
+        .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '/' | '=' | '_' | '-'))
+        .count();
+    if (alphabet as f64) / (total as f64) < ratio_threshold {
+        return false;
+    }
+
+    // Diversity filter: real base64-encoded random bytes use most of
+    // their 64-character alphabet. Strings with < 16 unique characters
+    // are almost certainly not base64 (typical false-positive: brace-
+    // wrapped repeated characters like `{xxxx...}`).
+    let mut unique = std::collections::HashSet::new();
+    for c in s.chars() {
+        unique.insert(c);
+        if unique.len() >= 16 {
+            return true;
+        }
+    }
+    false
+}
+
+fn looks_like_html(s: &str, min_open_brackets: usize) -> bool {
+    let opens = s.chars().filter(|c| *c == '<').count();
+    if opens < min_open_brackets {
+        return false;
+    }
+    // Cheap signal: opens are followed by an alpha char or `/` reasonably
+    // often. Avoids false-positives on math-heavy strings ("a < b").
+    let bytes = s.as_bytes();
+    let mut tag_starts = 0usize;
+    for (i, b) in bytes.iter().enumerate() {
+        if *b == b'<' {
+            if let Some(next) = bytes.get(i + 1) {
+                if next.is_ascii_alphabetic() || *next == b'/' || *next == b'!' {
+                    tag_starts += 1;
+                }
+            }
+        }
+    }
+    tag_starts >= min_open_brackets
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn cfg() -> ClassifyConfig {
+        ClassifyConfig::default()
+    }
+
+    #[test]
+    fn scalars_are_scalars() {
+        assert_eq!(classify_cell(&json!(1), &cfg()), CellClass::Scalar);
+        assert_eq!(classify_cell(&json!(1.5), &cfg()), CellClass::Scalar);
+        assert_eq!(classify_cell(&json!(true), &cfg()), CellClass::Scalar);
+        assert_eq!(classify_cell(&json!(null), &cfg()), CellClass::Scalar);
+        assert_eq!(classify_cell(&json!("short"), &cfg()), CellClass::Scalar);
+    }
+
+    #[test]
+    fn objects_and_arrays_pass_through() {
+        assert_eq!(
+            classify_cell(&json!({"a": 1}), &cfg()),
+            CellClass::JsonObject
+        );
+        assert_eq!(classify_cell(&json!([1, 2]), &cfg()), CellClass::JsonArray);
+    }
+
+    #[test]
+    fn stringified_json_object_is_parsed() {
+        let v = json!(r#"{"x":1,"y":2}"#);
+        match classify_cell(&v, &cfg()) {
+            CellClass::StringifiedJson(parsed) => {
+                assert_eq!(parsed, json!({"x": 1, "y": 2}));
+            }
+            other => panic!("expected StringifiedJson, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn stringified_json_array_is_parsed() {
+        let v = json!(r#"[1,2,3]"#);
+        match classify_cell(&v, &cfg()) {
+            CellClass::StringifiedJson(parsed) => {
+                assert_eq!(parsed, json!([1, 2, 3]));
+            }
+            other => panic!("expected StringifiedJson, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn stringified_scalar_is_not_recursed() {
+        // "123" parses as a JSON number, but we don't recurse on scalars.
+        assert_eq!(classify_cell(&json!("123"), &cfg()), CellClass::Scalar);
+        // Same for booleans, nulls.
+        assert_eq!(classify_cell(&json!("true"), &cfg()), CellClass::Scalar);
+    }
+
+    #[test]
+    fn malformed_brace_string_is_long_opaque_or_scalar() {
+        let short = json!("{not json}");
+        assert_eq!(classify_cell(&short, &cfg()), CellClass::Scalar);
+        let long = "{".to_string() + &"x".repeat(300) + "}";
+        match classify_cell(&Value::String(long), &cfg()) {
+            CellClass::Opaque(OpaqueKind::LongString) => {}
+            other => panic!("expected LongString, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn base64_blob_detected() {
+        let s = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/==".repeat(5);
+        match classify_cell(&Value::String(s), &cfg()) {
+            CellClass::Opaque(OpaqueKind::Base64Blob) => {}
+            other => panic!("expected Base64Blob, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn html_chunk_detected() {
+        let s = "<html><body><p>".to_string() + &"x".repeat(300) + "</p></body></html>";
+        match classify_cell(&Value::String(s), &cfg()) {
+            CellClass::Opaque(OpaqueKind::HtmlChunk) => {}
+            other => panic!("expected HtmlChunk, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn long_plain_string_is_long_opaque() {
+        let s = "the quick brown fox ".repeat(20);
+        match classify_cell(&Value::String(s), &cfg()) {
+            CellClass::Opaque(OpaqueKind::LongString) => {}
+            other => panic!("expected LongString, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn math_with_lt_is_not_html() {
+        let s = "a < b but not really ".repeat(20);
+        match classify_cell(&Value::String(s), &cfg()) {
+            CellClass::Opaque(OpaqueKind::LongString) => {}
+            other => panic!("expected LongString, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn config_threshold_respected() {
+        let mut c = cfg();
+        c.opaque_min_bytes = 10;
+        let s = json!("hello world this is long");
+        match classify_cell(&s, &c) {
+            CellClass::Opaque(_) => {}
+            other => panic!("expected Opaque under low threshold, got {other:?}"),
+        }
+    }
+}

crates/headroom-core/src/transforms/smart_crusher/compaction/compactor.rs

@@ -0,0 +1,733 @@
+//! TabularCompactor — array → [`Compaction`] IR.
+//!
+//! # Pipeline
+//!
+//! ```text
+//! &[Value]  →  detect uniformity  →  build schema  →  build rows
+//!                    │
+//!                    ├─ heterogeneous? → bucket by discriminator
+//!                    │                    (Compaction::Buckets)
+//!                    │
+//!                    └─ homogeneous → flatten nested-uniform columns
+//!                                        (Compaction::Table)
+//! ```
+//!
+//! # Decision rules
+//!
+//! - **Untouched fall-through.** Items < 2, non-object items, or a key
+//!   distribution too uneven for tabular form → return [`Compaction::Untouched`]
+//!   so the existing lossy path takes over.
+//! - **Schema = union of all keys**, sorted by descending frequency then
+//!   alphabetically. Sparse fields keep their slot — cells in rows that
+//!   lack the field render as [`CellValue::Missing`].
+//! - **Heterogeneous case.** When < 50% of keys appear in >= 80% of rows,
+//!   look for a discriminator (a string field present in every row whose
+//!   value distribution partitions cleanly). If found, emit
+//!   [`Compaction::Buckets`]; else [`Compaction::Untouched`].
+//! - **Nested-uniform flatten.** A field that's an object in every row
+//!   with the same inner key set, where flattening doesn't blow up the
+//!   column count by more than `max_flatten_inner_keys`, gets promoted
+//!   into dotted columns (`meta.region`, `meta.tier`).
+//! - **Stringified-JSON.** Cells that classify as
+//!   [`CellClass::StringifiedJson`] become [`CellValue::Nested`] when the
+//!   parsed value is an array of objects (recursive table); otherwise
+//!   [`CellValue::Scalar`] of the parsed value (saves escaping cost).
+//! - **Opaque blob.** [`CellClass::Opaque`] cells become
+//!   [`CellValue::OpaqueRef`] keyed by a 12-char SHA-256 prefix.
+//!
+//! [`CellClass`]: super::classifier::CellClass
+//! [`CellClass::StringifiedJson`]: super::classifier::CellClass::StringifiedJson
+//! [`CellClass::Opaque`]: super::classifier::CellClass::Opaque
+
+use std::collections::BTreeMap;
+
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+
+use super::classifier::{classify_cell, CellClass, ClassifyConfig};
+use super::ir::{Bucket, CellValue, Compaction, FieldSpec, Row, Schema};
+
+/// Config for the compactor.
+#[derive(Debug, Clone)]
+pub struct CompactConfig {
+    pub classify: ClassifyConfig,
+
+    /// Minimum item count to attempt tabular compaction. Below this,
+    /// return [`Compaction::Untouched`]. Default: 2.
+    pub min_items: usize,
+
+    /// A field is "core" if it appears in at least this fraction of
+    /// rows. Schemas with too few core fields trigger heterogeneous
+    /// (bucket) handling. Default: 0.8.
+    pub core_field_fraction: f64,
+
+    /// Heterogeneity threshold: when fewer than this fraction of all
+    /// observed keys are core, treat the array as heterogeneous and
+    /// look for a discriminator. Default: 0.5.
+    pub heterogeneous_core_ratio: f64,
+
+    /// Cap on inner-key count for nested-uniform flattening. Larger
+    /// inner schemas stay nested rather than exploding column count.
+    /// Default: 6.
+    pub max_flatten_inner_keys: usize,
+
+    /// Minimum bucket count before considering a candidate discriminator
+    /// "useful". Default: 2.
+    pub min_buckets: usize,
+
+    /// Maximum bucket count — too many buckets means the discriminator
+    /// is too granular (e.g. an ID column). Default: 8.
+    pub max_buckets: usize,
+}
+
+impl Default for CompactConfig {
+    fn default() -> Self {
+        Self {
+            classify: ClassifyConfig::default(),
+            min_items: 2,
+            core_field_fraction: 0.8,
+            heterogeneous_core_ratio: 0.6,
+            max_flatten_inner_keys: 6,
+            min_buckets: 2,
+            max_buckets: 8,
+        }
+    }
+}
+
+/// Top-level compaction entry point.
+pub fn compact(items: &[Value], cfg: &CompactConfig) -> Compaction {
+    if items.len() < cfg.min_items {
+        return Compaction::Untouched(Value::Array(items.to_vec()));
+    }
+    if !items.iter().all(|v| matches!(v, Value::Object(_))) {
+        return Compaction::Untouched(Value::Array(items.to_vec()));
+    }
+
+    let key_freqs = compute_key_freqs(items);
+    let total = items.len();
+    let core_threshold = (total as f64 * cfg.core_field_fraction).ceil() as usize;
+    let core_count = key_freqs.values().filter(|&&f| f >= core_threshold).count();
+    let total_keys = key_freqs.len();
+
+    let core_ratio = if total_keys == 0 {
+        1.0
+    } else {
+        core_count as f64 / total_keys as f64
+    };
+
+    if core_ratio < cfg.heterogeneous_core_ratio {
+        if let Some(disc) = detect_discriminator(items, &key_freqs, cfg) {
+            return bucket_by(items, &disc, cfg);
+        }
+        // No clean discriminator — fall through to a sparse Table
+        // rather than refusing. A sparse table is still better than
+        // letting the lossy path drop fields wholesale.
+    }
+
+    build_homogeneous_table(items, &key_freqs, cfg)
+}
+
+fn compute_key_freqs(items: &[Value]) -> BTreeMap<String, usize> {
+    let mut freqs: BTreeMap<String, usize> = BTreeMap::new();
+    for item in items {
+        if let Value::Object(map) = item {
+            for k in map.keys() {
+                *freqs.entry(k.clone()).or_insert(0) += 1;
+            }
+        }
+    }
+    freqs
+}
+
+fn build_homogeneous_table(
+    items: &[Value],
+    key_freqs: &BTreeMap<String, usize>,
+    cfg: &CompactConfig,
+) -> Compaction {
+    // Order: descending frequency, then alphabetical for stability.
+    let mut keys: Vec<(&String, &usize)> = key_freqs.iter().collect();
+    keys.sort_by(|a, b| b.1.cmp(a.1).then_with(|| a.0.cmp(b.0)));
+    let ordered_keys: Vec<String> = keys.into_iter().map(|(k, _)| k.clone()).collect();
+
+    let total = items.len();
+    let mut field_specs: Vec<FieldSpec> = ordered_keys
+        .iter()
+        .map(|k| FieldSpec {
+            name: k.clone(),
+            type_tag: infer_type_tag(items, k),
+            nullable: key_freqs[k] < total
+                || items
+                    .iter()
+                    .filter_map(|v| v.as_object())
+                    .any(|o| matches!(o.get(k), Some(Value::Null))),
+        })
+        .collect();
+
+    let mut rows: Vec<Row> = items
+        .iter()
+        .map(|item| build_row(item, &ordered_keys, cfg))
+        .collect();
+
+    flatten_uniform_nested(&mut field_specs, &mut rows, cfg);
+
+    Compaction::Table {
+        schema: Schema {
+            fields: field_specs,
+        },
+        rows,
+        original_count: items.len(),
+    }
+}
+
+fn build_row(item: &Value, ordered_keys: &[String], cfg: &CompactConfig) -> Row {
+    let obj = match item.as_object() {
+        Some(o) => o,
+        None => return Row::new(vec![]),
+    };
+    let cells: Vec<CellValue> = ordered_keys
+        .iter()
+        .map(|k| match obj.get(k) {
+            None => CellValue::Missing,
+            Some(v) => cell_from_value(v, cfg),
+        })
+        .collect();
+    Row::new(cells)
+}
+
+fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue {
+    match classify_cell(v, &cfg.classify) {
+        CellClass::Scalar => CellValue::Scalar(v.clone()),
+        CellClass::JsonObject => CellValue::Scalar(v.clone()), // flatten pass may promote
+        CellClass::JsonArray => {
+            // Recurse if the inner array is array-of-objects; else scalar.
+            if let Value::Array(items) = v {
+                if items.iter().all(|i| matches!(i, Value::Object(_))) && items.len() >= 2 {
+                    return CellValue::Nested(Box::new(compact(items, cfg)));
+                }
+            }
+            CellValue::Scalar(v.clone())
+        }
+        CellClass::StringifiedJson(parsed) => {
+            // If the parsed JSON is an array of objects, recurse; else
+            // store the parsed value as a Scalar (un-escapes for free).
+            if let Value::Array(items) = &parsed {
+                if items.iter().all(|i| matches!(i, Value::Object(_))) && items.len() >= 2 {
+                    return CellValue::Nested(Box::new(compact(items, cfg)));
+                }
+            }
+            CellValue::Scalar(parsed)
+        }
+        CellClass::Opaque(kind) => {
+            let bytes = match v {
+                Value::String(s) => s.as_bytes(),
+                _ => return CellValue::Scalar(v.clone()),
+            };
+            CellValue::OpaqueRef {
+                ccr_hash: hash_opaque(bytes),
+                byte_size: bytes.len(),
+                kind,
+            }
+        }
+    }
+}
+
+/// Promote fields whose every row holds an object with the same key
+/// set into dotted columns. Bounded by `cfg.max_flatten_inner_keys` so
+/// a 50-key inner schema doesn't blow up the table width.
+fn flatten_uniform_nested(specs: &mut Vec<FieldSpec>, rows: &mut [Row], cfg: &CompactConfig) {
+    let mut i = 0;
+    while i < specs.len() {
+        let inner_keys = match uniform_object_keys(specs, rows, i) {
+            Some(keys) if !keys.is_empty() && keys.len() <= cfg.max_flatten_inner_keys => keys,
+            _ => {
+                i += 1;
+                continue;
+            }
+        };
+
+        let parent_name = specs[i].name.clone();
+        let new_specs: Vec<FieldSpec> = inner_keys
+            .iter()
+            .map(|k| FieldSpec {
+                name: format!("{parent_name}.{k}"),
+                type_tag: "string".into(),
+                nullable: false,
+            })
+            .collect();
+        let n_new = new_specs.len();
+
+        // Splice into specs: replace specs[i] with new_specs.
+        specs.splice(i..i + 1, new_specs);
+
+        // Rewrite each row: replace row.0[i] with N expanded cells.
+        for row in rows.iter_mut() {
+            let original = row.0.remove(i);
+            let inner_obj: Option<serde_json::Map<String, Value>> = match original {
+                CellValue::Scalar(Value::Object(map)) => Some(map),
+                CellValue::Missing => None,
+                _ => unreachable!(
+                    "uniform_object_keys guarantees every cell is Scalar(Object) or Missing"
+                ),
+            };
+            let expanded: Vec<CellValue> = inner_keys
+                .iter()
+                .map(|k| match &inner_obj {
+                    None => CellValue::Missing,
+                    Some(map) => match map.get(k) {
+                        None => CellValue::Missing,
+                        Some(v) => CellValue::Scalar(v.clone()),
+                    },
+                })
+                .collect();
+            for (offset, cell) in expanded.into_iter().enumerate() {
+                row.0.insert(i + offset, cell);
+            }
+        }
+
+        // Refine type tags + nullability from data.
+        for offset in 0..n_new {
+            let col_idx = i + offset;
+            let mut nullable = false;
+            let inferred = infer_type_tag_from_cells(rows, col_idx, &mut nullable);
+            specs[col_idx].type_tag = inferred;
+            specs[col_idx].nullable = nullable;
+        }
+
+        i += n_new;
+    }
+}
+
+fn infer_type_tag_from_cells(rows: &[Row], col: usize, nullable: &mut bool) -> String {
+    let mut tag = "string";
+    let mut saw_value = false;
+    for row in rows {
+        if let Some(cell) = row.0.get(col) {
+            match cell {
+                CellValue::Missing => *nullable = true,
+                CellValue::Scalar(Value::Null) => *nullable = true,
+                CellValue::Scalar(v) => {
+                    if !saw_value {
+                        tag = type_tag_for(v);
+                        saw_value = true;
+                    } else if type_tag_for(v) != tag {
+                        tag = "json";
+                    }
+                }
+                _ => tag = "json",
+            }
+        }
+    }
+    tag.to_string()
+}
+
+/// Returns Some(inner_keys) if every row's cell at `col` is an object
+/// with the same key set (or Missing). None otherwise.
+fn uniform_object_keys(specs: &[FieldSpec], rows: &[Row], col: usize) -> Option<Vec<String>> {
+    if specs[col].name.contains('.') {
+        // Already a flattened column.
+        return None;
+    }
+    let mut canonical: Option<Vec<String>> = None;
+    let mut saw_object = false;
+    for row in rows {
+        let cell = row.0.get(col)?;
+        match cell {
+            CellValue::Missing => continue,
+            CellValue::Scalar(Value::Object(map)) => {
+                let keys: Vec<String> = map.keys().cloned().collect();
+                saw_object = true;
+                match &canonical {
+                    None => canonical = Some(keys),
+                    Some(existing) => {
+                        if existing != &keys {
+                            return None;
+                        }
+                    }
+                }
+            }
+            _ => return None,
+        }
+    }
+    if !saw_object {
+        return None;
+    }
+    canonical
+}
+
+fn infer_type_tag(items: &[Value], key: &str) -> String {
+    let mut tag: Option<&'static str> = None;
+    for it in items {
+        if let Some(v) = it.as_object().and_then(|m| m.get(key)) {
+            if matches!(v, Value::Null) {
+                continue;
+            }
+            let t = type_tag_for(v);
+            match tag {
+                None => tag = Some(t),
+                Some(existing) if existing != t => {
+                    tag = Some("json");
+                    break;
+                }
+                _ => {}
+            }
+        }
+    }
+    tag.unwrap_or("string").to_string()
+}
+
+fn type_tag_for(v: &Value) -> &'static str {
+    match v {
+        Value::Null => "null",
+        Value::Bool(_) => "bool",
+        Value::Number(n) if n.is_i64() || n.is_u64() => "int",
+        Value::Number(_) => "float",
+        Value::String(_) => "string",
+        Value::Object(_) | Value::Array(_) => "json",
+    }
+}
+
+fn hash_opaque(bytes: &[u8]) -> String {
+    let mut h = Sha256::new();
+    h.update(bytes);
+    let digest = h.finalize();
+    // 12-char hex prefix — collision-resistant enough for a single
+    // payload in flight, short enough to keep the marker compact.
+    let hex: String = digest.iter().take(6).map(|b| format!("{b:02x}")).collect();
+    hex
+}
+
+// ─────────────────────────── heterogeneous bucketing ───────────────────────────
+
+/// Find a discriminator field — string-typed, present in every row,
+/// with a value distribution that partitions cleanly into 2..=max_buckets
+/// non-trivial buckets.
+fn detect_discriminator(
+    items: &[Value],
+    key_freqs: &BTreeMap<String, usize>,
+    cfg: &CompactConfig,
+) -> Option<String> {
+    let total = items.len();
+    let mut best: Option<(String, usize)> = None; // (key, bucket_count)
+
+    for (k, &freq) in key_freqs {
+        if freq < total {
+            continue; // must be present in every row
+        }
+        // Collect values; require all strings.
+        let mut values: Vec<&str> = Vec::with_capacity(total);
+        let mut all_strings = true;
+        for item in items {
+            match item.as_object().and_then(|m| m.get(k)) {
+                Some(Value::String(s)) => values.push(s.as_str()),
+                _ => {
+                    all_strings = false;
+                    break;
+                }
+            }
+        }
+        if !all_strings {
+            continue;
+        }
+        let mut distinct: std::collections::HashSet<&str> = std::collections::HashSet::new();
+        for v in &values {
+            distinct.insert(*v);
+        }
+        let n = distinct.len();
+        if n < cfg.min_buckets || n > cfg.max_buckets {
+            continue;
+        }
+        // Reject discriminators that are essentially unique (1 row per
+        // bucket — that's an ID, not a category).
+        if n as f64 / total as f64 > 0.7 {
+            continue;
+        }
+        let score = n; // prefer more buckets up to max
+        match &best {
+            None => best = Some((k.clone(), score)),
+            Some((_, s)) if score > *s => best = Some((k.clone(), score)),
+            _ => {}
+        }
+    }
+    best.map(|(k, _)| k)
+}
+
+fn bucket_by(items: &[Value], discriminator: &str, cfg: &CompactConfig) -> Compaction {
+    let mut groups: BTreeMap<String, Vec<Value>> = BTreeMap::new();
+    for item in items {
+        let key = item
+            .as_object()
+            .and_then(|m| m.get(discriminator))
+            .and_then(|v| v.as_str())
+            .unwrap_or("__missing__")
+            .to_string();
+        groups.entry(key).or_default().push(item.clone());
+    }
+    let buckets: Vec<Bucket> = groups
+        .into_iter()
+        .map(|(key, group_items)| {
+            let inner = compact(&group_items, cfg);
+            match inner {
+                Compaction::Table { schema, rows, .. } => Bucket {
+                    key: Value::String(key),
+                    schema,
+                    rows,
+                },
+                _ => {
+                    // Sub-compaction declined — fall back to a degenerate
+                    // single-column "value" table holding the raw items.
+                    Bucket {
+                        key: Value::String(key),
+                        schema: Schema {
+                            fields: vec![FieldSpec {
+                                name: "value".into(),
+                                type_tag: "json".into(),
+                                nullable: false,
+                            }],
+                        },
+                        rows: group_items
+                            .into_iter()
+                            .map(|v| Row::new(vec![CellValue::Scalar(v)]))
+                            .collect(),
+                    }
+                }
+            }
+        })
+        .collect();
+    Compaction::Buckets {
+        discriminator: discriminator.to_string(),
+        buckets,
+        original_count: items.len(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::ir::OpaqueKind;
+    use super::*;
+    use serde_json::json;
+
+    fn cfg() -> CompactConfig {
+        CompactConfig::default()
+    }
+
+    #[test]
+    fn empty_or_single_is_untouched() {
+        let items: Vec<Value> = vec![];
+        assert!(matches!(compact(&items, &cfg()), Compaction::Untouched(_)));
+        let items = vec![json!({"a": 1})];
+        assert!(matches!(compact(&items, &cfg()), Compaction::Untouched(_)));
+    }
+
+    #[test]
+    fn non_object_array_is_untouched() {
+        let items = vec![json!(1), json!(2), json!(3)];
+        assert!(matches!(compact(&items, &cfg()), Compaction::Untouched(_)));
+    }
+
+    #[test]
+    fn pure_tabular_produces_table() {
+        let items = vec![
+            json!({"id": 1, "name": "alice", "status": "ok"}),
+            json!({"id": 2, "name": "bob", "status": "ok"}),
+            json!({"id": 3, "name": "carol", "status": "fail"}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table {
+                schema,
+                rows,
+                original_count,
+            } => {
+                assert_eq!(original_count, 3);
+                assert_eq!(rows.len(), 3);
+                let names = schema.field_names();
+                assert!(names.contains(&"id"));
+                assert!(names.contains(&"name"));
+                assert!(names.contains(&"status"));
+                // Type inference
+                let id_spec = schema.fields.iter().find(|f| f.name == "id").unwrap();
+                assert_eq!(id_spec.type_tag, "int");
+            }
+            other => panic!("expected Table, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn nested_uniform_is_flattened() {
+        let items = vec![
+            json!({"id": 1, "meta": {"region": "us", "tier": "gold"}}),
+            json!({"id": 2, "meta": {"region": "eu", "tier": "silver"}}),
+            json!({"id": 3, "meta": {"region": "us", "tier": "bronze"}}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { schema, rows, .. } => {
+                let names = schema.field_names();
+                assert!(names.contains(&"meta.region"), "got {names:?}");
+                assert!(names.contains(&"meta.tier"), "got {names:?}");
+                assert!(!names.contains(&"meta"));
+                assert_eq!(rows[0].len(), schema.fields.len());
+            }
+            other => panic!("expected Table, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn nested_mixed_keys_stay_nested() {
+        let items = vec![
+            json!({"id": 1, "meta": {"region": "us"}}),
+            json!({"id": 2, "meta": {"region": "eu", "tier": "silver"}}),
+            json!({"id": 3, "meta": {"tier": "bronze"}}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { schema, .. } => {
+                let names = schema.field_names();
+                // No flatten — all-different key sets per row
+                assert!(names.contains(&"meta"));
+                assert!(!names.iter().any(|n| n.starts_with("meta.")));
+            }
+            other => panic!("expected Table, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn stringified_json_array_recurses() {
+        let items = vec![
+            json!({"event": "batch", "payload": r#"[{"x":1},{"x":2},{"x":3}]"#}),
+            json!({"event": "batch", "payload": r#"[{"x":4},{"x":5}]"#}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { rows, .. } => {
+                // payload column should be Nested(Compaction::Table).
+                let payload_idx = 1; // depends on order; check both
+                let cell0 = &rows[0].0[0];
+                let cell1 = &rows[0].0[1];
+                let nested_count = [cell0, cell1]
+                    .iter()
+                    .filter(|c| matches!(***c, CellValue::Nested(_)))
+                    .count();
+                let _ = payload_idx;
+                assert_eq!(nested_count, 1, "expected exactly one Nested cell");
+            }
+            other => panic!("expected Table, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn opaque_cell_becomes_ccr_ref() {
+        let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8);
+        let items = vec![
+            json!({"id": 1, "blob": big.clone()}),
+            json!({"id": 2, "blob": big.clone()}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { rows, schema, .. } => {
+                let blob_idx = schema
+                    .fields
+                    .iter()
+                    .position(|f| f.name == "blob")
+                    .expect("blob col");
+                match &rows[0].0[blob_idx] {
+                    CellValue::OpaqueRef {
+                        ccr_hash,
+                        byte_size,
+                        kind,
+                    } => {
+                        assert!(!ccr_hash.is_empty());
+                        assert_eq!(*byte_size, big.len());
+                        assert_eq!(*kind, OpaqueKind::Base64Blob);
+                    }
+                    other => panic!("expected OpaqueRef, got {other:?}"),
+                }
+            }
+            other => panic!("expected Table, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn heterogeneous_array_buckets_by_discriminator() {
+        let items = vec![
+            json!({"type": "user", "id": 1, "name": "alice"}),
+            json!({"type": "user", "id": 2, "name": "bob"}),
+            json!({"type": "order", "id": 99, "total": 50}),
+            json!({"type": "order", "id": 100, "total": 75}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Buckets {
+                discriminator,
+                buckets,
+                original_count,
+            } => {
+                assert_eq!(discriminator, "type");
+                assert_eq!(buckets.len(), 2);
+                assert_eq!(original_count, 4);
+                let total_rows: usize = buckets.iter().map(|b| b.rows.len()).sum();
+                assert_eq!(total_rows, 4);
+            }
+            other => panic!("expected Buckets, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn id_like_field_not_chosen_as_discriminator() {
+        // Every "id" is unique → reject as discriminator.
+        let items = vec![
+            json!({"id": "a1", "kind": "x"}),
+            json!({"id": "a2", "kind": "x"}),
+            json!({"id": "a3", "kind": "y"}),
+            json!({"id": "a4", "kind": "y"}),
+        ];
+        // Schema is well-defined (homogeneous) so we won't even enter
+        // the discriminator path. But verify directly.
+        let mut freqs = BTreeMap::new();
+        freqs.insert("id".to_string(), 4);
+        freqs.insert("kind".to_string(), 4);
+        let disc = detect_discriminator(&items, &freqs, &cfg());
+        assert_eq!(disc.as_deref(), Some("kind"));
+    }
+
+    #[test]
+    fn stable_field_ordering() {
+        // Frequency descending then alphabetical.
+        let items = vec![
+            json!({"common": 1, "z_rare": 1}),
+            json!({"common": 2, "a_rare": 1}),
+            json!({"common": 3}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { schema, .. } => {
+                assert_eq!(schema.fields[0].name, "common");
+                // Two rare fields with same freq: alphabetical
+                assert_eq!(schema.fields[1].name, "a_rare");
+                assert_eq!(schema.fields[2].name, "z_rare");
+            }
+            other => panic!("got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn nullable_field_marked() {
+        let items = vec![
+            json!({"id": 1, "tag": "a"}),
+            json!({"id": 2}),
+            json!({"id": 3, "tag": null}),
+        ];
+        match compact(&items, &cfg()) {
+            Compaction::Table { schema, .. } => {
+                let tag = schema.fields.iter().find(|f| f.name == "tag").unwrap();
+                assert!(tag.nullable);
+                let id = schema.fields.iter().find(|f| f.name == "id").unwrap();
+                assert!(!id.nullable);
+            }
+            other => panic!("got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn hash_opaque_stable_and_short() {
+        let h1 = hash_opaque(b"hello world");
+        let h2 = hash_opaque(b"hello world");
+        let h3 = hash_opaque(b"different");
+        assert_eq!(h1, h2);
+        assert_ne!(h1, h3);
+        assert_eq!(h1.len(), 12);
+    }
+}

crates/headroom-core/src/transforms/smart_crusher/compaction/formatter.rs

@@ -0,0 +1,594 @@
+//! Formatter trait + two built-in implementations.
+//!
+//! [`Formatter`] walks a [`Compaction`] tree and renders bytes. It's the
+//! pluggable seam where users (or Enterprise plugins) choose how the
+//! compacted output looks.
+//!
+//! # Built-ins
+//!
+//! - [`JsonFormatter`] — single-line / pretty JSON. Easy to parse,
+//!   wider model familiarity, larger byte size. Default for the
+//!   debugging path.
+//! - [`CsvSchemaFormatter`] — `[N]{cols}` row-count-and-shape
+//!   declaration + typed column header + CSV-escaped rows. Steals
+//!   TOON's most useful idea (the `[N]{cols}` declaration) without
+//!   adopting TOON's bespoke escaping rules — every model has seen
+//!   millions of CSV examples in training.
+//!
+//! # Nested cells
+//!
+//! Both formatters handle [`CellValue::Nested`] by recursively
+//! formatting the sub-compaction and embedding the result. The CSV
+//! formatter wraps nested output in CSV-quoted form; the JSON
+//! formatter embeds it as a structured JSON object.
+//!
+//! # Opaque cells
+//!
+//! [`CellValue::OpaqueRef`] renders as a structured marker the model
+//! can recognize: `<<ccr:HASH,KIND,SIZE>>`. This format is fixed across
+//! both built-in formatters so downstream consumers can pattern-match
+//! markers regardless of which formatter produced them.
+
+use serde_json::{json, Value};
+
+use super::ir::{CellValue, Compaction, OpaqueKind, Row, Schema};
+
+/// Format a `Compaction` tree into bytes.
+pub trait Formatter: Send + Sync {
+    /// Stable name for telemetry (e.g. `"json"`, `"csv-schema"`).
+    fn name(&self) -> &str;
+
+    /// Render the compaction. Implementations should be deterministic
+    /// for stable test parity.
+    fn format(&self, c: &Compaction) -> String;
+
+    /// Cheap byte-size estimate. Default impl renders and measures.
+    /// Override for cases where rendering is expensive.
+    fn estimate_bytes(&self, c: &Compaction) -> usize {
+        self.format(c).len()
+    }
+}
+
+// ─────────────────────────── JSON formatter ───────────────────────────
+
+/// Renders a `Compaction` as structured JSON. Single-line by default
+/// for token-tight output; set `pretty = true` for human inspection.
+#[derive(Debug, Clone, Default)]
+pub struct JsonFormatter {
+    pub pretty: bool,
+}
+
+impl JsonFormatter {
+    pub fn new() -> Self {
+        Self::default()
+    }
+    pub fn pretty(mut self) -> Self {
+        self.pretty = true;
+        self
+    }
+}
+
+impl Formatter for JsonFormatter {
+    fn name(&self) -> &str {
+        "json"
+    }
+
+    fn format(&self, c: &Compaction) -> String {
+        let v = compaction_to_json(c);
+        if self.pretty {
+            serde_json::to_string_pretty(&v).unwrap_or_default()
+        } else {
+            serde_json::to_string(&v).unwrap_or_default()
+        }
+    }
+}
+
+fn compaction_to_json(c: &Compaction) -> Value {
+    match c {
+        Compaction::Table {
+            schema,
+            rows,
+            original_count,
+        } => json!({
+            "_compaction": "table",
+            "_schema": schema_to_json(schema),
+            "_kept": rows.len(),
+            "_total": original_count,
+            "_rows": rows.iter().map(row_to_json).collect::<Vec<_>>(),
+        }),
+        Compaction::Buckets {
+            discriminator,
+            buckets,
+            original_count,
+        } => json!({
+            "_compaction": "buckets",
+            "_discriminator": discriminator,
+            "_total": original_count,
+            "_buckets": buckets
+                .iter()
+                .map(|b| json!({
+                    "_key": b.key.clone(),
+                    "_schema": schema_to_json(&b.schema),
+                    "_rows": b.rows.iter().map(row_to_json).collect::<Vec<_>>(),
+                }))
+                .collect::<Vec<_>>(),
+        }),
+        Compaction::OpaqueRef {
+            ccr_hash,
+            byte_size,
+            kind,
+        } => json!({
+            "_compaction": "ccr",
+            "_hash": ccr_hash,
+            "_size": byte_size,
+            "_kind": opaque_kind_str(kind),
+        }),
+        Compaction::Untouched(v) => v.clone(),
+    }
+}
+
+fn schema_to_json(s: &Schema) -> Value {
+    Value::Array(
+        s.fields
+            .iter()
+            .map(|f| {
+                let mut obj = serde_json::Map::new();
+                obj.insert("name".into(), Value::String(f.name.clone()));
+                obj.insert("type".into(), Value::String(f.type_tag.clone()));
+                if f.nullable {
+                    obj.insert("nullable".into(), Value::Bool(true));
+                }
+                Value::Object(obj)
+            })
+            .collect(),
+    )
+}
+
+fn row_to_json(row: &Row) -> Value {
+    Value::Array(row.0.iter().map(cell_to_json).collect())
+}
+
+fn cell_to_json(c: &CellValue) -> Value {
+    match c {
+        CellValue::Scalar(v) => v.clone(),
+        CellValue::Missing => Value::Null,
+        CellValue::Nested(sub) => compaction_to_json(sub),
+        CellValue::OpaqueRef {
+            ccr_hash,
+            byte_size,
+            kind,
+        } => json!({
+            "_ccr": ccr_hash,
+            "_size": byte_size,
+            "_kind": opaque_kind_str(kind),
+        }),
+    }
+}
+
+fn opaque_kind_str(k: &OpaqueKind) -> String {
+    match k {
+        OpaqueKind::Base64Blob => "base64".into(),
+        OpaqueKind::LongString => "string".into(),
+        OpaqueKind::HtmlChunk => "html".into(),
+        OpaqueKind::Other(s) => s.clone(),
+    }
+}
+
+// ─────────────────────────── CSV+schema formatter ───────────────────────────
+
+/// Renders a `Compaction` as `[N]{col1:type1,col2:type2}` declaration +
+/// CSV-escaped rows. Nested cells render as JSON inline; opaque cells
+/// render as `<<ccr:...>>` markers.
+#[derive(Debug, Clone, Default)]
+pub struct CsvSchemaFormatter {
+    /// If true, emit a `__total:N` line when rows were dropped under
+    /// budget. Costs a few bytes; useful for downstream telemetry.
+    pub include_drop_summary: bool,
+}
+
+impl CsvSchemaFormatter {
+    pub fn new() -> Self {
+        Self::default()
+    }
+    pub fn with_drop_summary(mut self) -> Self {
+        self.include_drop_summary = true;
+        self
+    }
+}
+
+impl Formatter for CsvSchemaFormatter {
+    fn name(&self) -> &str {
+        "csv-schema"
+    }
+
+    fn format(&self, c: &Compaction) -> String {
+        let mut out = String::new();
+        write_compaction(&mut out, c, self);
+        out
+    }
+}
+
+fn write_compaction(out: &mut String, c: &Compaction, fmt: &CsvSchemaFormatter) {
+    match c {
+        Compaction::Table {
+            schema,
+            rows,
+            original_count,
+        } => {
+            write_table(out, schema, rows, *original_count, fmt);
+        }
+        Compaction::Buckets {
+            discriminator,
+            buckets,
+            original_count,
+        } => {
+            out.push_str("__buckets:");
+            out.push_str(discriminator);
+            if fmt.include_drop_summary {
+                let kept: usize = buckets.iter().map(|b| b.rows.len()).sum();
+                if kept < *original_count {
+                    out.push_str(&format!(" __dropped:{}", original_count - kept));
+                }
+            }
+            out.push('\n');
+            for b in buckets {
+                out.push_str(&format!("__key:{}\n", json_scalar_to_csv(&b.key)));
+                write_table(out, &b.schema, &b.rows, b.rows.len(), fmt);
+            }
+        }
+        Compaction::OpaqueRef {
+            ccr_hash,
+            byte_size,
+            kind,
+        } => {
+            out.push_str(&format_ccr_marker(ccr_hash, *byte_size, kind));
+        }
+        Compaction::Untouched(v) => {
+            out.push_str(&serde_json::to_string(v).unwrap_or_default());
+        }
+    }
+}
+
+fn write_table(
+    out: &mut String,
+    schema: &Schema,
+    rows: &[Row],
+    original_count: usize,
+    fmt: &CsvSchemaFormatter,
+) {
+    // Declaration line: [N]{col:type,col:type,...}
+    out.push('[');
+    out.push_str(&rows.len().to_string());
+    out.push_str("]{");
+    let col_decl: Vec<String> = schema
+        .fields
+        .iter()
+        .map(|f| {
+            if f.nullable {
+                format!("{}:{}?", f.name, f.type_tag)
+            } else {
+                format!("{}:{}", f.name, f.type_tag)
+            }
+        })
+        .collect();
+    out.push_str(&col_decl.join(","));
+    out.push('}');
+    if fmt.include_drop_summary && rows.len() < original_count {
+        out.push_str(&format!(" __dropped:{}", original_count - rows.len()));
+    }
+    out.push('\n');
+
+    // Rows.
+    for row in rows {
+        let cells: Vec<String> = row.0.iter().map(format_cell).collect();
+        out.push_str(&cells.join(","));
+        out.push('\n');
+    }
+}
+
+fn format_cell(c: &CellValue) -> String {
+    match c {
+        CellValue::Missing => String::new(),
+        CellValue::Scalar(v) => json_scalar_to_csv(v),
+        CellValue::Nested(sub) => {
+            // Render nested as compact JSON; CSV-quote because it
+            // contains commas and structural chars.
+            let nested_fmt = JsonFormatter::new();
+            csv_quote(&nested_fmt.format(sub))
+        }
+        CellValue::OpaqueRef {
+            ccr_hash,
+            byte_size,
+            kind,
+        } => format_ccr_marker(ccr_hash, *byte_size, kind),
+    }
+}
+
+fn format_ccr_marker(hash: &str, byte_size: usize, kind: &OpaqueKind) -> String {
+    let kind_str = match kind {
+        OpaqueKind::Base64Blob => "base64",
+        OpaqueKind::LongString => "string",
+        OpaqueKind::HtmlChunk => "html",
+        OpaqueKind::Other(s) => s.as_str(),
+    };
+    format!(
+        "<<ccr:{},{},{}>>",
+        hash,
+        kind_str,
+        humanize_bytes(byte_size)
+    )
+}
+
+fn humanize_bytes(n: usize) -> String {
+    if n < 1024 {
+        return format!("{n}B");
+    }
+    let kb = n as f64 / 1024.0;
+    if kb < 1024.0 {
+        return format!("{kb:.1}KB");
+    }
+    let mb = kb / 1024.0;
+    format!("{mb:.1}MB")
+}
+
+fn json_scalar_to_csv(v: &Value) -> String {
+    match v {
+        Value::Null => String::new(),
+        Value::Bool(b) => if *b { "true" } else { "false" }.to_string(),
+        Value::Number(n) => n.to_string(),
+        Value::String(s) => {
+            if needs_csv_quote(s) {
+                csv_quote(s)
+            } else {
+                s.clone()
+            }
+        }
+        // Object/array fall back to JSON-quoted (rare — usually
+        // already promoted to Nested by the compactor).
+        _ => csv_quote(&serde_json::to_string(v).unwrap_or_default()),
+    }
+}
+
+fn needs_csv_quote(s: &str) -> bool {
+    s.contains(',') || s.contains('"') || s.contains('\n') || s.contains('\r')
+}
+
+fn csv_quote(s: &str) -> String {
+    let mut out = String::with_capacity(s.len() + 2);
+    out.push('"');
+    for c in s.chars() {
+        if c == '"' {
+            out.push('"');
+            out.push('"');
+        } else {
+            out.push(c);
+        }
+    }
+    out.push('"');
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::transforms::smart_crusher::compaction::compactor::compact;
+    use crate::transforms::smart_crusher::compaction::compactor::CompactConfig;
+    use serde_json::json;
+
+    fn cfg() -> CompactConfig {
+        CompactConfig::default()
+    }
+
+    // ── JsonFormatter ──
+
+    #[test]
+    fn json_formatter_renders_table() {
+        let items = vec![
+            json!({"id": 1, "name": "alice"}),
+            json!({"id": 2, "name": "bob"}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = JsonFormatter::new().format(&c);
+        assert!(out.contains("\"_compaction\":\"table\""), "got: {out}");
+        assert!(out.contains("\"_kept\":2"));
+        assert!(out.contains("alice"));
+    }
+
+    #[test]
+    fn json_formatter_renders_untouched_verbatim() {
+        let c = Compaction::Untouched(json!({"a": 1, "b": [2, 3]}));
+        let out = JsonFormatter::new().format(&c);
+        assert_eq!(out, r#"{"a":1,"b":[2,3]}"#);
+    }
+
+    #[test]
+    fn json_formatter_renders_opaque_ref_marker() {
+        let mut row = Row::new(vec![CellValue::OpaqueRef {
+            ccr_hash: "abc123def456".into(),
+            byte_size: 2048,
+            kind: OpaqueKind::Base64Blob,
+        }]);
+        let c = Compaction::Table {
+            schema: Schema {
+                fields: vec![super::super::ir::FieldSpec {
+                    name: "blob".into(),
+                    type_tag: "ccr".into(),
+                    nullable: false,
+                }],
+            },
+            rows: vec![std::mem::replace(&mut row, Row::new(vec![]))],
+            original_count: 1,
+        };
+        let out = JsonFormatter::new().format(&c);
+        assert!(out.contains("\"_ccr\":\"abc123def456\""));
+        assert!(out.contains("base64"));
+    }
+
+    // ── CsvSchemaFormatter ──
+
+    #[test]
+    fn csv_formatter_pure_tabular() {
+        let items = vec![
+            json!({"id": 1, "name": "alice", "status": "ok"}),
+            json!({"id": 2, "name": "bob", "status": "ok"}),
+            json!({"id": 3, "name": "carol", "status": "fail"}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        let lines: Vec<&str> = out.trim_end().lines().collect();
+        // First line: declaration with [3]{...}
+        assert!(lines[0].starts_with("[3]{"), "got line[0]: {}", lines[0]);
+        assert!(lines[0].contains("id:int"));
+        assert!(lines[0].contains("name:string"));
+        assert!(lines[0].contains("status:string"));
+        assert_eq!(lines.len(), 4);
+    }
+
+    #[test]
+    fn csv_formatter_quotes_strings_with_commas() {
+        let items = vec![
+            json!({"id": 1, "name": "alice, the great"}),
+            json!({"id": 2, "name": "bob"}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        assert!(out.contains(r#""alice, the great""#));
+    }
+
+    #[test]
+    fn csv_formatter_escapes_internal_quotes() {
+        let items = vec![
+            json!({"id": 1, "msg": "she said \"hi\""}),
+            json!({"id": 2, "msg": "ok"}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        assert!(out.contains(r#""she said ""hi""""#));
+    }
+
+    #[test]
+    fn csv_formatter_renders_buckets() {
+        let items = vec![
+            json!({"type": "user", "id": 1, "name": "alice"}),
+            json!({"type": "user", "id": 2, "name": "bob"}),
+            json!({"type": "order", "id": 99, "total": 50}),
+            json!({"type": "order", "id": 100, "total": 75}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        assert!(out.starts_with("__buckets:type"));
+        assert!(out.contains("__key:order"));
+        assert!(out.contains("__key:user"));
+    }
+
+    #[test]
+    fn csv_formatter_emits_ccr_marker() {
+        let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8);
+        let items = vec![
+            json!({"id": 1, "blob": big.clone()}),
+            json!({"id": 2, "blob": big.clone()}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        assert!(out.contains("<<ccr:"), "got: {out}");
+        assert!(out.contains(",base64,"));
+    }
+
+    #[test]
+    fn csv_formatter_nested_cell_inline_json() {
+        let items = vec![
+            json!({"event": "batch", "payload": r#"[{"x":1},{"x":2},{"x":3}]"#}),
+            json!({"event": "batch", "payload": r#"[{"x":4},{"x":5}]"#}),
+        ];
+        let c = compact(&items, &cfg());
+        let out = CsvSchemaFormatter::new().format(&c);
+        // Nested compaction is JSON-rendered then CSV-quoted, so a
+        // `_compaction":"table"` substring should appear inside quotes.
+        assert!(out.contains("_compaction"), "got: {out}");
+    }
+
+    #[test]
+    fn csv_formatter_drop_summary_opt_in() {
+        let mut rows = vec![Row::new(vec![CellValue::Scalar(json!(1))])];
+        rows.push(Row::new(vec![CellValue::Scalar(json!(2))]));
+        let c = Compaction::Table {
+            schema: Schema {
+                fields: vec![super::super::ir::FieldSpec {
+                    name: "x".into(),
+                    type_tag: "int".into(),
+                    nullable: false,
+                }],
+            },
+            rows,
+            original_count: 5, // 3 dropped
+        };
+        let with_summary = CsvSchemaFormatter::new().with_drop_summary().format(&c);
+        assert!(with_summary.contains("__dropped:3"));
+        let without = CsvSchemaFormatter::new().format(&c);
+        assert!(!without.contains("__dropped"));
+    }
+
+    #[test]
+    fn estimate_matches_format_len() {
+        let items = vec![json!({"a": 1}), json!({"a": 2})];
+        let c = compact(&items, &cfg());
+        let f = CsvSchemaFormatter::new();
+        assert_eq!(f.estimate_bytes(&c), f.format(&c).len());
+    }
+
+    // ── Cross-formatter property: same input → smaller CSV than JSON ──
+    // This is the headline value prop. If it doesn't hold for "obviously
+    // tabular" input, the formatter is broken or the fixture is wrong.
+
+    #[test]
+    fn csv_smaller_than_json_for_tabular() {
+        let items: Vec<Value> = (0..50)
+            .map(|i| {
+                json!({
+                    "id": i,
+                    "name": format!("user_{i}"),
+                    "email": format!("user_{i}@example.com"),
+                    "status": if i % 3 == 0 { "ok" } else { "pending" },
+                })
+            })
+            .collect();
+        let c = compact(&items, &cfg());
+        let json_out = JsonFormatter::new().format(&c);
+        let csv_out = CsvSchemaFormatter::new().format(&c);
+        // CSV should beat the structured-JSON formatter (both
+        // deduplicate the schema, so the win comes from removing
+        // structural punctuation only — modest, but real).
+        assert!(
+            csv_out.len() < json_out.len(),
+            "csv {} bytes vs json {} bytes",
+            csv_out.len(),
+            json_out.len()
+        );
+    }
+
+    #[test]
+    fn csv_substantially_smaller_than_raw_json() {
+        // The headline value prop: CSV+schema beats naïve JSON
+        // serialization of the same array (where every row repeats
+        // every field name) by a wide margin.
+        let items: Vec<Value> = (0..50)
+            .map(|i| {
+                json!({
+                    "id": i,
+                    "name": format!("user_{i}"),
+                    "email": format!("user_{i}@example.com"),
+                    "status": if i % 3 == 0 { "ok" } else { "pending" },
+                })
+            })
+            .collect();
+        let c = compact(&items, &cfg());
+        let csv_out = CsvSchemaFormatter::new().format(&c);
+        let raw_json = serde_json::to_string(&Value::Array(items.clone())).unwrap();
+        assert!(
+            csv_out.len() * 10 < raw_json.len() * 7,
+            "csv {} bytes vs raw json {} bytes — expected >30% reduction",
+            csv_out.len(),
+            raw_json.len()
+        );
+    }
+}

crates/headroom-core/src/transforms/smart_crusher/compaction/ir.rs

@@ -0,0 +1,252 @@
+//! Compaction IR — recursive tree representation for lossless / row-lossy
+//! compaction of JSON arrays.
+//!
+//! The IR is the boundary between [`TabularCompactor`] (which produces it)
+//! and [`Formatter`] implementations (which consume it). Renderer-agnostic.
+//!
+//! # Recursive structure
+//!
+//! A `Compaction::Table` has rows of [`CellValue`]s, and a `CellValue` may
+//! itself hold a nested `Compaction`. This enables multi-level compression:
+//! an array whose rows hold stringified-JSON gets recursively compacted
+//! into a sub-table; an opaque blob gets CCR-substituted; a heterogeneous
+//! array gets bucketed by discriminator.
+//!
+//! [`TabularCompactor`]: super::compactor::TabularCompactor
+//! [`Formatter`]: super::formatter::Formatter
+
+use serde_json::Value;
+
+/// What kind of opaque payload was substituted by CCR.
+///
+/// Carried for telemetry and so formatters can render a one-line hint
+/// next to the CCR pointer (e.g. `<<ccr:abc123 base64,2.1KB>>`) without
+/// re-parsing the original bytes.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum OpaqueKind {
+    /// Looks base64-encoded — long, restricted alphabet.
+    Base64Blob,
+    /// Long opaque string the classifier couldn't otherwise place.
+    LongString,
+    /// HTML/XML chunk (detected by `<` density).
+    HtmlChunk,
+    /// Detected format the classifier knows about by name (e.g. "diff",
+    /// "code"). Routing of these into the right transform is deferred
+    /// to a later PR; for now they're treated as `LongString`.
+    Other(String),
+}
+
+/// One column's metadata in a tabular compaction.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FieldSpec {
+    /// Column name. May be dotted for flattened nested fields,
+    /// e.g. `"meta.region"`.
+    pub name: String,
+    /// Inferred type tag. One of: `"int"`, `"float"`, `"string"`,
+    /// `"bool"`, `"null"`, `"json"` (cells render as JSON literals —
+    /// last-resort), `"ccr"` (cells are CCR pointers).
+    pub type_tag: String,
+    /// True if at least one row had this field absent or `null`.
+    pub nullable: bool,
+}
+
+/// Column set for a homogeneous table.
+#[derive(Debug, Clone, PartialEq)]
+pub struct Schema {
+    pub fields: Vec<FieldSpec>,
+}
+
+impl Schema {
+    pub fn field_names(&self) -> Vec<&str> {
+        self.fields.iter().map(|f| f.name.as_str()).collect()
+    }
+}
+
+/// One cell in a row. Most cells are scalar; nested/opaque/recursive
+/// cells branch the tree.
+#[derive(Debug, Clone)]
+pub enum CellValue {
+    /// Scalar JSON value (number, string, bool, null). Formatter renders
+    /// directly per its conventions.
+    Scalar(Value),
+    /// Recursive sub-compaction. Created for inner arrays, parsed
+    /// stringified-JSON, or nested-mixed objects. Formatter recurses.
+    Nested(Box<Compaction>),
+    /// CCR pointer substituting an opaque/large payload. The original
+    /// bytes live in the CCR store keyed by `ccr_hash`.
+    OpaqueRef {
+        ccr_hash: String,
+        byte_size: usize,
+        kind: OpaqueKind,
+    },
+    /// Field is absent in this row. Distinct from `Scalar(Value::Null)`
+    /// — `Missing` means the original object had no such key, while
+    /// `Scalar(Value::Null)` means the key existed and was null.
+    Missing,
+}
+
+/// A row of a tabular compaction. Order and length match the parent
+/// table's [`Schema::fields`].
+#[derive(Debug, Clone)]
+pub struct Row(pub Vec<CellValue>);
+
+impl Row {
+    pub fn new(cells: Vec<CellValue>) -> Self {
+        Self(cells)
+    }
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+/// One bucket of a heterogeneous array, partitioned by a discriminator
+/// field's value (e.g. all rows where `type == "user"`).
+#[derive(Debug, Clone)]
+pub struct Bucket {
+    /// The discriminator value that defines this bucket.
+    pub key: Value,
+    pub schema: Schema,
+    pub rows: Vec<Row>,
+}
+
+/// Top-level compaction result. Tree-shaped via `Nested` cells.
+///
+/// [`Compaction::Table`] is the common case. [`Compaction::Buckets`]
+/// only fires for heterogeneous arrays where a discriminator field
+/// cleanly partitions rows. [`Compaction::Untouched`] is the
+/// fall-through when the compactor declines to operate (e.g. mixed
+/// scalars, or fewer than 2 rows).
+#[derive(Debug, Clone)]
+pub enum Compaction {
+    /// Homogeneous tabular form: N rows × C columns.
+    Table {
+        schema: Schema,
+        rows: Vec<Row>,
+        /// Row count BEFORE any row-dropping under budget pressure.
+        /// `original_count - rows.len()` = rows we had to drop.
+        original_count: usize,
+    },
+    /// Heterogeneous array bucketed by discriminator field.
+    Buckets {
+        discriminator: String,
+        buckets: Vec<Bucket>,
+        /// Total rows across all buckets BEFORE row-dropping.
+        original_count: usize,
+    },
+    /// Single CCR pointer — top-level opaque content. Rare; usually
+    /// CCR refs live inside table cells, not at the top.
+    OpaqueRef {
+        ccr_hash: String,
+        byte_size: usize,
+        kind: OpaqueKind,
+    },
+    /// Compactor declined to compact; pass-through original value.
+    /// The crusher will fall back to the existing lossy path.
+    Untouched(Value),
+}
+
+impl Compaction {
+    /// Total kept rows in this compaction (sum across buckets if
+    /// applicable). 0 for `OpaqueRef` and `Untouched`.
+    pub fn kept_row_count(&self) -> usize {
+        match self {
+            Compaction::Table { rows, .. } => rows.len(),
+            Compaction::Buckets { buckets, .. } => buckets.iter().map(|b| b.rows.len()).sum(),
+            Compaction::OpaqueRef { .. } | Compaction::Untouched(_) => 0,
+        }
+    }
+
+    /// Original (pre-drop) row count. 0 for `OpaqueRef` and `Untouched`.
+    pub fn original_row_count(&self) -> usize {
+        match self {
+            Compaction::Table { original_count, .. } => *original_count,
+            Compaction::Buckets { original_count, .. } => *original_count,
+            Compaction::OpaqueRef { .. } | Compaction::Untouched(_) => 0,
+        }
+    }
+
+    pub fn was_compacted(&self) -> bool {
+        matches!(
+            self,
+            Compaction::Table { .. } | Compaction::Buckets { .. } | Compaction::OpaqueRef { .. }
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn schema_field_names_returns_in_order() {
+        let s = Schema {
+            fields: vec![
+                FieldSpec {
+                    name: "id".into(),
+                    type_tag: "int".into(),
+                    nullable: false,
+                },
+                FieldSpec {
+                    name: "name".into(),
+                    type_tag: "string".into(),
+                    nullable: false,
+                },
+            ],
+        };
+        assert_eq!(s.field_names(), vec!["id", "name"]);
+    }
+
+    #[test]
+    fn untouched_is_not_compacted() {
+        let c = Compaction::Untouched(json!([1, 2, 3]));
+        assert!(!c.was_compacted());
+        assert_eq!(c.kept_row_count(), 0);
+        assert_eq!(c.original_row_count(), 0);
+    }
+
+    #[test]
+    fn table_row_counts() {
+        let c = Compaction::Table {
+            schema: Schema { fields: vec![] },
+            rows: vec![Row::new(vec![]), Row::new(vec![])],
+            original_count: 5,
+        };
+        assert!(c.was_compacted());
+        assert_eq!(c.kept_row_count(), 2);
+        assert_eq!(c.original_row_count(), 5);
+    }
+
+    #[test]
+    fn buckets_aggregate_row_counts() {
+        let c = Compaction::Buckets {
+            discriminator: "type".into(),
+            buckets: vec![
+                Bucket {
+                    key: json!("user"),
+                    schema: Schema { fields: vec![] },
+                    rows: vec![Row::new(vec![]), Row::new(vec![])],
+                },
+                Bucket {
+                    key: json!("order"),
+                    schema: Schema { fields: vec![] },
+                    rows: vec![Row::new(vec![])],
+                },
+            ],
+            original_count: 10,
+        };
+        assert_eq!(c.kept_row_count(), 3);
+        assert_eq!(c.original_row_count(), 10);
+    }
+
+    #[test]
+    fn cell_missing_distinct_from_scalar_null() {
+        let m = CellValue::Missing;
+        let n = CellValue::Scalar(Value::Null);
+        // Smoke test: just confirm both variants exist and Debug differs.
+        assert_ne!(format!("{m:?}"), format!("{n:?}"));
+    }
+}

crates/headroom-core/src/transforms/smart_crusher/compaction/mod.rs

@@ -0,0 +1,85 @@
+//! Compaction subsystem — Stage 3c.2 PR2.
+//!
+//! Lossless-first compaction of JSON arrays. Pipeline:
+//!
+//! ```text
+//! input array
+//!    ↓
+//! [TabularCompactor] → Compaction IR (recursive tree)
+//!    ↓
+//! [Formatter trait] → bytes
+//! ```
+//!
+//! The IR ([`ir::Compaction`]) is a recursive tree so we can express
+//! multi-level compression: nested-uniform objects flatten into dotted
+//! columns, stringified-JSON cells become sub-tables, opaque blobs
+//! become CCR pointers, heterogeneous arrays partition into buckets.
+//!
+//! Formatters consume the IR. [`JsonFormatter`] keeps byte-equal parity
+//! with today's SmartCrusher output. [`CsvSchemaFormatter`] emits a
+//! token-efficient `[N]{cols}:` declaration + JSON schema header + CSV
+//! rows that LLMs read reliably.
+//!
+//! [`JsonFormatter`]: format_json::JsonFormatter
+//! [`CsvSchemaFormatter`]: format_csv_schema::CsvSchemaFormatter
+
+pub mod classifier;
+pub mod compactor;
+pub mod formatter;
+pub mod ir;
+
+pub use classifier::{classify_cell, CellClass, ClassifyConfig};
+pub use compactor::{compact, CompactConfig};
+pub use formatter::{CsvSchemaFormatter, Formatter, JsonFormatter};
+pub use ir::{Bucket, CellValue, Compaction, FieldSpec, OpaqueKind, Row, Schema};
+
+/// Composed compaction stage: a config + formatter pair.
+///
+/// Plug into [`SmartCrusher`] via the builder's `with_compaction(...)`.
+/// When configured, `crush_array` runs compaction as an opt-in
+/// lossless-first stage; when absent (default), behavior is byte-equal
+/// with today's lossy-only path.
+///
+/// [`SmartCrusher`]: super::SmartCrusher
+pub struct CompactionStage {
+    pub config: CompactConfig,
+    pub formatter: Box<dyn Formatter>,
+}
+
+impl CompactionStage {
+    /// CSV+schema formatter, default config — the recommended OSS preset.
+    pub fn default_csv_schema() -> Self {
+        Self {
+            config: CompactConfig::default(),
+            formatter: Box::new(CsvSchemaFormatter::new()),
+        }
+    }
+
+    /// JSON formatter, default config — useful for debugging or for
+    /// downstream consumers that want structured rather than CSV-shaped
+    /// output.
+    pub fn default_json() -> Self {
+        Self {
+            config: CompactConfig::default(),
+            formatter: Box::new(JsonFormatter::new()),
+        }
+    }
+
+    /// Run the stage end-to-end: compact + format. Returns the
+    /// [`Compaction`] tree (so callers can inspect kept/total row
+    /// counts) alongside the rendered bytes.
+    pub fn run(&self, items: &[serde_json::Value]) -> (Compaction, String) {
+        let c = compact(items, &self.config);
+        let rendered = self.formatter.format(&c);
+        (c, rendered)
+    }
+}
+
+impl std::fmt::Debug for CompactionStage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CompactionStage")
+            .field("config", &self.config)
+            .field("formatter", &self.formatter.name())
+            .finish()
+    }
+}

crates/headroom-core/src/transforms/smart_crusher/crusher.rs

@@ -38,6 +38,7 @@ use serde_json::Value;
 use super::analyzer::SmartAnalyzer;
 use super::builder::SmartCrusherBuilder;
 use super::classifier::{classify_array, ArrayType};
+use super::compaction::{Compaction, CompactionStage};
 use super::config::SmartCrusherConfig;
 use super::crushers::{compute_k_split, crush_number_array, crush_object, crush_string_array};
 use super::planning::SmartCrusherPlanner;
@@ -49,15 +50,29 @@ use crate::transforms::anchor_selector::AnchorSelector;
 
 /// Return type for `crush_array` — mirrors Python's
 /// `(crushed_items, strategy_info, ccr_hash, dropped_summary)` tuple.
+///
+/// Stage 3c.2 PR2 added `compacted` and `compaction_kind` for the new
+/// lossless-first compaction stage. They stay `None` when no
+/// [`CompactionStage`] is configured on the crusher (default), so
+/// existing parity guarantees are preserved.
 pub struct CrushArrayResult {
     pub items: Vec<Value>,
     /// Strategy debug string, e.g. `"smart_sample"`, `"top_n"`,
     /// `"none:adaptive_at_limit"`, `"skip:unique_entities_no_signal"`.
+    /// When compaction wins, this is `"compaction:<formatter_name>"`.
     pub strategy_info: String,
     /// CCR retrieval hash if caching is enabled. Stub: always `None`.
     pub ccr_hash: Option<String>,
     /// Categorical summary of dropped items. Stub: always empty.
     pub dropped_summary: String,
+    /// Rendered bytes from the compaction stage. `Some(s)` when the
+    /// stage was configured AND produced non-`Untouched` output;
+    /// `None` otherwise (default OSS path).
+    pub compacted: Option<String>,
+    /// Top-level [`Compaction`] variant tag — `"table"`, `"buckets"`,
+    /// `"ccr"`, or `"untouched"`. Mirrors `compacted` — populated only
+    /// when the compaction stage ran.
+    pub compaction_kind: Option<&'static str>,
 }
 
 /// Top-level SmartCrusher.
@@ -78,6 +93,11 @@ pub struct SmartCrusher {
     pub analyzer: SmartAnalyzer,
     pub constraints: Vec<Box<dyn Constraint>>,
     pub observers: Vec<Box<dyn Observer>>,
+    /// Optional lossless-first compaction stage (Stage 3c.2 PR2). When
+    /// set, `crush_array` runs compaction up front and short-circuits
+    /// the lossy path on success. When `None` (default OSS), parity
+    /// with the pre-PR2 lossy-only pipeline is preserved exactly.
+    pub compaction: Option<CompactionStage>,
 }
 
 impl SmartCrusher {
@@ -116,6 +136,7 @@ impl SmartCrusher {
     /// [`SmartCrusherBuilder::build`] — not part of the public stable
     /// API. Prefer the builder.
     #[doc(hidden)]
+    #[allow(clippy::too_many_arguments)]
     pub fn from_parts(
         config: SmartCrusherConfig,
         anchor_selector: AnchorSelector,
@@ -123,6 +144,7 @@ impl SmartCrusher {
         analyzer: SmartAnalyzer,
         constraints: Vec<Box<dyn Constraint>>,
         observers: Vec<Box<dyn Observer>>,
+        compaction: Option<CompactionStage>,
     ) -> Self {
         SmartCrusher {
             config,
@@ -131,6 +153,7 @@ impl SmartCrusher {
             analyzer,
             constraints,
             observers,
+            compaction,
         }
     }
 
@@ -362,6 +385,25 @@ impl SmartCrusher {
     /// 7. `execute_plan(plan, items)` → result.
     /// 8. Strategy info = `analysis.recommended_strategy.as_str()`.
     pub fn crush_array(&self, items: &[Value], query_context: &str, bias: f64) -> CrushArrayResult {
+        // ── Stage 0 (PR2): lossless-first compaction (opt-in) ──
+        //
+        // Runs only when a CompactionStage is configured. Sits BEFORE
+        // the existing lossy pipeline so the lossless path can
+        // short-circuit on success. When the compactor declines
+        // (returns Untouched) the rest of crush_array runs unchanged
+        // — preserving byte-equal parity with the pre-PR2 path.
+        let compaction_result: Option<(String, &'static str)> =
+            if let Some(stage) = &self.compaction {
+                let (c, rendered) = stage.run(items);
+                if c.was_compacted() {
+                    Some((rendered, compaction_kind_str(&c)))
+                } else {
+                    None
+                }
+            } else {
+                None
+            };
+
         let item_strings: Vec<String> = items
             .iter()
             .map(|i| serde_json::to_string(i).unwrap_or_default())
@@ -382,9 +424,11 @@ impl SmartCrusher {
             // in this stage).
             return CrushArrayResult {
                 items: items.to_vec(),
-                strategy_info: "none:adaptive_at_limit".to_string(),
+                strategy_info: compaction_strategy_or(&compaction_result, "none:adaptive_at_limit"),
                 ccr_hash: None,
                 dropped_summary: String::new(),
+                compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
+                compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
             };
         }
 
@@ -401,9 +445,11 @@ impl SmartCrusher {
             };
             return CrushArrayResult {
                 items: items.to_vec(),
-                strategy_info: reason,
+                strategy_info: compaction_strategy_or(&compaction_result, &reason),
                 ccr_hash: None,
                 dropped_summary: String::new(),
+                compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
+                compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
             };
         }
 
@@ -419,13 +465,16 @@ impl SmartCrusher {
         let result = self.execute_plan(&plan, items);
 
         // CCR/telemetry/TOIN-record paths stubbed.
-        let strategy_info = analysis.recommended_strategy.as_str().to_string();
+        let strategy_info =
+            compaction_strategy_or(&compaction_result, analysis.recommended_strategy.as_str());
 
         CrushArrayResult {
             items: result,
             strategy_info,
             ccr_hash: None,
             dropped_summary: String::new(),
+            compacted: compaction_result.as_ref().map(|(s, _)| s.clone()),
+            compaction_kind: compaction_result.as_ref().map(|(_, k)| *k),
         }
     }
 
@@ -619,6 +668,28 @@ fn canonical_json_for_match(value: &Value) -> String {
     crate::transforms::anchor_selector::python_json_dumps_sort_keys(value)
 }
 
+/// Maps a `Compaction` to a stable kind tag exposed via `CrushArrayResult`.
+fn compaction_kind_str(c: &Compaction) -> &'static str {
+    match c {
+        Compaction::Table { .. } => "table",
+        Compaction::Buckets { .. } => "buckets",
+        Compaction::OpaqueRef { .. } => "ccr",
+        Compaction::Untouched(_) => "untouched",
+    }
+}
+
+/// If compaction won, override the lossy strategy string with
+/// `"compaction:<formatter_name>:<kind>"`. Else return `default`.
+fn compaction_strategy_or(
+    compaction_result: &Option<(String, &'static str)>,
+    default: &str,
+) -> String {
+    match compaction_result {
+        Some((_, kind)) => format!("compaction:{kind}"),
+        None => default.to_string(),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -880,4 +951,63 @@ mod tests {
         let result = c.crush_array(&items, "anything", 1.0);
         assert!(result.items.len() <= 30);
     }
+
+    // ---------- Stage 3c.2 PR2: lossless-first compaction wiring ----------
+
+    #[test]
+    fn no_compaction_stage_yields_none_compacted_field() {
+        // Default SmartCrusher::new() sets compaction = None. Existing
+        // parity preserved: compacted/compaction_kind both None.
+        let c = crusher();
+        let items: Vec<Value> = (0..30).map(|_| json!({"status": "ok"})).collect();
+        let result = c.crush_array(&items, "", 1.0);
+        assert!(result.compacted.is_none());
+        assert!(result.compaction_kind.is_none());
+    }
+
+    #[test]
+    fn compaction_stage_runs_and_populates_compacted() {
+        let c = SmartCrusherBuilder::new(SmartCrusherConfig::default())
+            .with_default_oss_setup()
+            .with_default_compaction()
+            .build();
+        let items: Vec<Value> = (0..50)
+            .map(|i| json!({"id": i, "name": format!("u_{i}"), "status": "ok"}))
+            .collect();
+        let result = c.crush_array(&items, "", 1.0);
+        let compacted = result.compacted.expect("compacted should be set");
+        assert!(compacted.starts_with("[50]{"), "got: {compacted}");
+        assert_eq!(result.compaction_kind, Some("table"));
+        assert!(result.strategy_info.starts_with("compaction:table"));
+    }
+
+    #[test]
+    fn compaction_does_not_alter_lossy_items_field() {
+        // The compacted bytes are emitted alongside the existing lossy
+        // result. items field still holds the lossy-path output so
+        // downstream consumers can choose either form.
+        let c = SmartCrusherBuilder::new(SmartCrusherConfig::default())
+            .with_default_oss_setup()
+            .with_default_compaction()
+            .build();
+        let items: Vec<Value> = (0..30).map(|_| json!({"status": "ok"})).collect();
+        let with_compaction = c.crush_array(&items, "", 1.0);
+
+        let baseline = crusher().crush_array(&items, "", 1.0);
+        assert_eq!(with_compaction.items.len(), baseline.items.len());
+    }
+
+    #[test]
+    fn compaction_skips_non_object_array() {
+        // Compactor returns Untouched for non-object arrays → no
+        // compacted field populated, no kind tag.
+        let c = SmartCrusherBuilder::new(SmartCrusherConfig::default())
+            .with_default_oss_setup()
+            .with_default_compaction()
+            .build();
+        let items: Vec<Value> = (0..30).map(|i| json!(i)).collect();
+        let result = c.crush_array(&items, "", 1.0);
+        assert!(result.compacted.is_none());
+        assert!(result.compaction_kind.is_none());
+    }
 }

crates/headroom-core/src/transforms/smart_crusher/mod.rs

@@ -36,6 +36,7 @@ mod analyzer;
 mod anchors;
 mod builder;
 mod classifier;
+pub mod compaction;
 mod config;
 mod constraints;
 mod crusher;

plugins/headroom-agent-hooks/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "headroom",
-  "version": "0.10.18",
+  "version": "0.11.0",
   "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
   "author": {
     "name": "Headroom Contributors",

plugins/headroom-agent-hooks/.github/plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "headroom",
-  "version": "0.10.18",
+  "version": "0.11.0",
   "description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
   "author": {
     "name": "Headroom Contributors",