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) + )