feat: attribute reread waste to over-compression via marker check (#901)

## Description

Fixes #899. The `reread` signal (#853/#854) counts re-served tool
results but cannot answer the question that motivated it: **did Headroom
cause the re-read?** A re-read after an intact first serve is agent
behavior; a re-read after Headroom markerized the first serve is
over-compression cost. This PR splits the signal so the actionable part
is visible.

Request-local, no store lookups: the client resends full history each
turn and the pipeline recompresses it deterministically, so the current
request already holds the evidence. `TransformPipeline.apply` passes
`current_messages` into `parse_messages(compressed_messages=...)`. For
each counted reread group, if the transformed copy of the **first
serve** carries a CCR retrieval marker and its original text is gone,
the group's counted repeats go into `reread_compressed_tokens`. Lossless
reshaping (no marker) is deliberately not attributed.

Closes #899.

## Type of Change

- [x] New feature (non-breaking change that adds functionality)

## Changes Made

- `parser.py`: `parse_messages` gains an optional `compressed_messages`
param; the content-hash reread loop accumulates per-group
`counted_tokens` and attributes them to `reread_compressed_tokens` when
the first serve's transformed copy carries a CCR marker
(`CCR_RETRIEVAL_MARKER_RE`, kept local to avoid a transforms import
cycle).
- `transforms/pipeline.py`: pass `current_messages` (post-transform
copy) into the existing waste-detection `parse_messages` call.
- `config.py`: new `reread_compressed_tokens` WasteSignals field;
`dashboard.html` + `reporting/generator.py` surface it.
- Tests: `tests/test_reread_attribution.py` + WasteSignals contract
update.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] New tests added for new functionality
- [x] New and existing unit tests pass locally with my changes

### Test Output

```text
$ pytest tests/test_reread_attribution.py tests/test_parser.py tests/test_gemini_function_response_waste.py tests/test_codex_responses_waste_signals.py -q
122 passed in 1.50s

$ pytest tests/ -k "waste or pipeline or reporting or config or reread" -q
348 passed, 33 skipped, 6010 deselected
# (1 unrelated env-dependent failure: test_proxy_gemini_native_integration::test_generation_config — 404, reproduces on main without these changes; needs a Gemini key locally)

$ ruff check headroom/parser.py headroom/transforms/pipeline.py
All checks passed!
```

## Real Behavior Proof

- Environment: local macOS, repo .venv, Python 3.11.9
- Exact command / steps: rebased onto current main to resolve conflicts
with #909 (merged), then ran the reread + parser + waste suites above
- Observed result: a reread whose first serve is markerized attributes
to `reread_compressed_tokens`; an intact first serve and a lossless
(no-marker) reshape do not. #909's re-issued-call detection (same call,
different bytes) continues to count and dedup correctly alongside it —
all 122 targeted tests pass.
- Not tested: live proxy traffic; the one gemini-native route test above
(environmental 404, not introduced here).

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Additional Notes

**Rebased onto current main after #909 merged.** #909 added a
re-issued-call reread pass *after* the original content-hash loop this
PR modifies — the conflict was textual/adjacent, not a re-architecture.
Resolution preserves #909's `counted_results` dedup contract and leaves
its new pass unchanged; #901's attribution stays scoped to the
content-hash groups it was reviewed against (attributing #909's call-key
pass too would be a separate follow-up). The diff differs from the prior
approval only by this reshape — worth a quick re-glance.

headroom/config.py

@@ -537,6 +537,11 @@ class WasteSignals:
     dynamic_date_tokens: int = 0  # Dynamic dates in system prompt
     repetition_tokens: int = 0  # Repeated content
     reread_tokens: int = 0  # Tool results re-served after already appearing earlier
+    # Subset of reread_tokens whose first serve was compressed away (CCR
+    # marker left in its place) — re-reads attributable to over-compression
+    # rather than agent behavior (#899). Excluded from total() because the
+    # same tokens are already counted in reread_tokens.
+    reread_compressed_tokens: int = 0
 
     def total(self) -> int:
         """Total waste tokens detected."""
@@ -560,6 +565,7 @@ class WasteSignals:
             "dynamic_date": self.dynamic_date_tokens,
             "repetition": self.repetition_tokens,
             "reread": self.reread_tokens,
+            "reread_compressed": self.reread_compressed_tokens,
         }
 
 

headroom/dashboard/templates/dashboard.html

@@ -2289,6 +2289,7 @@
                         dynamic_date: 'Dynamic Dates',
                         repetition: 'Repetition',
                         reread: 'Re-read Tool Results',
