fix(ccr): use shared compression store (#875)

## Description

Use shared get_compression_store() singleton in MCP _get_local_store so
headroom_retrieve sees proxy-compressed content.

Fixes #860

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Testing

Describe the tests you ran to verify your changes:

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality

headroom/ccr/mcp_server.py

@@ -342,14 +342,17 @@ class HeadroomMCPServer:
         self._setup_handlers()
 
     def _get_local_store(self) -> Any:
-        """Get or create the local compression store (lazy init)."""
-        if self._local_store is None:
-            from headroom.cache.compression_store import CompressionStore
+        """Get the shared compression store singleton (lazy init).
 
-            self._local_store = CompressionStore(
-                max_entries=500,
-                default_ttl=MCP_SESSION_TTL,
-            )
+        Returns the same instance the proxy and response_handler use so
+        retrieval can see content either side compressed in-process.
+        Called with no args to keep one shared config; the compress path
+        passes its own per-entry ``ttl`` at store time.
+        """
+        if self._local_store is None:
+            from headroom.cache.compression_store import get_compression_store
+
+            self._local_store = get_compression_store()
         return self._local_store
 
     def _compress_content(self, content: str) -> dict[str, Any]:

tests/test_ccr_mcp_server.py

@@ -1,7 +1,14 @@
 from __future__ import annotations
 
+import asyncio
 import json
 
+import pytest
+
+from headroom.cache.compression_store import (
+    get_compression_store,
+    reset_compression_store,
+)
 from headroom.ccr import mcp_server
 
 
@@ -22,3 +29,39 @@ def test_shared_stats_work_without_fcntl(monkeypatch, tmp_path) -> None:
 
     events = mcp_server._read_shared_events(window_seconds=60)
     assert events == [{"type": "compress", "timestamp": 1000.0, "pid": 4242}]
+
+
+# --- Shared compression store wiring ---------------------------------------
+# MCP's _get_local_store() must return the get_compression_store() singleton —
+# the same instance the proxy and response_handler use — so content compressed
+# on either side is retrievable in-process. These pin that wiring so a private
+# store can't creep back.
+
+
+@pytest.fixture
+def fresh_store():
+    reset_compression_store()
+    yield
+    reset_compression_store()
+
+
+def test_mcp_uses_shared_singleton_store(fresh_store) -> None:
+    """MCP's store is the global singleton, not a private instance."""
+    pytest.importorskip("mcp", reason="MCP SDK required")
+    server = mcp_server.HeadroomMCPServer(check_proxy=False)
+    assert server._get_local_store() is get_compression_store()
+
+
+def test_mcp_retrieves_proxy_stored_content(fresh_store) -> None:
+    """Content stored via the singleton (as the proxy does) is retrievable
+    through MCP's local-store path. The HTTP fallback is disabled so this
+    passes only via the shared store."""
+    pytest.importorskip("mcp", reason="MCP SDK required")
+    original = '{"some": "original proxy-compressed content"}'
+    hash_key = get_compression_store().store(original, '{"compressed": true}')
+
+    server = mcp_server.HeadroomMCPServer(check_proxy=False)
+    result = asyncio.run(server._retrieve_content(hash_key, query=None))
+
+    assert result.get("source") == "local"
+    assert result["original_content"] == original