feat: detect re-served tool results as over-compression waste signal (#854)

Closes #853

## What

Adds a `reread` waste signal: identical `tool_result` content appearing
at more than one message position means the agent re-fetched something
already in context — the dominant failure signature of over-compression
(Manus context-engineering; JetBrains "Complexity Trap",
arXiv:2508.21433). Per-request savings can't see this cost; this signal
makes it visible.

- `WasteSignals.reread_tokens` — new field, in `total()`, exported as
`"reread"` in `to_dict()`.
- `parse_messages()` groups `tool_result` blocks by their **existing**
`content_hash` and counts every repeat beyond the first serve. No new
hashing or tokenization; one O(blocks) dict pass.
- `REREAD_MIN_TOKENS = 50` guard: short outputs ("ok", empty diffs)
legitimately repeat and are skipped. Duplicates within a single message
(same `source_index`) are not counted.
- Works across all formats the parser already normalizes to
`tool_result` blocks: OpenAI `role=tool`, Anthropic `tool_result`,
Strands/Bedrock `toolResult` (#813/#815).
- Flows through existing generic plumbing with zero handler changes:
pipeline → `RequestOutcome.waste_signals` → Prometheus
`headroom_waste_signal_tokens_total{signal="reread"}` → dashboard "Waste
Detected" panel. Dashboard gains label/color entries for the new key.

## Tests

7 new tests in `tests/test_parser.py::TestRereadDetection` (red before,
green after): OpenAI + Anthropic format detection, repeat-counting
semantics (first serve free), single-occurrence, short-duplicate guard,
same-message guard, `total()`/`to_dict()` participation. Updated 2
exact-shape assertions in `tests/test_config.py`.

Local runs: `tests/test_parser.py` (72 passed), `tests/test_config.py` +
outcome/reporting/observability/storage/proxy-hooks suites (190 passed),
`tests/test_canonical_pipeline.py` +
`tests/test_proxy_pipeline_lifecycle.py` (11 passed). `ruff check` +
`ruff format --check` clean.

## Real behavior proof

**Setup:** macOS (Darwin 25.5), Python 3.11.9, this branch, real proxy
server (`python -m headroom.proxy.server --port 18970
--anthropic-api-url http://127.0.0.1:18971`) with a local mock Anthropic
upstream returning a canned `/v1/messages` response (no real key
needed).

**Steps:** POSTed an Anthropic-format conversation to the live proxy:
agent fetches a 14 KB JSON log array via `get_logs` tool, then fetches
the identical content again under a different `tool_use_id` (the
re-read).

**Observed result** — `curl http://127.0.0.1:18970/metrics` after the
request:

```
# HELP headroom_waste_signal_tokens_total Tokens attributed to detected waste signals
# TYPE headroom_waste_signal_tokens_total counter
headroom_waste_signal_tokens_total{signal="json_bloat"} 9858
headroom_waste_signal_tokens_total{signal="reread"} 4935
```

`reread` = 4935 tokens, exactly the second serve of the ~4.9k-token tool
result (json_bloat counts both occurrences ≈ 2×). `/stats` shows the
same: `"waste_signals": {"json_bloat": 9858, "reread": 4935, ...}` —
which is what the dashboard panel renders.

Also verified the negative path live: a conversation whose tool results
contain non-compressible plain code text produced no waste-signal
entries (the pipeline only attributes waste when compression actually
engaged, unchanged behavior).

**Not tested:** Gemini `functionResponse` path (parser doesn't produce
`tool_result` blocks for it — pre-existing gap tracked in #819);
dashboard rendering only verified via the `/stats` payload the panel
binds to, not a browser screenshot.

## Out of scope (per #853)

Tool-call argument matching, compression-marker attribution,
tokens-per-task metric, cache hit-rate panel.

---------

Co-authored-by: Ash Rhodes <ashley.rhodes@king.com>

headroom/config.py

@@ -536,6 +536,7 @@ class WasteSignals:
     whitespace_tokens: int = 0  # Repeated whitespace
     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
 
     def total(self) -> int:
         """Total waste tokens detected."""
@@ -546,6 +547,7 @@ class WasteSignals:
             + self.whitespace_tokens
             + self.dynamic_date_tokens
             + self.repetition_tokens
+            + self.reread_tokens
         )
 
     def to_dict(self) -> dict[str, int]:
@@ -557,6 +559,7 @@ class WasteSignals:
             "whitespace": self.whitespace_tokens,
             "dynamic_date": self.dynamic_date_tokens,
             "repetition": self.repetition_tokens,
+            "reread": self.reread_tokens,
         }
 
 

headroom/dashboard/templates/dashboard.html

@@ -2081,6 +2081,7 @@
                         whitespace: 'Whitespace',
                         dynamic_date: 'Dynamic Dates',
                         repetition: 'Repetition',
+                        reread: 'Re-read Tool Results',
                     };
                     return labels[signal] || signal;
                 },
@@ -2093,6 +2094,7 @@
                         whitespace: 'bg-blue-500',
                         dynamic_date: 'bg-purple-500',
                         repetition: 'bg-pink-500',