+                        reread_compressed: 'Re-read After Compression',
                     };
                     return labels[signal] || signal;
                 },
@@ -2302,6 +2303,7 @@
                         dynamic_date: 'bg-purple-500',
                         repetition: 'bg-pink-500',
                         reread: 'bg-teal-500',
+                        reread_compressed: 'bg-rose-500',
                     };
                     return colors[signal] || 'bg-gray-500';
                 },
@@ -2315,6 +2317,7 @@
                         dynamic_date: 'bg-purple-500/20 text-purple-400',
                         repetition: 'bg-pink-500/20 text-pink-400',
                         reread: 'bg-teal-500/20 text-teal-400',
+                        reread_compressed: 'bg-rose-500/20 text-rose-400',
                     };
                     return colors[signal] || 'bg-gray-500/20 text-gray-400';
                 },

headroom/parser.py

@@ -24,6 +24,11 @@ JSON_BLOCK_PATTERN = re.compile(r"\{[\s\S]{500,}\}")
 # exit codes) and are not evidence of a re-read.
 REREAD_MIN_TOKENS = 50
 
+# Canonical CCR retrieval-marker shapes. Mirrors the alternation in
+# transforms/compression_units._CCR_MARKER_RE; kept local because the parser
+# is a base module and importing from transforms would create a cycle.
+CCR_RETRIEVAL_MARKER_RE = re.compile(r"Retrieve more: hash=|Retrieve original: hash=|<<ccr:[^>]+>>")
+
 # Repeats this close (in message positions) to the previous serve are
 # polling, not re-reads. Consecutive tool turns sit 2 apart (the
 # assistant tool_use message lies between results); 3 also absorbs a
