fix(smart-crusher): honor enable_ccr_marker on the opaque-blob path (#1130)

## Description

Closes #1091.

SmartCrusher's array compaction is lossless-first, but the
**opaque-blob** substitution path emitted `<<ccr:HASH,string,KB>>`
markers **unconditionally** — it did not honor `enable_ccr_marker` /
`inject_retrieval_marker`, which gate only the lossy **row-drop** path.
As the issue notes, the consequence was that *no configuration produced
guaranteed-lossless, marker-free output*: any array with a single string
cell over `opaque_min_bytes` (256B default) still emitted a CCR marker,
forcing a retrieval round-trip for consumers that need verbatim output.

**Root cause:** the row-drop path is gated (`crusher.rs` — `if
dropped_count > 0 && self.config.enable_ccr_marker`), but opaque
classification in `compaction/classifier.rs` keyed purely on byte
length, with no reference to the flag, and both emit sites (`walker.rs`,
`crusher.rs`) then produced a marker.

**Fix:** thread the gate into classification. `ClassifyConfig` gains an
`emit_opaque_markers` field (default `true`); when `false`, a long
string is classified `Scalar` (kept verbatim) instead of `Opaque`, so no
marker is emitted and nothing is written to the CCR store anywhere
downstream. The flag is set from `enable_ccr_marker` at both
`ClassifyConfig` construction sites in `crusher.rs`.

> Design note: gating at the classifier (rather than at marker-emit
time) is the single complete fix — it covers all three emit paths
(walker inline-substitution, the crusher string path, and the compactor
`OpaqueRef`→formatter path, which no longer has the original string by
the time it formats). One consequence: with markers **off**, an array
dominated by unique long-string cells now falls through to a
conservative passthrough (`skip:unique_entities_no_signal`) instead of a
lossy opaque table — still lossless and marker-free, which is the point
of disabling markers. If you'd rather preserve structural table
compaction with the blob inlined verbatim, that's a larger change at the
emit + compactor layers; happy to take it that direction if preferred.
Default behavior (`enable_ccr_marker=true`) is unchanged.

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

-
`crates/headroom-core/src/transforms/smart_crusher/compaction/classifier.rs`:
add `emit_opaque_markers: bool` (default `true`) to `ClassifyConfig`; in
`classify_cell`, keep long strings `Scalar` when it is `false`. New unit
test `long_string_stays_scalar_when_opaque_markers_disabled`.
- `crates/headroom-core/src/transforms/smart_crusher/crusher.rs`: set
`classify.emit_opaque_markers = config.enable_ccr_marker` at both
`ClassifyConfig` construction sites (the `CompactConfig` builder and the
standalone string path).
- `tests/test_smart_crusher_toin_attachment.py`: regression test pinning
both directions — markers ON ⇒ opaque marker present (input really
triggers the path); markers OFF ⇒ no marker, blob verbatim.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Rust tests pass (`cargo test`)
- [x] Linting passes (`ruff check .`, `cargo fmt --check`, `cargo clippy
-- -D warnings`)
- [ ] Type checking (`mypy headroom`) — N/A (no headroom/ Python source
changed)
- [x] New tests added

### Test Output

```text
# Rust
$ cargo test -p headroom-core --lib smart_crusher
test result: ok. 319 passed; 0 failed
  (incl. new: ...classifier::tests::long_string_stays_scalar_when_opaque_markers_disabled ... ok)
$ cargo fmt --check && cargo clippy --workspace -- -D warnings
ok

# Python (after `uv pip install -e .` to rebuild the Rust core)
$ pytest tests/test_smart_crusher_toin_attachment.py tests/test_transforms/ tests/test_ccr_row_drop_store_bridge.py
289 passed, 35 skipped

# Full suite is green except the 5 pre-existing caplog logging-isolation
# flakes that are unrelated to this change and fixed separately in #1117.
```

## Real Behavior Proof

- Environment: macOS, Python 3.13.3, Rust core rebuilt via `uv pip
install -e .`.
- Exact command / steps: crush a 60-row array whose rows carry a
distinct >256B `blob` string, with `inject_retrieval_marker` ON then
OFF.
- Observed result: with `inject_retrieval_marker` OFF (after this fix)
the crushed output contains NO `<<ccr:` marker and the original
`sentinel5_…` blob survives verbatim; before the fix the same input
still emitted `<<ccr:…,string,407B>>` (the bug); with markers ON
behavior is unchanged. Concretely:
- markers ON → `strategy=lossless:table`, output contains
`<<ccr:…,string,407B>>` (blob replaced).
- markers OFF (before fix) → `lossless:table` **still emitted
`<<ccr:…>>`** (the bug).
- markers OFF (after fix) → no `<<ccr:` marker, the original
`sentinel5_…` blob present verbatim.
- Not tested: behavior under CI's sharded jobs specifically; fix is
deterministic and config-gated.

## 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 — N/A
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md — N/A

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -57,6 +57,11 @@ pub struct ClassifyConfig {
     /// `<` count above which a long string is considered HTML-ish.
     /// Default: 3.
     pub html_min_open_brackets: usize,
+    /// When false, long strings are NOT classified as opaque — they stay
+    /// `Scalar` and render verbatim, so output is marker-free and
+    /// guaranteed-lossless. Mirrors the row-drop path's `enable_ccr_marker`
+    /// gate (see `crusher.rs`). Default: true.
+    pub emit_opaque_markers: bool,
 }
 
 impl Default for ClassifyConfig {
@@ -65,6 +70,7 @@ impl Default for ClassifyConfig {
             opaque_min_bytes: 256,
             base64_alphabet_ratio: 0.95,
             html_min_open_brackets: 3,
+            emit_opaque_markers: true,
         }
     }
 }
