fix(ccr): stop persisting retrieval markers as original content (#2694) (#2703)

## Description

CCR entries could end up holding a `<<ccr:...>>` marker — or nothing at
all — where the original bytes belonged, so `headroom_retrieve(hash)`
answered with the very placeholder the caller was trying to resolve. For
a base64/credential field that is permanent, silent data loss: the inner
marker's hash is the only handle on the real payload, and it disappears
from anywhere the model can see.

Four sites, one root cause — **a compressed intermediate (or nothing)
was stored in place of the source**, the same defect class as #1209 (tag
placeholders persisted as originals).

Closes #2694

## 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/walker.rs`** — `walk_array` compacted through the
store-LESS `compact()`. Opaque cells inside a compacted table got a
marker whose payload was **never written**, so retrieval 404'd forever.
Now uses `compact_with_store` so the emitted hash resolves.
- **`compaction/classifier.rs`** — nothing stopped an already-marked
string from being offloaded a second time, which stashed the MARKER as
the new entry's "original". Marker-bearing text is our own output, not
source content, so it is never classified opaque. One guard at the choke
point both the walker and the table compactor share.
- **`smart_crusher/crusher.rs`** — on the prose-hook path the row-drop
marker hashed and stored rows whose leaves were **already** rewritten
(prose compressed, blobs marker-substituted), so retrieving dropped rows
returned compressed output. Now hashes and stashes the pre-processing
array via `crush_array_with_source`.
- **`content_router.py`** — compression pinning matched only `Retrieve
more: hash=` / `Retrieve original: hash=`, **not** `<<ccr:`, so
opaque-blob output was readmitted to the compressor on a later turn —
the path that feeds the corruption above. Consolidated into
`_is_already_compressed()` and applied at all three pinning sites.
- **`cache/compression_store.py`** — store-level guard: refuse to
persist a *bare* marker as `original_content` and log at ERROR, so a
future producer regression surfaces loudly instead of silently
converting "retrievable" into "gone". Deliberately narrow — originals
may legally *contain* markers (nested offloads); only a bare marker is
rejected.
- **Regression tests** —
`test_nested_table_markers_resolve_to_source_bytes` (asserts payloads
are verbatim-retrievable, not merely that a marker was emitted) and
`test_already_marked_content_is_not_re_offloaded`.

## Testing

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

### Test Output

```text
$ cargo build -p headroom-core
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 2m 37s

$ cargo test -p headroom-core --lib smart_crusher
test result: ok. 329 passed; 0 failed; 0 ignored; 0 measured; 583 filtered out; finished in 0.19s

$ python -m pytest tests/test_transforms/test_smart_crusher_ccr_roundtrip.py -q
16 passed in 0.68s

$ python -m pytest tests/test_ccr_row_drop_store_bridge.py tests/test_ccr_tool_injection.py -q
50 passed in 12.52s

$ python -m pytest tests/test_compression_store.py tests/test_lossless_mode.py -q
100 passed in 13.14s

$ ruff check headroom/transforms/content_router.py headroom/cache/compression_store.py \
      tests/test_transforms/test_smart_crusher_ccr_roundtrip.py
All checks passed!

$ mypy headroom/transforms/content_router.py headroom/cache/compression_store.py
Success: no issues found in 2 source files
```

## Real Behavior Proof

- **Environment:** macOS (Darwin 25.4.0), Python 3.12.6, headroom-ai
0.33.0 editable, `HEADROOM_CCR_BACKEND=memory`, Rust extension rebuilt
via `maturin develop --release`.
- **Exact command / steps:** compact a nested document — 5 rows whose
`detail` field is a stringified sub-array of 6 base64 blobs (1600 B
each) — then, for every `<<ccr:HASH>>` marker in the output, call
`ccr_get(HASH)` and check the payload is the verbatim source rather than
a marker.

```python
inner = [{"k": f"key{i}", "v": i, "tok": blob(1200)} for i in range(6)]
doc   = {"rows": [{"id": i, "detail": json.dumps(inner), "note": "x"} for i in range(5)]}
out   = SmartCrusher().compact_document_json(json.dumps(doc))
for h in re.findall(r"<<ccr:([0-9a-f]+)", out):
    payload = crusher.ccr_get(h)          # must be real bytes, not a marker
```

- **Observed result — BEFORE (on `main`):** all six payloads collapsed
into a single dead marker. The rendered sub-table was re-classified
opaque (`html`, because `<<` reads as a tag), offloaded again, and its
payload never stored — so the six inner hashes were erased from the
visible text *and* the outer hash resolved to nothing.

