fix(cache/ccr): don't evict a live entry on a duplicate store at capacity (#2082)

## Description

`CompressionStore.store` (`headroom/cache/compression_store.py`) runs
eviction **before** it knows
whether the incoming `hash_key` is new or a re-store of an
already-present key:

```python
with self._lock:
    self._evict_if_needed()          # <-- runs first

    existing = self._backend.get(hash_key)
    if existing is not None:
        ...                          # duplicate / collision: overwrite in place
        self._stale_heap_entries += 1
    self._backend.set(hash_key, entry)
```

When the store is full and the incoming key **already exists** (a
duplicate re-store),
`_evict_if_needed()` removes the oldest *distinct* entry to "make room"
— but then `set()` merely
overwrites the existing key in place, so no room was ever needed. Net
effect: `count` drops to
`max_entries - 1` and a **live, never-retrieved entry is destroyed**.
That entry's `<<ccr:...>>`
marker, still sitting in the conversation history, then resolves to a
404 on `/v1/retrieve`.

This is not a corner case: the CCR mirror bridge
(`_mirror_single_hash_to_python_store` in
`smart_crusher.py`) re-`store()`s the same `explicit_hash` every turn a
`<<ccr:…>>` marker is
re-encountered, and markers persist across turns — so a full store
silently deletes a live sibling
entry on each duplicate.