@@ -94,8 +100,10 @@ fn classify_string(s: &str, cfg: &ClassifyConfig) -> CellClass {
         }
     }
 
-    // Opaque-blob check — only for strings above the byte threshold.
-    if s.len() <= cfg.opaque_min_bytes {
+    // Opaque-blob check — only for strings above the byte threshold, and
+    // only when opaque markers are enabled. With markers off, keep the full
+    // string verbatim (Scalar) so the output stays lossless and marker-free.
+    if s.len() <= cfg.opaque_min_bytes || !cfg.emit_opaque_markers {
         return CellClass::Scalar;
     }
 
@@ -234,6 +242,22 @@ mod tests {
         }
     }
 
+    #[test]
+    fn long_string_stays_scalar_when_opaque_markers_disabled() {
+        // #1091: with opaque markers disabled, a long string must NOT be
+        // classified Opaque (which would emit a `<<ccr:>>` marker); it stays
+        // Scalar and renders verbatim, so the output is lossless.
+        let v = Value::String("x".repeat(512));
+        // Default config classifies it Opaque.
+        assert!(matches!(classify_cell(&v, &cfg()), CellClass::Opaque(_)));
+        // Markers disabled → Scalar (verbatim).
+        let no_markers = ClassifyConfig {
+            emit_opaque_markers: false,
+            ..ClassifyConfig::default()
+        };
+        assert_eq!(classify_cell(&v, &no_markers), CellClass::Scalar);
+    }
+
     #[test]
     fn base64_blob_detected() {
         let s = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/==".repeat(5);

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

@@ -163,6 +163,13 @@ impl SmartCrusher {
             max_flatten_inner_keys: config.compaction_max_flatten_inner_keys,
             min_buckets: config.compaction_min_buckets,
             max_buckets: config.compaction_max_buckets,
+            // Honor the CCR marker gate for opaque-blob cells too (not just
+            // the row-drop path), so `enable_ccr_marker=false` yields
+            // marker-free, lossless output. Fixes #1091.
+            classify: ClassifyConfig {
+                emit_opaque_markers: config.enable_ccr_marker,
+                ..ClassifyConfig::default()
+            },
             ..CompactConfig::default()
         };
         SmartCrusherBuilder::new(config)
@@ -595,7 +602,12 @@ impl SmartCrusher {
         // 2. Opaque blob: substitute with CCR marker AND stash the
         // original in the store (PR8) so retrieval works. Hash + format
         // identical to walker.rs via the shared helper — zero drift.
-        let cfg = ClassifyConfig::default();
+        // Gated by `enable_ccr_marker` so disabling markers stays lossless
+        // here too (#1091).
+        let cfg = ClassifyConfig {
+            emit_opaque_markers: self.config.enable_ccr_marker,
+            ..ClassifyConfig::default()
+        };
         if let CellClass::Opaque(kind) = classify_cell(&Value::String(s.to_string()), &cfg) {
             let marker = emit_opaque_ccr_marker(s, &kind, self.ccr_store.as_ref());
             let kind_label = opaque_kind_label(&kind);

tests/test_smart_crusher_toin_attachment.py

@@ -194,6 +194,41 @@ def test_ccr_inject_marker_false_suppresses_markers_in_output(fresh_toin):
     assert "_ccr_dropped" not in result.compressed
 
 
+def test_ccr_inject_marker_false_suppresses_opaque_blob_markers(fresh_toin):
+    """#1091: `inject_retrieval_marker=False` must also suppress the
+    *opaque-blob* CCR markers, not just the row-drop path.
+
+    A long string cell (> opaque_min_bytes) used to be substituted with a
+    `<<ccr:HASH,string,KB>>` marker unconditionally — so no config produced
+    guaranteed-lossless output. This test pins both directions: with markers
+    ON the opaque blob IS replaced by a marker (proving the input genuinely
+    triggers the opaque path), and with markers OFF the blob survives verbatim
+    with no marker."""
+    import json
+
+    from headroom.config import CCRConfig
+
+    # Distinct >256-byte string cells trigger the opaque-blob path.
+    payload = json.dumps(
+        [{"id": i, "name": f"row{i}", "blob": f"sentinel{i}_" + "x" * 400} for i in range(60)]
+    )
+
+    on = SmartCrusher(
+        SmartCrusherConfig(),
+        ccr_config=CCRConfig(enabled=True, inject_retrieval_marker=True),
+    ).crush(payload, query="", bias=1.0)
+    # Sanity: the input really does exercise the opaque-blob path.
+    assert "<<ccr:" in on.compressed, "input should trigger an opaque marker when markers are ON"
+
+    off = SmartCrusher(
+        SmartCrusherConfig(),
+        ccr_config=CCRConfig(enabled=True, inject_retrieval_marker=False),
+    ).crush(payload, query="", bias=1.0)
+    assert "<<ccr:" not in off.compressed, f"expected no opaque marker, got: {off.compressed!r}"
+    # The original blob content must survive verbatim (guaranteed-lossless).
+    assert "sentinel5_" in off.compressed
+
+
 def test_ccr_inject_marker_true_emits_markers_when_lossy(fresh_toin):
     """The opt-in case keeps marker emission on. If the lossy path
     runs (which it should for a sufficiently big crushable payload),