fix(ccr): report embedded hashes from compress endpoint (#717)

## Description

Fixes `/v1/compress` so its `ccr_hashes` response includes retrievable
CCR hashes embedded in compressed message content, including row-drop
and recursive JSON markers that may not be present in
`TransformResult.markers_inserted`.

The original PR also changed query-based JSON row search. Current `main`
intentionally made CCR retrieval a hash-only, full-content lookup in
#1532, so that obsolete half is not restored. This reconciliation
preserves the reporting bug fix without reversing the current retrieval
contract.

## 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 causes existing functionality
to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- extract 12–24 character CCR hashes from `Retrieve more`, `Retrieve
original`, and `<<ccr:...>>` markers
- scan both transform marker metadata and nested rendered message values
- preserve stable encounter order and deduplicate case-insensitively
- exclude non-retrieval transform metadata such as tool digests and
stable-prefix hashes
- return the normalized hashes from `/v1/compress`
- add helper-level and endpoint-level regression coverage

## Testing

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

### Test Output

```text
$ uv run --extra dev pytest tests/test_proxy_compress_endpoint.py -q
50 passed, 1 warning in 6.54s

$ uv run --extra dev ruff check headroom/proxy/handlers/openai.py tests/test_proxy_compress_endpoint.py
All checks passed!

$ uv run --extra dev ruff format --check headroom/proxy/handlers/openai.py tests/test_proxy_compress_endpoint.py
2 files already formatted

$ git diff --check
# no output
```

## Real Behavior Proof

- Environment: macOS, Python 3.12.13, current `main` at `7940c05e`,
project native extension built by `uv`
- Exact command / steps: ran the complete `/v1/compress` endpoint test
module, including a mocked pipeline response containing an embedded
row-drop marker but only unrelated tool-digest marker metadata
- Observed result: endpoint returned exactly the embedded retrievable
hash; helper coverage also proved nested markers, case normalization,
deduplication, stable ordering, and exclusion of unrelated metadata
- Not tested: full repository test and CI matrix; GitHub CI covers the
broader matrix

## 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 the non-obvious marker filtering behavior
- [x] Documentation is unchanged because the public response contract is
corrected, not expanded
- [x] My changes generate no new warnings
- [x] I have added tests that prove the fix is effective
- [x] New and existing endpoint tests pass locally
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from the Conventional Commit PR title

## Screenshots (if applicable)

Not applicable. This changes a JSON API response and tests, with no
graphical UI changes.

## Additional Notes

The query-based JSON row-search changes from the original branch were
made obsolete by #1532 and are deliberately excluded rather than
reviving a retired API behavior. The original contributor remains the
commit author for the reconciled fix.

---------

Co-authored-by: JD Davis <jd@jds-macbook-air.tail2a279.ts.net>

headroom/proxy/handlers/openai.py

@@ -12,6 +12,7 @@ import hashlib
 import json
 import logging
 import os
+import re
 import threading
 import time
 import uuid
@@ -94,6 +95,45 @@ _OPENAI_RESPONSES_UNIT_CACHE_INIT_LOCK = threading.RLock()
 _OPENAI_RESPONSES_UNIT_EXECUTOR_LOCK = threading.RLock()
 _OPENAI_RESPONSES_UNIT_EXECUTOR: ThreadPoolExecutor | None = None
 _CODEX_WS_COMPRESSION_TIMEOUT_SECONDS = 5.0
+_CCR_HASH_RE = re.compile(
+    r"(?:Retrieve (?:more|original): hash=|<<ccr:)([a-fA-F0-9]{12,24})(?=[^a-fA-F0-9]|$)"
+)
+_BARE_CCR_HASH_RE = re.compile(r"[a-fA-F0-9]{12,24}")
+
+
+def _response_ccr_hashes(messages: list[dict[str, Any]], markers: list[str]) -> list[str]:
+    """Return the distinct retrievable CCR hashes exposed by a response.
+
+    ``TransformResult.markers_inserted`` is not a hash-only collection: it can
+    also contain tool-digest and stable-prefix metadata. Extract only supported
+    CCR retrieval markers, then scan the rendered messages because row-drop and
+    recursive JSON paths can embed a marker without registering it separately.
+    """
+    hashes: list[str] = []
+    seen: set[str] = set()
+
+    def collect(value: Any, *, allow_bare_hash: bool = False) -> None:
+        if isinstance(value, str):
+            candidates = [value] if allow_bare_hash and _BARE_CCR_HASH_RE.fullmatch(value) else []
+            candidates.extend(match.group(1) for match in _CCR_HASH_RE.finditer(value))
+            for candidate in candidates:
+                normalized = candidate.lower()
+                if normalized not in seen:
+                    seen.add(normalized)
+                    hashes.append(normalized)
+            return
+        if isinstance(value, dict):
+            for child in value.values():
+                collect(child)
+            return
+        if isinstance(value, list):
+            for child in value:
+                collect(child)
+
+    for marker in markers:
+        collect(marker, allow_bare_hash=True)
+    collect(messages)
+    return hashes
 
 
 def _codex_ws_compression_timeout_seconds() -> float:
@@ -8978,6 +9018,7 @@ class OpenAIHandlerMixin:
                 ),
                 timeout=COMPRESSION_TIMEOUT_SECONDS,
             )
