From f9285766dda77b116c7834165849264e55339720 Mon Sep 17 00:00:00 2001 From: Focused Instability <70747559+MrAshRhodes@users.noreply.github.com> Date: Sat, 13 Jun 2026 17:43:35 +0200 Subject: [PATCH] feat: attribute reread waste to over-compression via marker check (#901) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 | 6 + headroom/dashboard/templates/dashboard.html | 3 + headroom/parser.py | 34 +++- headroom/reporting/generator.py | 10 +- headroom/transforms/pipeline.py | 11 +- tests/test_config.py | 3 +- tests/test_reread_attribution.py | 173 ++++++++++++++++++++ 7 files changed, 235 insertions(+), 5 deletions(-) create mode 100644 tests/test_reread_attribution.py diff --git a/headroom/config.py b/headroom/config.py index b0fda80dc..db9363570 100644 --- a/headroom/config.py +++ b/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, } diff --git a/headroom/dashboard/templates/dashboard.html b/headroom/dashboard/templates/dashboard.html index 0da82f9d2..fd6939845 100644 --- a/headroom/dashboard/templates/dashboard.html +++ b/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'; }, diff --git a/headroom/parser.py b/headroom/parser.py index 28560e220..4ef7692d6 100644 --- a/headroom/parser.py +++ b/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=|<]+>>") + # 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 diff --git a/headroom/reporting/generator.py b/headroom/reporting/generator.py index 24823d1db..3484350b0 100644 --- a/headroom/reporting/generator.py +++ b/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", } diff --git a/headroom/transforms/pipeline.py b/headroom/transforms/pipeline.py index 973fc30bf..5447ea741 100644 --- a/headroom/transforms/pipeline.py +++ b/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: diff --git a/tests/test_config.py b/tests/test_config.py index 64dd5c8fc..403ec386f 100644 --- a/tests/test_config.py +++ b/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: diff --git a/tests/test_reread_attribution.py b/tests/test_reread_attribution.py new file mode 100644 index 000000000..6009382ee --- /dev/null +++ b/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": "<>"} + + _, _, 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) + )