```text
{"rows":"[5]{detail:string,id:int,note:string}
         <<ccr:3fb1d44933da,html,289B>>,0,x
         <<ccr:3fb1d44933da,html,289B>>,1,x ..."}

3fb1d44933da -> RUST MISS          # unrecoverable — 6 × 1600 B gone
```

- **Observed result — AFTER (this branch):** the sub-table stays inline,
each blob keeps its own marker, and every marker resolves to verbatim
source.

```text
{"rows":"[5]{detail:string,id:int,note:string}
         \"[6]{k:string,tok:string,v:int}
         key0,\"\"<<ccr:955b1fed2ef7,base64,1.6KB>>\"\",0 ..."}

  6ad5846997f4: resolves, len=1600, is-verbatim-source=True
  78a0bd9364a7: resolves, len=1600, is-verbatim-source=True
  955b1fed2ef7: resolves, len=1600, is-verbatim-source=True
  a0cef69da7f0: resolves, len=1600, is-verbatim-source=True
  dfcde5e940c0: resolves, len=1600, is-verbatim-source=True
  e57c4e0a3ce8: resolves, len=1600, is-verbatim-source=True

RESULT: PASS — every marker resolves to real source bytes
```

## Notes for reviewers

- The issue also reports **function words dropped from retained prose**
(`is`, `a`, `the`) and **interleaved log output corrupting `headroom
doctor`'s table borders**. Those are separate defects on different paths
(extractive prose compression and log-handler buffering respectively)
and are **not** addressed here — this PR is scoped to the CCR
store/retrieve corruption. They should be tracked separately; the prose
one overlaps #2586.
- The `crusher.rs` prose-hook fix is on the Rust pipeline
(`json_offload`) rather than the Python proxy path, but it is the same
store-the-intermediate bug and was cheap to close while in the file.

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

@@ -21,6 +21,12 @@ use serde_json::Value;
 
 use super::ir::OpaqueKind;
 