Concrete (with the repo's `max_entries=3` fixture): store c0, c1, c2
(full), then re-store c1
(same content ⇒ same hash). Eviction pops the oldest (c0), deletes it,
then c1 is overwritten in
place. Final state: {c1, c2}, count 2, and **c0 is gone** — its marker
is now unredeemable.

Closes: no issue filed — found while auditing the compression store.

## Fix

Decide novelty before evicting: only `_evict_if_needed()` for a
genuinely new key; a
duplicate/replace overwrites in place (no eviction). The
collision/duplicate logging and
stale-heap accounting are unchanged.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- `headroom/cache/compression_store.py`: `store()` reads `existing`
first and only evicts when the key is new.
- `tests/test_compression_store.py`: add
`test_duplicate_store_at_capacity_does_not_evict` (re-store an existing
hash at capacity keeps all entries and count at `max_entries`).

## Testing

- [x] New regression test added (`tests/test_compression_store.py`)
- [x] Linting/formatting clean — run with the CI-pinned `ruff==0.15.17`
- [ ] Full `pytest` deferred to CI (local-OOM reason below).

```text
$ uvx ruff@0.15.17 check headroom/cache/compression_store.py tests/test_compression_store.py
All checks passed!
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.10, headroom from this branch.
Importing `headroom` pulls in the torch/transformers stack and a full
`pytest` gets OOM-killed on this box, so I verified the store/evict
logic with a dependency-free script (replicating the backend + eviction
heap) and left the full pytest to CI.
- Exact command / steps: filled a `max_entries=3` store with c0/c1/c2,
then re-stored c1 (duplicate), through the old (evict-first) and new
(check-first) logic; also confirmed a genuinely new key still evicts the
oldest.
- Observed result: the old logic drops c0 (count 2); the new keeps all
three; and a new key at capacity still evicts the oldest:

```text
OLD: after duplicate re-store of h1 -> keys=['h1', 'h2'] count=2
NEW: after duplicate re-store of h1 -> keys=['h0', 'h1', 'h2'] count=3
NEW still evicts oldest for a genuinely new key at capacity
DUPLICATE-STORE EVICTION FIX VERIFIED (old drops a live entry; new keeps it)
```

- Not tested: a full proxy CCR round-trip (needs the heavy stack). The
fix is confined to `store()` and the new test drives it directly with
the `max_entries=3` fixture. Existing eviction tests use distinct keys
and stay green. Full local `pytest` deferred to CI (OOM, per above).

## 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 my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes — ran
lint + a standalone logic check; full pytest deferred to CI (local OOM,
disclosed above)
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

- Internal reordering only, no signature change; no call sites or
backend mocks break.
- @JerrettDavis tagging you — this silently drops a live CCR entry
(making its marker 404) whenever a duplicate hash is re-stored at
capacity, which the mirror bridge does routinely. Thanks.

Co-authored-by: Tejas Chopra <chopratejas@gmail.com>

CHANGELOG.md

@@ -102,6 +102,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug Fixes
 
+* **cache/ccr:** don't evict a live entry when a duplicate hash is re-stored at capacity. `CompressionStore.store` ran `_evict_if_needed()` before checking whether the key already existed, so re-storing an already-present hash while the store was full evicted the oldest *distinct* entry to "make room" and then merely overwrote the existing key in place — no room was ever needed. The store dropped below `max_entries` and a live, never-retrieved entry was destroyed, so its `<<ccr:...>>` marker (still in the conversation) resolved to a 404. The CCR mirror bridge re-stores the same `explicit_hash` every turn a marker is re-encountered, so this fired routinely. Eviction now runs only for a genuinely new key.
 * **tokenizers:** recurse into a native `tool_result` whose content is a list of blocks instead of JSON-serializing it. `_count_content_parts` counted a `tool_result` with list content via `_count_serialized` (json.dumps + sample), so a base64 image nested in a tool result (computer-use / MCP screenshot tools) was priced as text — a ~50-200x overcount (a ~200KB screenshot read as ~70K tokens instead of ~1600). It now recurses into the nested blocks, matching the sibling Strands `toolResult` branch, so the image is priced structurally. The overcount made a single screenshot appear to blow past the model's context window and triggered unnecessary/over-aggressive compression.
 * **tokenizers:** price dense scripts (CJK/Kana/Hangul) in the fixed-ratio estimator path. `EstimatingTokenCounter.count_text` applied the CJK correction only on the auto-detect path; its fixed-ratio early return divided by the Latin ratio with no adjustment. The registry builds every provider-calibrated counter with a fixed ratio (Anthropic 3.5, Google 4.0, Cohere 4.0, Moonshot 3.1), and the Anthropic/Gemini proxy handlers count via `get_tokenizer(model).count_messages`, so a CJK-heavy context read as ~40-55% of its true token size — it could fall under the size/backpressure gates and skip compression, and every `x-headroom-tokens-before` metric for CJK traffic was materially wrong. The fixed-ratio path now applies the same dense-script split.
 * **ccr:** don't crash `parse_tool_call` on a CCR tool call whose arguments aren't an object. For the OpenAI/`openai_responses` shape the arguments are `json.loads`-decoded and only `JSONDecodeError` was caught, so a model that emitted `arguments='[]'`/`'"abc"'`/`'123'` (decoding to a list/str/number) — or a non-dict Anthropic `input` — reached `input_data.get("hash")` and raised `AttributeError`; a null `arguments` raised an uncaught `TypeError` from `json.loads(None)`. Both are now handled: the decode also catches `TypeError`, and a non-dict `input_data` returns `None` (not a valid CCR call) instead of crashing CCR response processing.

headroom/cache/compression_store.py

@@ -347,16 +347,22 @@ class CompressionStore:
             self.process_pending_feedback()
 
         with self._lock:
-            self._evict_if_needed()
-
-            # CRITICAL FIX: Hash collision detection
-            # If hash already exists with DIFFERENT content, log a warning.
-            # This indicates either a hash collision or duplicate store calls.
+            # Decide whether this is a NEW key before evicting. Evicting to make
+            # room only applies to a genuinely new entry; a re-store of an
+            # existing key overwrites in place (no room needed). Evicting first
+            # for a duplicate would needlessly destroy a live, unrelated entry
+            # and drop the store below capacity, making that entry's <<ccr:...>>
+            # marker (still sitting in the conversation) unredeemable — a 404.
+            # The CCR mirror bridge re-stores the same explicit_hash on every
+            # turn a marker is re-encountered, so duplicate stores are common.
             existing = self._backend.get(hash_key)
-            if existing is not None:
+            if existing is None:
+                self._evict_if_needed()
+            else:
+                # Hash already present. Different content means a true (extremely
+                # rare with SHA256[:24]) collision; same content is a duplicate
+                # re-store. Either way we overwrite in place.
                 if existing.original_content != original:
-                    # True hash collision - different content, same hash
-                    # This is extremely rare with SHA256[:24] but should be logged
                     logger.warning(
                         "Hash collision detected: hash=%s tool=%s (existing_len=%d, new_len=%d)",
                         hash_key,
@@ -365,12 +371,11 @@ class CompressionStore:
                         len(original),
                     )
                 else:
-                    # Same content being stored again - this is fine, just update
                     logger.debug(
                         "Duplicate store for hash=%s, updating entry",
                         hash_key,
                     )
-                # Mark old heap entry as stale since we're replacing
+                # Mark old heap entry as stale since we're replacing it.
                 self._stale_heap_entries += 1
 
             self._backend.set(hash_key, entry)

tests/test_compression_store.py

@@ -706,6 +706,32 @@ class TestCompressionStoreEviction:
         assert store_with_small_capacity.exists(hashes[2])
         assert store_with_small_capacity.exists(new_hash)
 
+    def test_duplicate_store_at_capacity_does_not_evict(
+        self, store_with_small_capacity: CompressionStore
+    ):
+        """Re-storing an already-present hash at capacity overwrites in place and
+        must NOT evict an unrelated live entry (which would drop below capacity
+        and make that entry's marker unredeemable). The CCR mirror bridge
+        re-stores the same hash on later turns, so this is a common path."""
+        hashes = []
+        for i in range(3):
+            hashes.append(
+                store_with_small_capacity.store(
+                    original=f"content_{i}", compressed=f"compressed_{i}"
+                )
+            )
+            time.sleep(0.01)
+        assert store_with_small_capacity.get_stats()["entry_count"] == 3
+
+        # Re-store the SAME content for the oldest entry (a duplicate -> same hash).
+        dup = store_with_small_capacity.store(original="content_0", compressed="compressed_0")
+        assert dup == hashes[0]
+
+        # No eviction happened: all three entries survive and count stays at 3.
+        for h in hashes:
+            assert store_with_small_capacity.exists(h)
+        assert store_with_small_capacity.get_stats()["entry_count"] == 3
+
     def test_eviction_cleans_expired_first(self):
         """Eviction cleans expired entries before evicting valid ones."""
         store = CompressionStore(max_entries=3, default_ttl=1)