@@ -336,6 +341,7 @@ def parse_message_to_blocks(
 def parse_messages(
     messages: list[dict[str, Any]],
     tokenizer: Tokenizer,
+    compressed_messages: list[dict[str, Any]] | None = None,
 ) -> tuple[list[Block], dict[str, int], WasteSignals]:
     """
     Parse all messages into blocks with analysis.
@@ -343,6 +349,11 @@ def parse_messages(
     Args:
         messages: List of message dicts.
         tokenizer: Tokenizer instance for token counting.
+        compressed_messages: Optional post-transform copy of the same
+            messages. When provided (and the message count matches), reread
+            waste is additionally attributed: repeats whose first serve was
+            replaced by a CCR retrieval marker count into
+            ``reread_compressed_tokens`` (#899).
 
     Returns:
         Tuple of (blocks, block_breakdown, total_waste_signals)
@@ -374,6 +385,7 @@ def parse_messages(
     for block in all_blocks:
         if block.kind == "tool_result" and block.tokens_est >= REREAD_MIN_TOKENS:
             reread_groups.setdefault(block.content_hash, []).append(block)
+    attribute = compressed_messages is not None and len(compressed_messages) == len(messages)
     for group in reread_groups.values():
         # The message that first served the content is the original; only
         # copies appearing in *later* messages are re-reads. Duplicates
@@ -385,14 +397,34 @@ def parse_messages(
         # repeats advance the baseline without counting, so a long polling
         # chain never accumulates waste.
         prev_index = group[0].source_index
+        counted_tokens = 0
         for block in group:
             if block.source_index == prev_index:
                 continue
             is_polling = block.source_index - prev_index <= REREAD_ADJACENT_GAP
             prev_index = block.source_index
             if not is_polling:
-                total_waste.reread_tokens += block.tokens_est
+                counted_tokens += block.tokens_est
                 counted_results.add(id(block))
+        if not counted_tokens:
+            continue
+        total_waste.reread_tokens += counted_tokens
+        # Over-compression attribution (#899): if the transformed copy of the
+        # first serve carries a CCR retrieval marker and its original text is
+        # gone, the model never saw the full first serve — the repeats are
+        # attributable to compression. Lossless reshaping (no marker) is
+        # deliberately not attributed: the model saw all the data, so the
+        # re-read is agent behavior.
+        if attribute and compressed_messages is not None:
+            first = group[0]
+            transformed_blocks = parse_message_to_blocks(
+                compressed_messages[first.source_index], first.source_index, tokenizer
+            )
+            transformed_text = "\n".join(b.text for b in transformed_blocks)
+            if CCR_RETRIEVAL_MARKER_RE.search(transformed_text) and (
+                first.text not in transformed_text
+            ):
+                total_waste.reread_compressed_tokens += counted_tokens
 
     # Re-issued-call detection: the agent invoking the same tool with the
     # same arguments again is a re-fetch even when the result bytes differ

headroom/reporting/generator.py

@@ -396,6 +396,7 @@ def _build_waste_histogram(
         "whitespace": 0,
         "dynamic_date": 0,
         "reread": 0,
+        "reread_compressed": 0,
         "history_bloat": 0,
     }
 
@@ -415,8 +416,12 @@ def _build_waste_histogram(
             # Subtract known waste types. "reread" is excluded: it measures
             # over-compression cost (content the agent re-fetched), not
             # waste removed by compression, so it doesn't explain any part
-            # of tokens_saved.
-            known_waste = sum(v for k, v in waste.items() if k != "reread")
+            # of tokens_saved. "reread_compressed" is a subset of "reread"
+            # (#899) and is excluded for the same reason — counting it would
+            # also double-subtract.
+            known_waste = sum(
+                v for k, v in waste.items() if k not in ("reread", "reread_compressed")
+            )
             history_bloat = max(0, tokens_saved - known_waste)
             totals["history_bloat"] += history_bloat
 
@@ -430,6 +435,7 @@ def _build_waste_histogram(
         "whitespace": "Whitespace",
         "dynamic_date": "Dynamic Dates",
         "reread": "Re-served Tool Results",
+        "reread_compressed": "Re-served After Compression",
         "history_bloat": "History Bloat",
     }
 

headroom/transforms/pipeline.py

@@ -443,7 +443,16 @@ class TransformPipeline:
                 try:
                     from ..parser import parse_messages
 
-                    _, _, waste_signals = parse_messages(waste_messages or messages, tokenizer)
+                    # current_messages (the post-transform copy) enables reread
+                    # attribution: repeats whose first serve was markerized by
+                    # this pipeline run count into reread_compressed_tokens
+                    # (#899). The length guard in parse_messages makes the
+                    # waste_messages path (different indexing) a safe no-op.
+                    _, _, waste_signals = parse_messages(
+                        waste_messages or messages,
+                        tokenizer,
+                        compressed_messages=current_messages,
+                    )
                     if waste_signals.total() == 0:
                         waste_signals = None
                 except Exception:

tests/test_config.py

@@ -266,6 +266,7 @@ class TestWasteSignals:
             "dynamic_date": 10,
             "repetition": 15,
             "reread": 30,
+            "reread_compressed": 0,
         }
         assert signals.to_dict() == expected
 
@@ -274,7 +275,7 @@ class TestWasteSignals:
         signals = WasteSignals()
         result = signals.to_dict()
         assert all(v == 0 for v in result.values())
-        assert len(result) == 7
+        assert len(result) == 8
 
 
 class TestCachePrefixMetrics:

tests/test_reread_attribution.py

@@ -0,0 +1,173 @@
+"""Over-compression attribution for reread waste (issue #899).
+
+``parse_messages(compressed_messages=...)`` splits the existing ``reread``
+signal: repeats whose first serve was replaced by a CCR retrieval marker in
+the transformed output count into ``reread_compressed_tokens`` — re-reads
+attributable to Headroom rather than agent behavior. Lossless reshaping
+(no marker) and intact first serves are deliberately not attributed.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from headroom import OpenAIProvider, Tokenizer
+from headroom.config import HeadroomConfig, WasteSignals
+from headroom.parser import parse_messages
+from headroom.transforms.pipeline import TransformPipeline
+
+_provider = OpenAIProvider()
+
+
+@pytest.fixture
+def tokenizer() -> Tokenizer:
+    return Tokenizer(_provider.get_token_counter("gpt-4o"), "gpt-4o")
+
+
+def _uniform_rows(rows: int = 200) -> str:
+    return json.dumps(
+        [{"id": i, "name": f"item_{i}", "status": "ok", "score": i * 3.14} for i in range(rows)]
+    )
+
+
+_MARKER = "[200 items compressed to 12. Retrieve more: hash=abc123def4567890abcdef12]"
+
+
+def _conversation(first_serve: str, repeat: str) -> list[dict]:
+    """First serve at index 1, repeat at index 7 (gap 6 > REREAD_ADJACENT_GAP)."""
+    filler = [{"role": "user", "content": f"step {i}"} for i in range(5)]
+    return [
+        {"role": "user", "content": "read the data"},
+        {"role": "tool", "content": first_serve},
+        *filler,
+        {"role": "tool", "content": repeat},
+    ]
+
+
+class TestRereadAttribution:
+    def test_markerized_first_serve_attributes(self, tokenizer):
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+        compressed = [dict(m) for m in messages]
+        compressed[1] = {"role": "tool", "content": _MARKER}
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_tokens > 0
+        assert waste.reread_compressed_tokens == waste.reread_tokens
+
+    def test_intact_first_serve_not_attributed(self, tokenizer):
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+
+        _, _, waste = parse_messages(
+            messages, tokenizer, compressed_messages=[dict(m) for m in messages]
+        )
+        assert waste.reread_tokens > 0
+        assert waste.reread_compressed_tokens == 0
+
+    def test_lossless_reshape_without_marker_not_attributed(self, tokenizer):
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+        compressed = [dict(m) for m in messages]
+        # CSV-style compaction: content reshaped, all data retained, no marker.
+        compressed[1] = {"role": "tool", "content": "id,name,status,score\n0,item_0,ok,0.0"}
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_tokens > 0
+        assert waste.reread_compressed_tokens == 0
+
+    def test_marker_with_original_still_present_not_attributed(self, tokenizer):
+        # Marker appended but full original retained (e.g. partial compression
+        # of a different span in the same message) — model saw everything.
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+        compressed = [dict(m) for m in messages]
+        compressed[1] = {"role": "tool", "content": content + "\n" + _MARKER}
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_compressed_tokens == 0
+
+    def test_message_count_mismatch_skips_attribution(self, tokenizer):
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+        compressed = [dict(m) for m in messages]
+        compressed[1] = {"role": "tool", "content": _MARKER}
+        compressed.pop(0)
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_tokens > 0
+        assert waste.reread_compressed_tokens == 0
+
+    def test_default_no_compressed_messages(self, tokenizer):
+        content = _uniform_rows()
+        _, _, waste = parse_messages(_conversation(content, content), tokenizer)
+        assert waste.reread_tokens > 0
+        assert waste.reread_compressed_tokens == 0
+
+    def test_polling_repeats_not_attributed(self, tokenizer):
+        # Adjacent repeats (gap <= REREAD_ADJACENT_GAP) are polling, not
+        # rereads — attribution never runs for groups with no counted waste.
+        content = _uniform_rows()
+        messages = [
+            {"role": "tool", "content": content},
+            {"role": "user", "content": "poll"},
+            {"role": "tool", "content": content},
+        ]
+        compressed = [dict(m) for m in messages]
+        compressed[0] = {"role": "tool", "content": _MARKER}
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_tokens == 0
+        assert waste.reread_compressed_tokens == 0
+
+    def test_ccr_inline_marker_form_attributes(self, tokenizer):
+        content = _uniform_rows()
+        messages = _conversation(content, content)
+        compressed = [dict(m) for m in messages]
+        compressed[1] = {"role": "tool", "content": "<<ccr:a703e0aaa98f,string,1.1KB>>"}
+
+        _, _, waste = parse_messages(messages, tokenizer, compressed_messages=compressed)
+        assert waste.reread_compressed_tokens == waste.reread_tokens > 0
+
+
+class TestWasteSignalsContract:
+    def test_to_dict_exports_reread_compressed(self):
+        ws = WasteSignals(reread_tokens=100, reread_compressed_tokens=60)
+        d = ws.to_dict()
+        assert d["reread"] == 100
+        assert d["reread_compressed"] == 60
+
+    def test_total_excludes_reread_compressed(self):
+        # reread_compressed is a subset of reread — adding it to total()
+        # would double count.
+        ws = WasteSignals(reread_tokens=100, reread_compressed_tokens=60)
+        assert ws.total() == 100
+
+
+class TestPipelineAttribution:
+    def test_pipeline_passes_compressed_messages(self, tokenizer):
+        # End-to-end through TransformPipeline.apply: a large duplicated tool
+        # result far from its first serve produces reread waste, and
+        # reread_compressed is consistent (either 0 or the full group —
+        # never more than reread).
+        content = _uniform_rows(400)
+        filler = [{"role": "user", "content": f"working on step {i}"} for i in range(5)]
+        messages = [
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "tool", "content": content},
+            *filler,
+            {"role": "tool", "content": content},
+            {"role": "user", "content": "continue"},
+        ]
+        result = TransformPipeline(HeadroomConfig()).apply(
+            [dict(m) for m in messages], model="gpt-4o", model_limit=128000
+        )
+        assert result.waste_signals is not None
+        assert result.waste_signals.reread_tokens > 0
+        assert (
+            0
+            <= result.waste_signals.reread_compressed_tokens
+            <= (result.waste_signals.reread_tokens)
+        )