+            ccr_hashes = _response_ccr_hashes(result.messages, result.markers_inserted)
 
             tokens_before = result.tokens_before
             tokens_after = result.tokens_after
@@ -9025,7 +9066,7 @@ class OpenAIHandlerMixin:
                     ),
                     "transforms_applied": result.transforms_applied,
                     "transforms_summary": result.transforms_summary,
-                    "ccr_hashes": result.markers_inserted,
+                    "ccr_hashes": ccr_hashes,
                 }
             )
         except TimeoutError:

tests/test_platform_stabilization_functional.py

@@ -56,6 +56,7 @@ def test_v1_compress_success_reports_actual_metrics(monkeypatch) -> None:
     proxy = app.state.proxy
     request_messages = [{"role": "user", "content": "summarize this repeated payload"}]
     compressed_messages = [{"role": "user", "content": "summary payload"}]
+    ccr_hash = "abc123def4567890abc123de"
 
     def fake_apply(**kwargs):
         assert kwargs["messages"] == request_messages
@@ -65,7 +66,7 @@ def test_v1_compress_success_reports_actual_metrics(monkeypatch) -> None:
             tokens_before=100,
             tokens_after=40,
             transforms_applied=["test:compress"],
-            markers_inserted=["marker-1"],
+            markers_inserted=[ccr_hash],
         )
 
     # The default /v1/compress mode runs a marker-free pipeline derived from
@@ -88,7 +89,7 @@ def test_v1_compress_success_reports_actual_metrics(monkeypatch) -> None:
     assert body["compression_ratio"] == 0.4
     assert body["transforms_applied"] == ["test:compress"]
     assert body["transforms_summary"] == {"test:compress": 1}
-    assert body["ccr_hashes"] == ["marker-1"]
+    assert body["ccr_hashes"] == [ccr_hash]
 
 
 def test_v1_compress_timeout_fails_open_quickly(monkeypatch) -> None:

tests/test_proxy_compress_endpoint.py

@@ -134,6 +134,45 @@ class TestCompressEndpointBasic:
         assert data["tokens_saved"] >= 0
         assert data["compression_ratio"] > 0
 
+    def test_response_ccr_hashes_extracts_only_retrievable_hashes(self):
+        """Embedded CCR markers are reported without unrelated transform metadata."""
+        from headroom.proxy.handlers.openai import _response_ccr_hashes
+
+        messages = [
+            {
+                "role": "tool",
+                "content": ("[100 rows compressed. Retrieve more: hash=abc123def4567890abc123de]"),
+            },
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": "<<ccr:feedface00112233 10_rows_offloaded>>",
+                    },
+                    {
+                        "type": "text",
+                        "text": "Retrieve original: hash=ABC123DEF4567890ABC123DE",
+                    },
+                ],
+            },
+        ]
+
+        hashes = _response_ccr_hashes(
+            messages,
+            [
+                "deadbeef0000000000000000",
+                "<headroom:tool_digest sha256=1234567890abcdef>",
+                "stable_prefix_hash:feedface00112233",
+            ],
+        )
+
+        assert hashes == [
+            "deadbeef0000000000000000",
+            "abc123def4567890abc123de",
+            "feedface00112233",
+        ]
+
     def test_bypass_header_returns_uncompressed(self, client):
         """X-Headroom-Bypass header should skip compression."""
         messages = [
@@ -290,6 +329,38 @@ class TestCompressEndpointCompression:
         assert outcome.transforms_applied == ("test_transform",)
         assert outcome.total_latency_ms >= 0
 
+    def test_response_reports_embedded_ccr_hashes(self, client, monkeypatch):
+        """The endpoint reports a CCR marker even when the transform omitted its registry."""
+        proxy = client.app.state.proxy
+        ccr_hash = "abc123def4567890abc123de"
+        result = SimpleNamespace(
+            messages=[
+                {
+                    "role": "tool",
+                    "content": f"<<ccr:{ccr_hash} 10_rows_offloaded>>",
+                }
+            ],
+            tokens_before=12,
+            tokens_after=7,
+            transforms_applied=["test_transform"],
+            transforms_summary={"test_transform": 1},
+            markers_inserted=["<headroom:tool_digest sha256=1234567890abcdef>"],
+        )
+        monkeypatch.setattr(
+            proxy,
+            "_run_compression_in_executor",
+            AsyncMock(return_value=result),
+        )
+        monkeypatch.setattr(proxy, "_record_request_outcome", AsyncMock())
+
+        response = client.post(
+            "/v1/compress",
+            json={"messages": [{"role": "user", "content": "compress me"}], "model": "gpt-4"},
+        )
+
+        assert response.status_code == 200
+        assert response.json()["ccr_hashes"] == [ccr_hash]
+
     def test_compression_error_records_failed_request(self, client, monkeypatch):
         """A hard compression failure should increment failed metrics."""
         proxy = client.app.state.proxy