feat: add CompressionCache with LRU eviction for token headroom mode

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

headroom/cache/compression_cache.py

@@ -0,0 +1,86 @@
+"""Content-addressed compression cache with LRU eviction.
+
+Used in "token headroom mode" to avoid re-compressing messages across turns.
+Maps original content hashes to their compressed versions.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import OrderedDict
+from dataclasses import dataclass
+
+
+@dataclass
+class _CacheEntry:
+    """Internal cache entry storing compressed text and metadata."""
+
+    compressed: str
+    tokens_saved: int
+
+
+class CompressionCache:
+    """Content-addressed cache mapping content hashes to compressed versions.
+
+    Uses an OrderedDict for O(1) LRU eviction. Entries are evicted oldest-first
+    when the cache exceeds max_entries.
+    """
+
+    def __init__(self, max_entries: int = 10000) -> None:
+        self.max_entries = max_entries
+        self._cache: OrderedDict[str, _CacheEntry] = OrderedDict()
+        self._hits: int = 0
+        self._misses: int = 0
+        self._total_tokens_saved: int = 0
+
+    def get_compressed(self, hash: str) -> str | None:
+        """Retrieve compressed content by hash, refreshing LRU position on hit."""
+        entry = self._cache.get(hash)
+        if entry is None:
+            self._misses += 1
+            return None
+        self._hits += 1
+        self._cache.move_to_end(hash)
+        return entry.compressed
+
+    def store_compressed(self, hash: str, compressed: str, tokens_saved: int) -> None:
+        """Store a compressed version keyed by content hash.
+
+        If the hash already exists, the entry is overwritten and moved to the
+        end (most recently used). When the cache exceeds max_entries, the oldest
+        entry is evicted.
+        """
+        if hash in self._cache:
+            old_entry = self._cache[hash]
+            self._total_tokens_saved -= old_entry.tokens_saved
+            del self._cache[hash]
+
+        self._cache[hash] = _CacheEntry(compressed=compressed, tokens_saved=tokens_saved)
+        self._total_tokens_saved += tokens_saved
+
+        while len(self._cache) > self.max_entries:
+            _, evicted = self._cache.popitem(last=False)
+            self._total_tokens_saved -= evicted.tokens_saved
+
+    def get_stats(self) -> dict:
+        """Return cache statistics."""
+        return {
+            "entries": len(self._cache),
+            "hits": self._hits,
+            "misses": self._misses,
+            "tokens_saved": self._total_tokens_saved,
+        }
+
+    @staticmethod
+    def content_hash(content: str | list) -> str:
+        """Compute a truncated SHA-256 hash for string or list content.
+
+        For list content (Anthropic-format messages with type/text/content fields),
+        the list is JSON-serialized with sorted keys for deterministic hashing.
+        """
+        if isinstance(content, list):
+            raw = json.dumps(content, sort_keys=True, ensure_ascii=False)
+        else:
+            raw = content
+        return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]

tests/test_compression_cache.py

@@ -0,0 +1,109 @@
+"""Tests for CompressionCache with LRU eviction."""
+
+from __future__ import annotations
+
+import pytest
+
+from headroom.cache.compression_cache import CompressionCache
+
+
+@pytest.fixture
+def cache() -> CompressionCache:
+    return CompressionCache()
+
+
+@pytest.fixture
+def small_cache() -> CompressionCache:
+    return CompressionCache(max_entries=3)
+
+
+class TestCompressionCache:
+    def test_cache_miss_returns_none(self, cache: CompressionCache) -> None:
+        h = CompressionCache.content_hash("some content")
+        assert cache.get_compressed(h) is None
+
+    def test_store_and_retrieve(self, cache: CompressionCache) -> None:
+        content = "hello world this is a long message"
+        h = CompressionCache.content_hash(content)
+        cache.store_compressed(h, "hello world...compressed", tokens_saved=15)
+        assert cache.get_compressed(h) == "hello world...compressed"
+
+    def test_different_content_different_hash(self) -> None:
+        h1 = CompressionCache.content_hash("content A")
+        h2 = CompressionCache.content_hash("content B")
+        assert h1 != h2
+
+    def test_overwrite_same_hash(self, cache: CompressionCache) -> None:
+        h = CompressionCache.content_hash("some content")
+        cache.store_compressed(h, "v1", tokens_saved=10)
+        cache.store_compressed(h, "v2", tokens_saved=20)
+        assert cache.get_compressed(h) == "v2"
+
+    def test_stats_tracking(self, cache: CompressionCache) -> None:
+        h = CompressionCache.content_hash("content")
+        cache.store_compressed(h, "compressed", tokens_saved=5)
+
+        # One hit
+        cache.get_compressed(h)
+        # One miss
+        cache.get_compressed("nonexistent")
+
+        stats = cache.get_stats()
+        assert stats["hits"] == 1
+        assert stats["misses"] == 1
+        assert stats["entries"] == 1
+        assert stats["tokens_saved"] == 5
+
+    def test_eviction_at_max_entries(self, small_cache: CompressionCache) -> None:
+        h1 = CompressionCache.content_hash("a")
+        h2 = CompressionCache.content_hash("b")
+        h3 = CompressionCache.content_hash("c")
+        h4 = CompressionCache.content_hash("d")
+
+        small_cache.store_compressed(h1, "ca", tokens_saved=1)
+        small_cache.store_compressed(h2, "cb", tokens_saved=1)
+        small_cache.store_compressed(h3, "cc", tokens_saved=1)
+
+        # Adding a 4th should evict the oldest (h1)
+        small_cache.store_compressed(h4, "cd", tokens_saved=1)
+
+        assert small_cache.get_compressed(h1) is None
+        assert small_cache.get_compressed(h2) == "cb"
+        assert small_cache.get_compressed(h4) == "cd"
+
+    def test_access_refreshes_lru(self, small_cache: CompressionCache) -> None:
+        h1 = CompressionCache.content_hash("a")
+        h2 = CompressionCache.content_hash("b")
+        h3 = CompressionCache.content_hash("c")
+        h4 = CompressionCache.content_hash("d")
+
+        small_cache.store_compressed(h1, "ca", tokens_saved=1)
+        small_cache.store_compressed(h2, "cb", tokens_saved=1)
+        small_cache.store_compressed(h3, "cc", tokens_saved=1)
+
+        # Access h1 to refresh it
+        small_cache.get_compressed(h1)
+
+        # Adding h4 should evict h2 (oldest untouched), not h1
+        small_cache.store_compressed(h4, "cd", tokens_saved=1)
+
+        assert small_cache.get_compressed(h1) == "ca"
+        assert small_cache.get_compressed(h2) is None
+        assert small_cache.get_compressed(h4) == "cd"
+
+    def test_content_hash_list_content(self) -> None:
+        """content_hash handles Anthropic-format list content."""
+        list_content = [
+            {"type": "text", "text": "hello"},
+            {"type": "text", "text": "world"},
+        ]
+        h = CompressionCache.content_hash(list_content)
+        assert isinstance(h, str)
+        assert len(h) == 16
+
+        # Same content produces same hash
+        assert CompressionCache.content_hash(list_content) == h
+
+    def test_content_hash_string_length(self) -> None:
+        h = CompressionCache.content_hash("test")
+        assert len(h) == 16