fix(ccr): store opaque blobs from lossless:table compaction (#1083) (#1182)

## Description

SmartCrusher's `lossless:table` compaction path emits opaque-blob CCR
markers
(`<<ccr:HASH,KIND,SIZE>>`) but never wrote the original payload to the
CCR
store. As a result `GET /v1/retrieve/{hash}` and the `headroom_retrieve`
tool
return **404** for those hashes. The opaque-*string* path
(`walker::emit_opaque_ccr_marker`) already stores its payload; the table
compactor diverged simply because no store was threaded into it.

Closes #1083

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `compaction/compactor.rs`: add `compact_with_store(items, cfg, store)`
and a
  private `compact_inner`; thread `Option<&Arc<dyn CcrStore>>` through
`build_homogeneous_table` → `build_row` → `cell_from_value` and the
recursive
bucket/nested calls. In the `Opaque` branch, `store.put(&hash, payload)`
under
  the **same** `hash_opaque` value that becomes the marker hash (mirrors
`walker::emit_opaque_ccr_marker`). Public `compact` is unchanged — it
delegates
  with `None`.
- `compaction/mod.rs`: add `CompactionStage::run_with_store`; `run` is
unchanged.
- `crusher.rs`: the lossless branch now calls
`stage.run_with_store(items, self.ccr_store.as_ref())` instead of
`stage.run(items)`.
- Two new unit tests in `compactor.rs` (see below).

The IR (and therefore the rendered marker text) is identical whether or
not a
store is supplied — the store only gains the write that should already
have
happened, so existing output stays byte-for-byte the same.

## Testing

<!-- Check what you actually ran, then paste the real command output
below. -->

- [ ] Unit tests pass (`pytest`)
- [ ] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

> Note: this change is in the Rust core (`crates/headroom-core`), so the
> Python-specific checks above are N/A. The Rust equivalents were run:

### Test Output

```text
$ cargo test -p headroom-core --lib compaction
test result: ok. 70 passed; 0 failed; 0 ignored; 0 measured; 766 filtered out; finished in 0.01s

$ cargo fmt -p headroom-core -- --check
# clean (exit 0)
```

New tests:
- `opaque_payload_is_stored_under_marker_hash` — after
`compact_with_store`, the
original blob is retrievable via `store.get(marker_hash)`, and the
stored key
  equals `hash_opaque(payload)` (locks the key↔marker contract).
- `store_presence_does_not_change_the_ir` — `compact` and
`compact_with_store`
  produce identical IR; only the store write is added.

(The full `cargo test -p headroom-core --lib` run has 18 pre-existing
failures,
all in `transforms::magika_detector` — they require the ONNX
runtime/model and
are unrelated to this change. All 70 compaction + crusher tests pass.)

## Real Behavior Proof

- Environment: Windows, Rust 1.95.0, `cargo test -p headroom-core` (no
live proxy).
- Exact command / steps: build a 2-item array with a long opaque-blob
field →
  `compact_with_store(&items, &cfg, Some(&InMemoryCcrStore))` → read the
  `OpaqueRef.ccr_hash` from the IR → `store.get(ccr_hash)`.
- Observed result: before the fix the store is empty (retrieval would
404);
after the fix `store.get(ccr_hash) == Some(original_payload)` and the
marker
  hash is unchanged.
- Not tested: end-to-end through a running proxy / a real `GET
/v1/retrieve/{hash}`
HTTP round-trip. Verified at the unit level that the store now receives
the
payload under the exact marker hash, which is the write that was
missing.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

- Docs/CHANGELOG checklist items are N/A — this is an internal
correctness fix
  with no user-facing API change.
- Scope is intentionally minimal: public `compact`/`run` signatures are
  preserved (delegating with `None`), so all existing callers and the 68
in-crate compaction tests are unaffected. Only the lossless
`crush_array`
  branch opts into the store-threading via `run_with_store`.

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

@@ -40,12 +40,14 @@
 //! [`CellClass::Opaque`]: super::classifier::CellClass::Opaque
 
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 use serde_json::Value;
 use sha2::{Digest, Sha256};
 
 use super::classifier::{classify_cell, CellClass, ClassifyConfig};
 use super::ir::{Bucket, CellValue, Compaction, FieldSpec, Row, Schema};
+use crate::ccr::CcrStore;
 
 /// Config for the compactor.
 #[derive(Debug, Clone)]
@@ -95,7 +97,37 @@ impl Default for CompactConfig {
 }
 
 /// Top-level compaction entry point.
+///
+/// Opaque blobs become CCR pointers, but the original payload is **not**
+/// stored — callers that need `<<ccr:...>>` markers to resolve on
+/// retrieval must use [`compact_with_store`] instead.
 pub fn compact(items: &[Value], cfg: &CompactConfig) -> Compaction {
+    compact_inner(items, cfg, None)
+}
+
+/// Like [`compact`], but stash every opaque-blob payload in `store` under
+/// the same 12-char hash that ends up in its `<<ccr:HASH,...>>` marker, so
+/// the runtime can serve the original back on a `headroom_retrieve` call /
+/// `GET /v1/retrieve/{hash}`. Mirrors the contract already honored by the
+/// opaque-string path in [`super::walker::emit_opaque_ccr_marker`].
+///
+/// The IR (and therefore the rendered marker text) is identical whether or
+/// not a store is supplied — the store only adds a side-effecting write, so
+/// `compact(items, cfg)` and `compact_with_store(items, cfg, Some(store))`
+/// return the same [`Compaction`].
+pub fn compact_with_store(
+    items: &[Value],
+    cfg: &CompactConfig,
+    store: Option<&Arc<dyn CcrStore>>,
+) -> Compaction {
+    compact_inner(items, cfg, store)
+}
+
+fn compact_inner(
+    items: &[Value],
+    cfg: &CompactConfig,
+    store: Option<&Arc<dyn CcrStore>>,
+) -> Compaction {
     if items.len() < cfg.min_items {
         return Compaction::Untouched(Value::Array(items.to_vec()));
     }
@@ -117,14 +149,14 @@ pub fn compact(items: &[Value], cfg: &CompactConfig) -> Compaction {
 
     if core_ratio < cfg.heterogeneous_core_ratio {
         if let Some(disc) = detect_discriminator(items, &key_freqs, cfg) {
-            return bucket_by(items, &disc, cfg);
+            return bucket_by(items, &disc, cfg, store);
         }
         // 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)
+    build_homogeneous_table(items, &key_freqs, cfg, store)
 }
 
 fn compute_key_freqs(items: &[Value]) -> BTreeMap<String, usize> {
@@ -143,6 +175,7 @@ fn build_homogeneous_table(
     items: &[Value],
     key_freqs: &BTreeMap<String, usize>,
     cfg: &CompactConfig,
+    store: Option<&Arc<dyn CcrStore>>,
 ) -> Compaction {
     // Order: descending frequency, then alphabetical for stability.
     let mut keys: Vec<(&String, &usize)> = key_freqs.iter().collect();
@@ -165,7 +198,7 @@ fn build_homogeneous_table(
 
     let mut rows: Vec<Row> = items
         .iter()
-        .map(|item| build_row(item, &ordered_keys, cfg))
+        .map(|item| build_row(item, &ordered_keys, cfg, store))
         .collect();
 
     flatten_uniform_nested(&mut field_specs, &mut rows, cfg);
@@ -179,7 +212,12 @@ fn build_homogeneous_table(
     }
 }
 
-fn build_row(item: &Value, ordered_keys: &[String], cfg: &CompactConfig) -> Row {
+fn build_row(
+    item: &Value,
+    ordered_keys: &[String],
+    cfg: &CompactConfig,
+    store: Option<&Arc<dyn CcrStore>>,
+) -> Row {
     let obj = match item.as_object() {
         Some(o) => o,
         None => return Row::new(vec![]),
@@ -188,13 +226,13 @@ fn build_row(item: &Value, ordered_keys: &[String], cfg: &CompactConfig) -> Row
         .iter()
         .map(|k| match obj.get(k) {
             None => CellValue::Missing,
-            Some(v) => cell_from_value(v, cfg),
+            Some(v) => cell_from_value(v, cfg, store),
         })
         .collect();
     Row::new(cells)
 }
 
-fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue {
+fn cell_from_value(v: &Value, cfg: &CompactConfig, store: Option<&Arc<dyn CcrStore>>) -> CellValue {
     match classify_cell(v, &cfg.classify) {
         CellClass::Scalar => CellValue::Scalar(v.clone()),
         CellClass::JsonObject => CellValue::Scalar(v.clone()), // flatten pass may promote
@@ -202,7 +240,7 @@ fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue {
             // 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)));
+                    return CellValue::Nested(Box::new(compact_inner(items, cfg, store)));
                 }
             }
             CellValue::Scalar(v.clone())
@@ -212,19 +250,28 @@ fn cell_from_value(v: &Value, cfg: &CompactConfig) -> CellValue {
             // 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)));
+                    return CellValue::Nested(Box::new(compact_inner(items, cfg, store)));
                 }
             }
             CellValue::Scalar(parsed)
         }
         CellClass::Opaque(kind) => {
-            let bytes = match v {
-                Value::String(s) => s.as_bytes(),
+            let s = match v {
+                Value::String(s) => s,
                 _ => return CellValue::Scalar(v.clone()),
             };
+            let ccr_hash = hash_opaque(s.as_bytes());
+            // Stash the original so `GET /v1/retrieve/{hash}` and the
+            // `headroom_retrieve` tool can serve it back — mirrors
+            // `walker::emit_opaque_ccr_marker`. Without this write the
+            // marker points at a key that was never stored and retrieval
+            // 404s (issue #1083).
+            if let Some(store) = store {
+                store.put(&ccr_hash, s);
+            }
             CellValue::OpaqueRef {
-                ccr_hash: hash_opaque(bytes),
-                byte_size: bytes.len(),
+                ccr_hash,
+                byte_size: s.len(),
                 kind,
             }
         }
@@ -451,7 +498,12 @@ fn detect_discriminator(
     best.map(|(k, _)| k)
 }
 
-fn bucket_by(items: &[Value], discriminator: &str, cfg: &CompactConfig) -> Compaction {
+fn bucket_by(
+    items: &[Value],
+    discriminator: &str,
+    cfg: &CompactConfig,
+    store: Option<&Arc<dyn CcrStore>>,
+) -> Compaction {
     let mut groups: BTreeMap<String, Vec<Value>> = BTreeMap::new();
     for item in items {
         let key = item
@@ -465,7 +517,7 @@ fn bucket_by(items: &[Value], discriminator: &str, cfg: &CompactConfig) -> Compa
     let buckets: Vec<Bucket> = groups
         .into_iter()
         .map(|(key, group_items)| {
-            let inner = compact(&group_items, cfg);
+            let inner = compact_inner(&group_items, cfg, store);
             match inner {
                 Compaction::Table { schema, rows, .. } => Bucket {
                     key: Value::String(key),
@@ -730,4 +782,69 @@ mod tests {
         assert_ne!(h1, h3);
         assert_eq!(h1.len(), 12);
     }
+
+    #[test]
+    fn opaque_payload_is_stored_under_marker_hash() {
+        use crate::ccr::{CcrStore, InMemoryCcrStore};
+        use std::sync::Arc;
+
+        // Same blob the `opaque_cell_becomes_ccr_ref` test uses — known to
+        // classify as Opaque, so the OpaqueRef / `<<ccr:HASH,...>>` path runs.
+        let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8);
+        let items = vec![
+            json!({"id": 1, "blob": big.clone()}),
+            json!({"id": 2, "blob": big.clone()}),
+        ];
+
+        let store: Arc<dyn CcrStore> = Arc::new(InMemoryCcrStore::new());
+        let c = compact_with_store(&items, &cfg(), Some(&store));
+
+        // Pull the hash the rendered marker will carry out of the IR.
+        let hash = match &c {
+            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, .. } => ccr_hash.clone(),
+                    other => panic!("expected OpaqueRef, got {other:?}"),
+                }
+            }
+            other => panic!("expected Table, got {other:?}"),
+        };
+
+        // Issue #1083: the original payload must be retrievable under the
+        // exact hash the marker carries (before the fix the store was empty).
+        assert_eq!(store.get(&hash).as_deref(), Some(big.as_str()));
+        // Lock the key<->marker contract: stored key == hash_opaque(payload).
+        assert_eq!(hash, hash_opaque(big.as_bytes()));
+    }
+
+    #[test]
+    fn store_presence_does_not_change_the_ir() {
+        use crate::ccr::{CcrStore, InMemoryCcrStore};
+        use std::sync::Arc;
+
+        let big = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=".repeat(8);
+        let items = vec![
+            json!({"id": 1, "blob": big.clone()}),
+            json!({"id": 2, "blob": big.clone()}),
+        ];
+
+        let without = compact(&items, &cfg());
+        let store: Arc<dyn CcrStore> = Arc::new(InMemoryCcrStore::new());
+        let with = compact_with_store(&items, &cfg(), Some(&store));
+
+        // Marker text is store-independent — only a side-effecting write is
+        // added, so the two IRs render identically. Compare a deterministic
+        // formatter's output rather than Debug formatting, which is not a
+        // stable contract.
+        use super::super::{Formatter, JsonFormatter};
+        let fmt = JsonFormatter::new();
+        assert_eq!(fmt.format(&without), fmt.format(&with));
+        // ...and that write actually happened.
+        assert!(!store.is_empty());
+    }
 }

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

@@ -30,7 +30,7 @@ pub mod ir;
 pub mod walker;
 
 pub use classifier::{classify_cell, CellClass, ClassifyConfig};
-pub use compactor::{compact, CompactConfig};
+pub use compactor::{compact, compact_with_store, CompactConfig};
 pub use formatter::{CsvSchemaFormatter, Formatter, JsonFormatter, MarkdownKvFormatter};
 pub use ir::{Bucket, CellValue, Compaction, FieldSpec, OpaqueKind, Row, Schema};
 pub use walker::{
@@ -115,6 +115,24 @@ impl CompactionStage {
         let rendered = self.formatter.format(&c);
         (c, rendered)
     }
+
+    /// Like [`Self::run`], but stash every opaque-blob payload into `store`
+    /// under the same hash the rendered `<<ccr:HASH,...>>` marker carries,
+    /// so `GET /v1/retrieve/{hash}` and the `headroom_retrieve` tool can
+    /// serve the original back. `SmartCrusher::crush_array`'s lossless
+    /// branch passes the proxy's CCR store here; previously it called
+    /// [`Self::run`], which rendered markers whose payload was never stored
+    /// (issue #1083). When `store` is `None`, behaves exactly like
+    /// [`Self::run`].
+    pub fn run_with_store(
+        &self,
+        items: &[serde_json::Value],
+        store: Option<&std::sync::Arc<dyn crate::ccr::CcrStore>>,
+    ) -> (Compaction, String) {
+        let c = compact_with_store(items, &self.config, store);
+        let rendered = self.formatter.format(&c);
+        (c, rendered)
+    }
 }
 
 impl std::fmt::Debug for CompactionStage {

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

@@ -672,7 +672,11 @@ impl SmartCrusher {
         // ship it — nothing dropped, no CCR retrieval needed.
         // Otherwise fall through to the lossy path.
         if let Some(stage) = &self.compaction {
-            let (c, rendered) = stage.run(items);
+            // Thread the CCR store so opaque-blob `<<ccr:HASH,...>>` markers
+            // emitted by lossless:table compaction are actually retrievable
+            // (issue #1083); the row-drop lossy path below stores its own
+            // payload separately.
+            let (c, rendered) = stage.run_with_store(items, self.ccr_store.as_ref());
             if c.was_compacted() {
                 let input_bytes = estimate_array_bytes(&item_strings);
                 let savings_ratio = if input_bytes > 0 {