+                        reread: 'bg-teal-500',
                     };
                     return colors[signal] || 'bg-gray-500';
                 },
@@ -2105,6 +2107,7 @@
                         whitespace: 'bg-blue-500/20 text-blue-400',
                         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',
                     };
                     return colors[signal] || 'bg-gray-500/20 text-gray-400';
                 },

headroom/parser.py

@@ -20,6 +20,17 @@ BASE64_PATTERN = re.compile(r"[A-Za-z0-9+/]{50,}={0,2}")
 WHITESPACE_PATTERN = re.compile(r"[ \t]{4,}|\n{3,}")
 JSON_BLOCK_PATTERN = re.compile(r"\{[\s\S]{500,}\}")
 
+# Tool results below this size legitimately repeat ("ok", empty diffs,
+# exit codes) and are not evidence of a re-read.
+REREAD_MIN_TOKENS = 50
+
+# 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
+# thinking/user nudge in the loop. Larger gaps mean the agent moved on
+# and then came back — the over-compression signal we want.
+REREAD_ADJACENT_GAP = 3
+
 # Patterns for RAG detection (best effort)
 RAG_MARKERS = [
     r"\[Document\s*\d+\]",
@@ -300,6 +311,33 @@ def parse_messages(
                 total_waste.dynamic_date_tokens += ws.get("dynamic_date", 0)
                 total_waste.repetition_tokens += ws.get("repetition", 0)
 
+    # Cross-message re-read detection: identical tool_result content served
+    # at more than one position means the agent re-fetched something already
+    # in context — an over-compression signal (#853). The first serve is
+    # free; every repeat is counted as waste.
+    reread_groups: dict[str, list[Block]] = {}
+    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)
+    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
+        # within the original message are excluded, and so are polling
+        # repeats: agents that poll (repeated `git status`, CI checks)
+        # legitimately produce byte-identical results a couple of messages
+        # apart. A repeat only counts when it lands more than
+        # REREAD_ADJACENT_GAP messages after the previous serve; nearer
+        # repeats advance the baseline without counting, so a long polling
+        # chain never accumulates waste.
+        prev_index = group[0].source_index
+        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
+
     # Compute block breakdown
     breakdown: dict[str, int] = {}
     for block in all_blocks:

headroom/reporting/generator.py

@@ -395,6 +395,7 @@ def _build_waste_histogram(
         "base64": 0,
         "whitespace": 0,
         "dynamic_date": 0,
+        "reread": 0,
         "history_bloat": 0,
     }
 
@@ -411,8 +412,11 @@ def _build_waste_histogram(
         # Estimate history bloat from tokens saved
         if metrics.tokens_input_before > metrics.tokens_input_after:
             tokens_saved = metrics.tokens_input_before - metrics.tokens_input_after
-            # Subtract known waste types
-            known_waste = sum(waste.values())
+            # 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")
             history_bloat = max(0, tokens_saved - known_waste)
             totals["history_bloat"] += history_bloat
 
@@ -425,6 +429,7 @@ def _build_waste_histogram(
         "base64": "Base64 Blobs",
         "whitespace": "Whitespace",
         "dynamic_date": "Dynamic Dates",
+        "reread": "Re-served Tool Results",
         "history_bloat": "History Bloat",
     }
 

tests/test_config.py

@@ -256,6 +256,7 @@ class TestWasteSignals:
             whitespace_tokens=25,
             dynamic_date_tokens=10,
             repetition_tokens=15,
+            reread_tokens=30,
         )
         expected = {
             "json_bloat": 100,
@@ -264,6 +265,7 @@ class TestWasteSignals:
             "whitespace": 25,
             "dynamic_date": 10,
             "repetition": 15,
+            "reread": 30,
         }
         assert signals.to_dict() == expected
 
@@ -272,7 +274,7 @@ class TestWasteSignals:
         signals = WasteSignals()
         result = signals.to_dict()
         assert all(v == 0 for v in result.values())
-        assert len(result) == 6
+        assert len(result) == 7
 
 
 class TestCachePrefixMetrics:

tests/test_parser.py

@@ -407,6 +407,131 @@ class TestParseMessages:
         assert len(tool_result_blocks) >= 1
 
 
+# --- TestRereadDetection ---
+
+
+class TestRereadDetection:
+    """Tests for cross-message re-read detection in parse_messages."""
+
+    LARGE_CONTENT = "def handler(event):\n    return process(event)\n" * 10  # > 200 chars
+
+    def _expected_tokens(self, text):
+        """Mirror mock_tokenizer + message overhead used for tool_result blocks."""
+        return len(text) // 4 + 1 + 4
+
+    @staticmethod
+    def _filler(n):
+        """Interleaved turns that push a repeat beyond the polling gap."""
+        return [
+            {"role": "assistant" if i % 2 == 0 else "user", "content": f"step {i} of the task"}
+            for i in range(n)
+        ]
+
+    def test_reread_detected_openai_tool_messages(self, mock_tokenizer):
+        """Identical large tool outputs far apart count as re-read."""
+        messages = (
+            [{"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT}]
+            + self._filler(4)
+            + [{"role": "tool", "tool_call_id": "c2", "content": self.LARGE_CONTENT}]
+        )
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
+
+    def test_reread_detected_anthropic_tool_result_blocks(self, mock_tokenizer):
+        """Anthropic-format tool_result parts are matched by content, not id."""
+        part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
+        part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
+        messages = (
+            [{"role": "user", "content": [part]}]
+            + self._filler(4)
+            + [{"role": "user", "content": [part2]}]
+        )
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
+
+    def test_three_occurrences_count_repeats_only(self, mock_tokenizer):
+        """First serve is free; every distant repeat is counted."""
+        msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
+        messages = [dict(msg)] + self._filler(4) + [dict(msg)] + self._filler(4) + [dict(msg)]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 2 * self._expected_tokens(self.LARGE_CONTENT)
+
+    def test_single_occurrence_no_signal(self, mock_tokenizer):
+        """One large tool result is not a re-read."""
+        messages = [{"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT}]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 0
+
+    def test_short_duplicates_ignored(self, mock_tokenizer):
+        """Trivially short outputs (\"ok\") legitimately repeat and are skipped."""
+        messages = [
+            {"role": "tool", "tool_call_id": "c1", "content": "ok"},
+            {"role": "tool", "tool_call_id": "c2", "content": "ok"},
+            {"role": "tool", "tool_call_id": "c3", "content": "ok"},
+        ]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 0
+
+    def test_same_message_duplicates_ignored(self, mock_tokenizer):
+        """Duplicates within a single message are not a re-read."""
+        part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
+        part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
+        messages = [{"role": "user", "content": [part, part2]}]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 0
+
+    def test_mixed_same_message_duplicate_not_counted(self, mock_tokenizer):
+        """A duplicate inside the original message stays excluded even when
+        a later message also re-serves the content."""
+        part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
+        part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
+        part3 = {"type": "tool_result", "tool_use_id": "t3", "content": self.LARGE_CONTENT}
+        messages = (
+            [{"role": "user", "content": [part, part2]}]
+            + self._filler(4)
+            + [{"role": "user", "content": [part3]}]
+        )
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
+
+    def test_adjacent_polling_repeats_ignored(self, mock_tokenizer):
+        """Back-to-back identical results (poll loop) are not re-reads."""
+        messages = [
+            {"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT},
+            {"role": "assistant", "content": "Still pending, checking again."},
+            {"role": "tool", "tool_call_id": "c2", "content": self.LARGE_CONTENT},
+        ]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 0
+
+    def test_polling_chain_never_accumulates(self, mock_tokenizer):
+        """Each poll advances the baseline — long chains stay at zero."""
+        msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
+        nudge = {"role": "assistant", "content": "polling"}
+        messages = [dict(msg), dict(nudge), dict(msg), dict(nudge), dict(msg), dict(nudge)]
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == 0
+
+    def test_distant_repeat_after_polling_chain_counts(self, mock_tokenizer):
+        """A far repeat counts even when earlier repeats were polling."""
+        msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
+        messages = (
+            [dict(msg), {"role": "assistant", "content": "polling"}, dict(msg)]
+            + self._filler(4)
+            + [dict(msg)]
+        )
+        _, _, waste = parse_messages(messages, mock_tokenizer)
+        assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
+
+    def test_reread_in_total_and_dict(self):
+        """reread_tokens participates in total() and to_dict()."""
+        from headroom.config import WasteSignals
+
+        ws = WasteSignals(reread_tokens=42)
+        assert ws.total() == 42
+        assert ws.to_dict()["reread"] == 42
+
+
 # --- TestFindToolUnits ---
 
 

tests/test_reporting.py

@@ -92,7 +92,7 @@ def test_build_waste_histogram_empty_and_filtered_data() -> None:
             tokens_input_before=200,
             tokens_input_after=100,
             cache_alignment_score=70,
-            waste_signals={"json_bloat": 30, "html_noise": 10, "dynamic_date": 5},
+            waste_signals={"json_bloat": 30, "html_noise": 10, "dynamic_date": 5, "reread": 20},
         ),
         FakeMetrics(
             request_id="flat",
@@ -125,6 +125,11 @@ def test_build_waste_histogram_empty_and_filtered_data() -> None:
     assert histogram[1] == pytest.approx(
         {"label": "Tool JSON Bloat", "tokens": 30, "percentage": 54.54545454545454}
     )
+    # "reread" surfaces in the histogram but is excluded from known_waste,
+    # so History Bloat above stays 100 - 45 = 55.
+    assert any(
+        item["label"] == "Re-served Tool Results" and item["tokens"] == 20 for item in histogram
+    )
     assert any(item["label"] == "HTML Noise" and item["tokens"] == 10 for item in histogram)
     assert any(item["label"] == "Dynamic Dates" and item["tokens"] == 5 for item in histogram)
     assert any(item["label"] == "Base64 Blobs" and item["tokens"] == 0 for item in histogram)