+/// Prefix of every CCR marker this crate emits (`<<ccr:HASH,KIND,SIZE>>`,
+/// `<<ccr:HASH N_rows_offloaded>>`, `<<ccr:HASH>>`). Content carrying one is
+/// already-compressed output and must never be offloaded again — see
+/// [`classify_string`].
+const CCR_MARKER_PREFIX: &str = "<<ccr:";
+
 /// Per-cell classification result.
 #[derive(Debug, Clone, PartialEq)]
 pub enum CellClass {
@@ -86,6 +92,24 @@ pub fn classify_cell(value: &Value, cfg: &ClassifyConfig) -> CellClass {
 }
 
 fn classify_string(s: &str, cfg: &ClassifyConfig) -> CellClass {
+    // Never re-offload our own output (#2694). A string carrying a
+    // `<<ccr:…>>` marker is *compressed output*, not source content: the
+    // real bytes already live in the store under the marker's hash. Hashing
+    // it again would stash the MARKER as the new entry's "original", so
+    // `headroom_retrieve` hands the caller a placeholder instead of the
+    // data — silent, permanent loss of whatever the inner marker pointed at
+    // (its hash is no longer visible anywhere the model can reach). Same
+    // defect class as #1209 (tag placeholders persisted as originals).
+    //
+    // This is the shared choke point for both offload sites — the document
+    // walker (`walker::walk_string`) and the table compactor
+    // (`compactor::cell_from_value`) — so one guard covers both. A cell that
+    // is already a marker also renders no smaller, so keeping it Scalar
+    // costs nothing.
+    if s.contains(CCR_MARKER_PREFIX) {
+        return CellClass::Scalar;
+    }
+
     // 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

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

@@ -34,7 +34,7 @@ use std::sync::Arc;
 use serde_json::{Map, Value};
 
 use super::classifier::{classify_cell, CellClass};
-use super::compactor::{compact, CompactConfig};
+use super::compactor::{compact_with_store, CompactConfig};
 use super::formatter::{CsvSchemaFormatter, Formatter};
 use super::ir::OpaqueKind;
 use crate::ccr::CcrStore;
@@ -116,8 +116,12 @@ fn walk_array(items: Vec<Value>, ctx: &DocumentCompactor) -> Value {
     // becomes a rendered string before the outer table sees it.
     let inner: Vec<Value> = items.into_iter().map(|i| walk(i, ctx)).collect();
 
-    // Then try the array as a whole.
-    let c = compact(&inner, &ctx.config);
+    // Then try the array as a whole. `compact_with_store` (not the
+    // store-less `compact`) is required: the table compactor substitutes
+    // opaque cells with `<<ccr:HASH,…>>` markers, and without the store
+    // those markers point at a key nothing ever wrote — `headroom_retrieve`
+    // 404s and the cell's bytes are gone for good (#2694).
+    let c = compact_with_store(&inner, &ctx.config, ctx.ccr_store.as_ref());
     if c.was_compacted() {
         Value::String(ctx.formatter.format(&c))
     } else {

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

@@ -523,7 +523,11 @@ impl SmartCrusher {
                                 rows.extend(arr.iter().cloned());
                             }
 
-                            let result = self.crush_array(&rows, query_context, bias);
+                            // `arr` (not `rows`) is what the CCR marker must
+                            // resolve to: `rows` may already be prose-
+                            // compressed / marker-substituted by the hook.
+                            let result =
+                                self.crush_array_with_source(&rows, arr, query_context, bias);
                             // Lossless path won → substitute the array
                             // with the compacted string in place. This
                             // makes the lossless win visible to the
@@ -759,6 +763,27 @@ 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 {
+        self.crush_array_with_source(items, items, query_context, bias)
+    }
+
+    /// [`crush_array`](Self::crush_array), but hashing and stashing
+    /// `ccr_source` — not `items` — behind the row-drop marker.
+    ///
+    /// The two differ on the prose-hook path: there, `items` are rows whose
+    /// leaves have ALREADY been rewritten (prose extractively compressed,
+    /// opaque blobs swapped for `<<ccr:…>>` markers). Storing those as the
+    /// entry's "original" hands a retrieving caller compressed output rather
+    /// than the dropped rows — the data the marker promises is simply not in
+    /// the store (#2694, same defect class as #1209). `ccr_source` is the
+    /// pre-processing array, so the marker's hash and the stored bytes both
+    /// describe what the model actually lost.
+    fn crush_array_with_source(
+        &self,
+        items: &[Value],
+        ccr_source: &[Value],
+        query_context: &str,
+        bias: f64,
+    ) -> CrushArrayResult {
         let item_strings: Vec<String> = items
             .iter()
             .map(|i| serde_json::to_string(i).unwrap_or_default())
@@ -908,7 +933,9 @@ impl SmartCrusher {
             // same bytes get stored — eliminating a redundant tree clone
             // (`items.to_vec()`) and a redundant `serde_json::to_string`
             // pass that the previous version did per dropped array.
-            let canonical = canonical_array_json(items);
+            // `ccr_source` == `items` except on the prose-hook path, where
+            // it is the pre-processing array — see `crush_array_with_source`.
+            let canonical = canonical_array_json(ccr_source);
             let h = hash_canonical(&canonical);
             let marker = format!("<<ccr:{h} {dropped_count}_rows_offloaded>>");
             if let Some(store) = &self.ccr_store {

crates/headroom-core/tests/ccr_backends.rs

@@ -241,6 +241,14 @@ fn sqlite_get_refreshes_idle_ttl() {
 
 #[test]
 fn sqlite_max_lifetime_caps_sliding_window() {
+    // Timing note: the backend stores unix-SECONDS (`as_secs()` truncates)
+    // and purges on `last_accessed + ttl <= now`, so apparent elapsed time
+    // is `floor(t0 + s) - floor(t0)` — it rounds UP by nearly a second
+    // depending on where t0 lands within its second. Every margin here is
+    // therefore kept a full second clear of the boundary in both
+    // directions; a sub-second margin makes this test phase-dependent
+    // (the previous 1.5s-against-a-2s-window "still alive" assertion
+    // failed ~70% of runs whenever `frac(t0) >= 0.5`).
     let dir = tempfile::tempdir().expect("tempdir");
     let path = dir.path().join("ccr.sqlite");
     // Idle 2s with a 3s ceiling: constant access must not outlive t+3s.
@@ -248,14 +256,22 @@ fn sqlite_max_lifetime_caps_sliding_window() {
         SqliteCcrStore::open_with_ttls(&path, 2, 3).expect("open sqlite store with ceiling");
     let hash = compute_key(b"capped sqlite");
     store.put(&hash, "capped sqlite");
-    std::thread::sleep(Duration::from_millis(1_500));
+    // 0.5s: apparent elapsed is 0s or 1s — always under the 2s window.
+    std::thread::sleep(Duration::from_millis(500));
     assert_eq!(
         store.get(&hash).as_deref(),
         Some("capped sqlite"),
         "entry inside idle window and ceiling must hit"
     );
-    // Keep touching, but cross the 3s ceiling.
-    std::thread::sleep(Duration::from_millis(2_600));
+    // Keep touching, but cross the 3s ceiling. The touches must stay INSIDE
+    // the idle window or the entry dies of idleness and the assertion below
+    // passes without ever exercising the ceiling — the thing under test.
+    // 0.7s gaps read as at most 1s apparent, comfortably under the 2s idle
+    // window, while the 4 of them carry total age past 3s.
+    for _ in 0..4 {
+        std::thread::sleep(Duration::from_millis(700));
+        let _ = store.get(&hash);
+    }
     assert_eq!(
         store.get(&hash),
         None,

headroom/cache/compression_store.py

@@ -324,6 +324,36 @@ class CompressionStore:
             # deterministically under whichever function is in use.
             hash_key = hashlib.sha256(original.encode()).hexdigest()[:24]
 
+        # Refuse to persist a bare CCR marker as an entry's "original"
+        # (#2694). A marker is a *pointer* to content, never content: an
+        # entry like `hash=abc123 -> "<<ccr:abc123,base64,2.0KB>>"` answers a
+        # retrieve with the very placeholder the caller is trying to resolve,
+        # and (worse) can overwrite a good entry with a useless one. Any
+        # producer that gets here has lost the source bytes upstream, so fail
+        # loudly rather than silently converting "retrievable" into "gone".
+        # Narrow by design: only a *bare* marker is rejected. Legitimate
+        # originals may legally CONTAIN markers (nested offloads, a tool that
+        # echoed one), and refusing those would drop recoverable data.
+        #
+        # The rejected value is never echoed into the log. It is provably a
+        # bare marker here, but `original` is the store's credential-bearing
+        # payload in the general case (this issue was reported against an
+        # OAuth token), and an error path is exactly where that sort of leak
+        # survives review. `hash_key` already identifies the entry.
+        stripped = original.strip()
+        if stripped.startswith("<<ccr:") and stripped.endswith(">>") and "\n" not in stripped:
+            logger.error(
+                "CCR store: refusing to persist a bare retrieval marker as "
+                "original_content (hash=%s tool=%s strategy=%s len=%d) — the "
+                "producer lost the source bytes; retrieval for this hash will "
+                "miss instead of returning a placeholder",
+                hash_key,
+                tool_name,
+                compression_strategy,
+                len(stripped),
+            )
+            return hash_key
+
         entry = CompressionEntry(
             hash=hash_key,
             original_content=original,

headroom/transforms/content_router.py

@@ -116,6 +116,34 @@ _PROVIDER_KIND_RE = re.compile(r"^[a-z0-9_]{1,32}$")
 _PROVIDER_KIND_FALLBACK = "provider"
 
 
+# Every marker shape that means "this text is already compressed and the
+# real bytes live in the CCR store". The bracket forms come from
+# SmartCrusher's row-drop summary and read_lifecycle/read_maturation; the
+# `<<ccr:` form is emitted by the Rust opaque-blob and row-drop paths
+# (`<<ccr:HASH,KIND,SIZE>>`, `<<ccr:HASH N_rows_offloaded>>`, `<<ccr:HASH>>`).
+_ALREADY_COMPRESSED_MARKERS = (
+    "Retrieve more: hash=",
+    "Retrieve original: hash=",
+    "<<ccr:",
+)
+
+
+def _is_already_compressed(text: str) -> bool:
+    """True if ``text`` still carries a CCR retrieval marker.
+
+    Re-compressing such a block is never right. Beyond the prefix-cache
+    churn, the second pass treats the *compressed* text as source: a marker
+    that lands in a cell wide enough to be re-offloaded gets hashed and
+    stashed as the new entry's "original", so ``headroom_retrieve`` returns a
+    placeholder and the inner marker's hash — the only handle on the real
+    bytes — disappears from anywhere the model can see (#2694).
+
+    ``<<ccr:`` was missing from this check, which is how opaque-blob markers
+    (the base64/binary form) leaked back into the compressor.
+    """
+    return any(marker in text for marker in _ALREADY_COMPRESSED_MARKERS)
+
+
 def _estimate_tokens(text: str) -> int:
     """Size-proportional token estimate for section ratio decisions.
 
@@ -5015,7 +5043,7 @@ class ContentRouter(Transform):
             # (contains a CCR retrieval marker), skip recompression.
             # Recompressing would change byte content and break provider
             # prefix caching with no meaningful further reduction.
-            if "Retrieve more: hash=" in content or "Retrieve original: hash=" in content:
+            if _is_already_compressed(content):
                 result_slots[i] = message
                 route_counts.setdefault("already_compressed", 0)
                 route_counts["already_compressed"] += 1
@@ -5910,10 +5938,7 @@ class ContentRouter(Transform):
                     len(tool_text) > min_chars or self._has_lossless_fold(tool_text)
                 ):
                     # Compression pinning: skip already-compressed content
-                    if (
-                        "Retrieve more: hash=" in tool_text
-                        or "Retrieve original: hash=" in tool_text
-                    ):
+                    if _is_already_compressed(tool_text):
                         new_blocks.append(block)
                         if route_counts is not None:
                             route_counts.setdefault("already_compressed", 0)
@@ -5965,10 +5990,7 @@ class ContentRouter(Transform):
                     len(text_content) > min_chars or self._has_lossless_fold(text_content)
                 ):
                     # Pinning: skip already-compressed content
-                    if (
-                        "Retrieve more: hash=" in text_content
-                        or "Retrieve original: hash=" in text_content
-                    ):
+                    if _is_already_compressed(text_content):
                         new_blocks.append(block)
                         if route_counts is not None:
                             route_counts.setdefault("already_compressed", 0)

tests/test_transforms/test_smart_crusher_ccr_roundtrip.py

@@ -343,3 +343,68 @@ def test_distinct_payloads_have_distinct_hashes_and_separate_storage() -> None:
     # And they don't cross-contaminate.
     assert pa != b
     assert pb != a
+
+
+def test_nested_table_markers_resolve_to_source_bytes() -> None:
+    """Regression for #2694: every marker the walker emits must resolve.
+
+    Nested shape (rows whose field is a stringified sub-array of base64
+    blobs) hit two defects at once:
+
+    1. ``walk_array`` compacted via the store-LESS ``compact()``, so opaque
+       cells inside a compacted table got a ``<<ccr:HASH,...>>`` marker whose
+       payload was never written — retrieval 404'd forever.
+    2. The rendered sub-table (already full of markers) was itself long
+       enough to be re-classified opaque and offloaded again, so the CCR
+       entry's "original" was compressed output, and the inner hashes — the
+       only handle on the real bytes — vanished from the visible text.
+
+    Both collapsed 6 payloads into one dead marker. Assert the payloads are
+    verbatim-retrievable, not merely that a marker was emitted.
+    """
+    import re
+
+    from headroom.config import SmartCrusherConfig as PyConfig
+    from headroom.transforms.smart_crusher import SmartCrusher
+
+    crusher = SmartCrusher(PyConfig())
+    blobs = [
+        ("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" * 6) + f"{i:04d}"
+        for i in range(6)
+    ]
+    inner = [{"k": f"key{i}", "tok": b, "v": i} for i, b in enumerate(blobs)]
+    doc = {"rows": [{"id": i, "detail": json.dumps(inner), "note": "x"} for i in range(5)]}
+
+    out = crusher.compact_document_json(json.dumps(doc))
+
+    hashes = set(re.findall(r"<<ccr:([a-f0-9]+)", out))
+    assert hashes, f"expected retrieval markers, got: {out[:200]}"
+    for h in hashes:
+        payload = crusher.ccr_get(h)
+        assert payload is not None, f"marker <<ccr:{h}>> points at an unstored key (data loss)"
+        assert "<<ccr:" not in payload, (
+            f"<<ccr:{h}>> resolves to compressed output, not the original: {payload[:120]!r}"
+        )
+    # Every source blob is recoverable through some marker.
+    recovered = {crusher.ccr_get(h) for h in hashes}
+    assert set(blobs) <= recovered, "a source payload is unreachable from any emitted marker"
+
+
+def test_already_marked_content_is_not_re_offloaded() -> None:
+    """A cell that already carries a marker must never be offloaded again.
+
+    Re-offloading stores the MARKER as the new entry's original_content —
+    the exact corruption reported in #2694.
+    """
+    from headroom.config import SmartCrusherConfig as PyConfig
+    from headroom.transforms.smart_crusher import SmartCrusher
+
+    crusher = SmartCrusher(PyConfig())
+    # Long enough to clear the 256-byte opaque threshold, but it is our own
+    # output — a rendered row of markers, not source content.
+    marked = "\n".join(f"row{i},<<ccr:{i:012x},base64,1.6KB>>,ok" for i in range(12))
+    assert len(marked) > 256
+
+    out = crusher.compact_document_json(json.dumps({"table": marked}))
+
+    assert json.loads(out)["table"] == marked, "already-marked content was re-offloaded"