fix(read-lifecycle): persist STALE Read originals in the CCR store (#1488)

## Description

`read_lifecycle` emits STALE/SUPERSEDED Read markers containing
`Retrieve original: hash=...`, but `headroom_retrieve(hash)` 404s on
every such marker — the original content is never actually stored.

Affects the default config (`read_lifecycle=on`, `compress_stale=on`)
and the common Claude Code flow: read a file, edit it, then want the
prior content back.

**Root cause:** `ContentRouter.transform` instantiated
`ReadLifecycleManager` with
`compression_store=kwargs.get("compression_store")`, but no caller ever
sets that kwarg. `self.store` was always `None`, so `read_lifecycle.py`
emitted the marker with a SHA-256 hash but skipped the
`store.store(...)` call. Every other compressor (SmartCrusher, Kompress,
search/log/diff/code) resolves its store directly via
`get_compression_store()`.

Closes #

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

- `headroom/transforms/content_router.py`: inject a CCR store into
`ReadLifecycleManager` via an explicit `is None` check + guarded
`get_compression_store()` import (matches `smart_crusher.py`'s pattern).
Falls back to marker-only when the module is absent in stripped builds.
- `headroom/transforms/read_lifecycle.py`: wrap `store.store(...)` in
`try/except` with a precomputed fallback hash so a transient backend
failure can't break `compress()` (mirrors `read_maturation.py`). Pass
`explicit_hash=ccr_hash` to avoid double SHA-256 and keep marker/store
key in lockstep.
- `tests/test_transforms/test_read_lifecycle.py`: regression test
(`TestContentRouterIntegration`) that drives `headroom.compress()` and
asserts the STALE marker's hash resolves in the global CCR store.

## Testing

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

### Test Output

```text
$ HEADROOM_CCR_BACKEND=memory .venv/bin/python -m pytest tests/test_transforms/test_read_lifecycle.py -v
============================== 23 passed in 0.43s ==============================
```

## Real Behavior Proof

- Environment: Python 3.13, headroom-ai dev install (`uv sync --extra
dev`), `HEADROOM_CCR_BACKEND=memory`, Linux x86_64.
- Exact command / steps: Run `headroom.compress()` on a synthetic STALE
conversation (Read then Edit of the same file):
  ```python
  from headroom import compress
  result = compress([
{"role": "assistant", "content": [{"type": "tool_use", "id": "t1",
"name": "Read",
"input": {"file_path": "/tmp/foo.txt"}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id":
"t1",
                                    "content": "source line\n" * 500}]},
{"role": "assistant", "content": [{"type": "tool_use", "id": "t2",
"name": "Edit",
"input": {"file_path": "/tmp/foo.txt"}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id":
"t2",
                                    "content": "edited"}]},
  ], model="claude-sonnet-4-5-20250929")
  ```
  then `get_compression_store().retrieve(<hash-from-marker>)`.
- Observed result: post-fix `retrieve(hash)` returns HIT (`tool=Read`,
`strategy=read_lifecycle:stale`); pre-fix it returned MISS (the bug).
Full log:
  ```text
transforms_applied: ['read_lifecycle:stale:/tmp/foo.txt',
'router:excluded:tool', 'router:excluded:tool']
  hashes from markers: ['3fbd603ecf1bcf50a86650d2']
  store backend: InMemoryBackend
retrieve(3fbd603ecf1bcf50a86650d2) -> HIT tool=Read
strategy=read_lifecycle:stale
  ```
- Not tested: SQLite backend persistence across processes; Rust `_core`
extension code path; OpenAI / Gemini providers; Claude Code live (proxy
+ MCP server end-to-end).

## 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
(internal fix, no public API change)
- [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 — leaving to
maintainers' convention

## Additional Notes

- No existing issue. #389 describes the same symptom class with a
different root cause (SmartCrusher row-drop CCR bridge); it explicitly
lists `read_lifecycle.py` as a producer that populates the store — this
PR makes that claim true.
- Commits: `dde42478` (initial fix) → `55a0dfde` (Copilot round 1: `is
None` + import guard + best-effort `store.store()`) → `2e8c41a2`
(Copilot round 2: regression test + `explicit_hash`).

headroom/transforms/content_router.py

@@ -2419,9 +2419,20 @@ class ContentRouter(Transform):
         if self.config.read_lifecycle.enabled:
             from .read_lifecycle import ReadLifecycleManager
 
+            # is None (not truthiness) so falsy test doubles are honored;
+            # guarded import keeps read_lifecycle running in stripped builds.
+            injected_store = kwargs.get("compression_store")
+            if injected_store is None:
+                try:
+                    from ..cache.compression_store import get_compression_store
+
+                    injected_store = get_compression_store()
+                except ImportError:
+                    pass
+
             lifecycle_mgr = ReadLifecycleManager(
                 self.config.read_lifecycle,
-                compression_store=kwargs.get("compression_store"),
+                compression_store=injected_store,
             )
             lifecycle_result = lifecycle_mgr.apply(
                 messages,

headroom/transforms/read_lifecycle.py

@@ -474,21 +474,25 @@ class ReadLifecycleManager:
         if content_bytes < self.config.min_size_bytes:
             return False, content, None
 
-        # Store original in CCR if available
-        ccr_hash = None
+        # Best-effort CCR persistence (mirrors read_maturation.py): a store
+        # failure must not break compress().
+        ccr_hash = hashlib.sha256(content.encode()).hexdigest()[:24]
         if self.store is not None:
-            ccr_hash = self.store.store(
-                original=content,
-                compressed="",
-                tool_name="Read",
-                tool_call_id=classification.tool_call_id,
-                compression_strategy=f"read_lifecycle:{classification.state.value}",
-            )
-
-        # Generate marker
-        if ccr_hash is None:
-            # No CCR store — generate a content hash for reference
-            ccr_hash = hashlib.sha256(content.encode()).hexdigest()[:24]
+            try:
+                ccr_hash = self.store.store(
+                    original=content,
+                    compressed="",
+                    tool_name="Read",
+                    tool_call_id=classification.tool_call_id,
+                    compression_strategy=f"read_lifecycle:{classification.state.value}",
+                    explicit_hash=ccr_hash,
+                )
+            except Exception as e:  # noqa: BLE001 - storage failure must not break the request
+                logger.warning(
+                    "read_lifecycle: CCR store failed for %s: %s",
+                    classification.tool_call_id,
+                    e,
+                )
 
         file_display = classification.file_path or "unknown"
 

tests/test_transforms/test_read_lifecycle.py

@@ -610,3 +610,92 @@ class TestNoFilePathHandling:
         # Can't match file_path, so Read is not classified at all
         assert result.reads_total == 0
         assert result.messages[1]["content"] == LARGE_CONTENT
+
+
+class TestContentRouterIntegration:
+    """Regression: ContentRouter.transform must wire a real CCR store into
+    ReadLifecycleManager so STALE Read markers resolve via headroom_retrieve."""
+
+    def test_stale_read_marker_retrievable_via_compress(self, monkeypatch):
+        import re
+
+        # Force an in-memory backend so the test is hermetic.
+        monkeypatch.setenv("HEADROOM_CCR_BACKEND", "memory")
+
+        from headroom import compress
+        from headroom.cache.compression_store import (
+            get_compression_store,
+            reset_compression_store,
+        )
+
+        reset_compression_store()
+        try:
+            large_content = "source line\n" * 500  # above read_lifecycle min_size_bytes
+            messages = [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "tool_use",
+                            "id": "t1",
+                            "name": "Read",
+                            "input": {"file_path": "/tmp/foo.txt"},
+                        }
+                    ],
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "tool_result",
+                            "tool_use_id": "t1",
+                            "content": large_content,
+                        }
+                    ],
+                },
+                # Edit the same file -> the Read above becomes STALE.
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "tool_use",
+                            "id": "t2",
+                            "name": "Edit",
+                            "input": {"file_path": "/tmp/foo.txt"},
+                        }
+                    ],
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "tool_result",
+                            "tool_use_id": "t2",
+                            "content": "edited",
+                        }
+                    ],
+                },
+            ]
+
+            result = compress(messages, model="claude-sonnet-4-5-20250929")
+
+            hashes: list[str] = []
+            for m in result.messages:
+                content = m.get("content")
+                if isinstance(content, list):
+                    for b in content:
+                        if isinstance(b, dict) and b.get("type") == "tool_result":
+                            s = b.get("content", "")
+                            if isinstance(s, str):
+                                hashes.extend(re.findall(r"hash=([a-f0-9]+)", s))
+            assert hashes, "Expected a STALE Read marker with a hash"
+
+            store = get_compression_store()
+            entry = store.retrieve(hashes[0])
+            assert entry is not None, "STALE Read marker hash not in CCR store"
+            assert entry.tool_name == "Read"
+            assert entry.compression_strategy == "read_lifecycle:stale"
+        finally:
+            # Drop the memory-backend singleton so later tests in the suite
+            # see the env-driven default again.
+            reset_compression_store()