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>
2026-08-11 21:27:49 -07:00
parent a5b0a8f4cc
commit 685ebe457d
3 changed files with 116 additions and 3 deletions
@@ -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:
@@ -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:
@@ -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