Add pluggable adapter hooks for CCR, Storage, and TOIN backends

Enable SaaS packages to inject custom backends (Redis, PostgreSQL, etc.)
without forking OSS, using entry_points and ContextVars for tenant isolation.

CCR: Request-scoped ContextVar + HEADROOM_CCR_BACKEND env + entry_point loading
Storage: URL-scheme entry_point resolution for custom storage backends
TOIN: New TOINBackend protocol + FileSystemTOINBackend (extracted from toin.py)
+ HEADROOM_TOIN_BACKEND env + entry_point loading

31 tests covering protocol conformance, backend wiring, entry_point loading,
ContextVar thread isolation, and end-to-end adapter lifecycle.

headroom/cache/compression_store.py

@@ -37,9 +37,11 @@ import hashlib
 import heapq
 import json
 import logging
+import os
 import re
 import threading
 import time
+from contextvars import ContextVar
 from dataclasses import dataclass, field, replace
 from typing import TYPE_CHECKING, Any
 
@@ -823,41 +825,101 @@ class CompressionStore:
                     logger.debug("TOIN record_retrieval failed", exc_info=True)
 
 
+# Request-scoped store (for multi-tenant SaaS: one store per request/tenant)
+_request_ccr_store: ContextVar[CompressionStore | None] = ContextVar(
+    "headroom_request_ccr_store", default=None
+)
+
 # Global store instance (lazy initialization)
 _compression_store: CompressionStore | None = None
 _store_lock = threading.Lock()
 
 
+def set_request_compression_store(store: CompressionStore | None) -> None:
+    """Set the compression store for the current request context.
+
+    Used by middleware (e.g. SaaS) to provide a tenant-scoped store.
+    When set, get_compression_store() returns this store instead of the global one.
+
+    Args:
+        store: CompressionStore to use for this request, or None to clear.
+    """
+    _request_ccr_store.set(store)
+
+
+def clear_request_compression_store() -> None:
+    """Clear the request-scoped compression store."""
+    _request_ccr_store.set(None)
+
+
+def _create_default_ccr_backend() -> CompressionStoreBackend | None:
+    """Create a CCR backend from env (e.g. HEADROOM_CCR_BACKEND=redis).
+
+    Loads adapters via setuptools entry point 'headroom.ccr_backend'.
+    Returns None to use default InMemoryBackend.
+    """
+    backend_type = (os.environ.get("HEADROOM_CCR_BACKEND") or "").strip().lower()
+    if not backend_type or backend_type == "memory":
+        return None
+    try:
+        from importlib.metadata import entry_points
+
+        all_eps = entry_points(group="headroom.ccr_backend")
+        ep = next((e for e in all_eps if e.name == backend_type), None)
+        if ep is None:
+            logger.warning(
+                "HEADROOM_CCR_BACKEND=%s but no entry point headroom.ccr_backend[%s]",
+                backend_type,
+                backend_type,
+            )
+            return None
+        fn = ep.load()
+        kwargs = {
+            "url": os.environ.get("HEADROOM_REDIS_URL", ""),
+            "tenant_prefix": os.environ.get("HEADROOM_CCR_TENANT_PREFIX", ""),
+        }
+        backend: CompressionStoreBackend = fn(**kwargs)
+        return backend
+    except Exception as e:
+        logger.warning("Failed to load CCR backend %s: %s", backend_type, e)
+        return None
+
+
 def get_compression_store(
     max_entries: int = 1000,
     default_ttl: int = 300,
     backend: CompressionStoreBackend | None = None,
 ) -> CompressionStore:
-    """Get the global compression store instance.
+    """Get the compression store instance.
 
-    Uses lazy initialization with singleton pattern.
+    If a request-scoped store was set (e.g. by SaaS middleware), returns it.
+    Otherwise uses lazy-initialized global singleton. Backend can be supplied
+    explicitly or created from env (HEADROOM_CCR_BACKEND) when building the global.
 
     Args:
-        max_entries: Maximum entries (only used on first call).
-        default_ttl: Default TTL (only used on first call).
-        backend: Custom storage backend (only used on first call).
-                 Defaults to InMemoryBackend if not provided.
+        max_entries: Maximum entries (only used on first call for global store).
+        default_ttl: Default TTL (only used on first call for global store).
+        backend: Custom storage backend (only used on first call for global store).
+                 Defaults to InMemoryBackend if not provided; env backend used if backend is None.
 
     Returns:
-        Global CompressionStore instance.
+        Request-scoped CompressionStore if set, else global CompressionStore instance.
     """
-    global _compression_store
+    request_store = _request_ccr_store.get()
+    if request_store is not None:
+        return request_store
 
+    global _compression_store
     if _compression_store is None:
         with _store_lock:
-            # Double-check after acquiring lock
             if _compression_store is None:
+                if backend is None:
+                    backend = _create_default_ccr_backend()
                 _compression_store = CompressionStore(
                     max_entries=max_entries,
                     default_ttl=default_ttl,
                     backend=backend,
                 )
-
     return _compression_store
 
 

headroom/storage/__init__.py

@@ -15,10 +15,13 @@ def create_storage(store_url: str) -> Storage:
     """
     Create a storage instance from URL.
 
-    Supported URLs:
+    Supported URLs (built-in):
     - sqlite:///path/to/file.db
     - jsonl:///path/to/file.jsonl
 
+    Other schemes (e.g. postgres://) can be provided by packages that register
+    the setuptools entry point headroom.storage_backend with name=<scheme>.
+
     Args:
         store_url: Storage URL.
 
@@ -37,5 +40,19 @@ def create_storage(store_url: str) -> Storage:
             path = path
         return JSONLStorage(path)
     else:
-        # Default to SQLite
+        # Unknown scheme: try entry point headroom.storage_backend[name=scheme]
+        scheme = store_url.split("://", 1)[0].lower() if "://" in store_url else ""
+        if scheme:
+            try:
+                from importlib.metadata import entry_points
+
+                all_eps = entry_points(group="headroom.storage_backend")
+                ep = next((e for e in all_eps if e.name == scheme), None)
+                if ep is not None:
+                    create_fn = ep.load()
+                    result: Storage = create_fn(store_url)
+                    return result
+            except Exception:
+                pass
+        # Default to SQLite (legacy behavior)
         return SQLiteStorage(store_url)

headroom/telemetry/backends/__init__.py

@@ -0,0 +1,27 @@
+"""Storage backends for TOIN (Tool Output Intelligence Network).
+
+This module provides pluggable storage backends for TOIN pattern persistence.
+The default is filesystem storage (JSON), but alternative backends can be
+implemented for distributed/multi-tenant scenarios (Redis, PostgreSQL, etc.).
+
+Usage:
+    from headroom.telemetry.backends import TOINBackend, FileSystemTOINBackend
+
+    # Use default filesystem backend
+    backend = FileSystemTOINBackend("/path/to/toin.json")
+
+    # Use custom backend (e.g. from a SaaS adapter package)
+    class RedisBackend:
+        # Implement TOINBackend protocol
+        ...
+
+    toin = ToolIntelligenceNetwork(config, backend=RedisBackend(...))
+"""
+
+from .base import TOINBackend
+from .filesystem import FileSystemTOINBackend
+
+__all__ = [
+    "TOINBackend",
+    "FileSystemTOINBackend",
+]

headroom/telemetry/backends/base.py

@@ -0,0 +1,54 @@
+"""Base protocol for TOIN storage backends.
+
+This protocol defines the minimal interface that TOIN storage backends must
+implement. The interface is intentionally simple — it only handles serialized
+pattern data load/save. All TOIN logic (pattern aggregation, recommendations,
+merging) stays in ToolIntelligenceNetwork.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class TOINBackend(Protocol):
+    """Protocol for TOIN storage backends.
+
+    Implementations can use any storage mechanism: filesystem, Redis,
+    PostgreSQL, S3, etc.
+
+    Design Principles:
+    - Two operations only: load and save
+    - Data is a serialized dict (from ToolIntelligenceNetwork.export_patterns)
+    - Backend owns atomicity and durability
+    - Thread-safety is implementation's responsibility
+
+    Example implementation:
+        class MyBackend:
+            def load(self) -> dict[str, Any]:
+                return json.loads(self._redis.get("toin_data") or "{}")
+
+            def save(self, data: dict[str, Any]) -> None:
+                self._redis.set("toin_data", json.dumps(data))
+    """
+
+    def load(self) -> dict[str, Any]:
+        """Load serialized TOIN data.
+
+        Returns:
+            Dict with TOIN data (as produced by export_patterns),
+            or empty dict if no data exists.
+        """
+        ...
+
+    def save(self, data: dict[str, Any]) -> None:
+        """Save serialized TOIN data.
+
+        The implementation must ensure atomicity — a failed save must not
+        corrupt existing data.
+
+        Args:
+            data: Serialized TOIN data (from export_patterns).
+        """
+        ...

headroom/telemetry/backends/filesystem.py

@@ -0,0 +1,78 @@
+"""Filesystem storage backend for TOIN.
+
+Stores TOIN patterns as a JSON file with atomic writes.
+This is the default backend, matching the original toin.py behavior.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import tempfile
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class FileSystemTOINBackend:
+    """Filesystem-backed TOIN storage using atomic JSON writes.
+
+    Characteristics:
+    - Persists to a single JSON file
+    - Atomic writes via temp file + rename (POSIX)
+    - Creates parent directories on first save
+    - Returns empty dict if file doesn't exist or is corrupted
+
+    Args:
+        path: Path to the JSON storage file.
+    """
+
+    def __init__(self, path: str) -> None:
+        self._path = Path(path)
+
+    def load(self) -> dict[str, Any]:
+        """Load TOIN data from the JSON file.
+
+        Returns:
+            Parsed JSON data, or empty dict if file doesn't exist or is corrupt.
+        """
+        if not self._path.exists():
+            return {}
+
+        try:
+            with open(self._path) as f:
+                data: dict[str, Any] = json.load(f)
+                return data
+        except (json.JSONDecodeError, OSError) as e:
+            logger.warning("Failed to load TOIN data from %s: %s", self._path, e)
+            return {}
+
+    def save(self, data: dict[str, Any]) -> None:
+        """Save TOIN data to the JSON file with atomic write.
+
+        Uses a temporary file and rename to ensure atomicity.
+        If the write fails, the original file is preserved.
+
+        Args:
+            data: Serialized TOIN data to persist.
+        """
+        try:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+
+            json_data = json.dumps(data, indent=2)
+
+            fd, tmp_path = tempfile.mkstemp(dir=self._path.parent, prefix=".toin_", suffix=".tmp")
+            try:
+                with open(fd, "w") as f:
+                    f.write(json_data)
+                Path(tmp_path).replace(self._path)
+            except Exception:
+                try:
+                    Path(tmp_path).unlink()
+                except OSError:
+                    pass
+                raise
+
+        except OSError as e:
+            logger.warning("Failed to save TOIN data to %s: %s", self._path, e)

headroom/telemetry/toin.py

@@ -346,15 +346,32 @@ class ToolIntelligenceNetwork:
     Thread-safe for concurrent access.
     """
 
-    def __init__(self, config: TOINConfig | None = None):
+    def __init__(
+        self,
+        config: TOINConfig | None = None,
+        backend: Any | None = None,
+    ):
         """Initialize TOIN.
 
         Args:
             config: Configuration options.
+            backend: Storage backend implementing TOINBackend protocol.
+                     If None, creates a FileSystemTOINBackend from config.storage_path.
+                     Pass a custom backend for Redis, PostgreSQL, etc.
         """
+        from .backends import FileSystemTOINBackend
+
         self._config = config or TOINConfig()
         self._lock = threading.RLock()  # RLock for reentrant locking (save calls export_patterns)
 
+        # Storage backend
+        if backend is not None:
+            self._backend = backend
+        elif self._config.storage_path:
+            self._backend = FileSystemTOINBackend(self._config.storage_path)
+        else:
+            self._backend = None
+
         # Pattern database: structure_hash -> ToolPattern
         self._patterns: dict[str, ToolPattern] = {}
 
@@ -367,9 +384,9 @@ class ToolIntelligenceNetwork:
         self._last_save_time = time.time()
         self._dirty = False
 
-        # Load existing data
-        if self._config.storage_path:
-            self._load_from_disk()
+        # Load existing data from backend
+        if self._backend is not None:
+            self._load_from_backend()
 
     def _generate_stable_instance_id(self) -> str:
         """Generate a stable instance ID that doesn't change across restarts.
@@ -1464,76 +1481,43 @@ class ToolIntelligenceNetwork:
         self._update_recommendations(existing)
 
     def save(self) -> None:
-        """Save TOIN data to disk with atomic write.
-
-        Uses a temporary file and rename to ensure atomicity.
-        If the write fails, the original file is preserved.
+        """Save TOIN data via the storage backend.
 
         HIGH FIX: Serialize under lock but write outside lock to prevent
         blocking other threads during slow file I/O.
         """
-        if not self._config.storage_path:
+        if self._backend is None:
             return
 
-        import tempfile
-
         # Step 1: Serialize under lock (fast in-memory operation)
         with self._lock:
             data = self.export_patterns()
 
         # Step 2: Write outside lock (slow I/O operation)
-        path = Path(self._config.storage_path)
-
         try:
-            # Create parent directories if needed
-            path.parent.mkdir(parents=True, exist_ok=True)
-
-            # Serialize to string (outside lock but before file ops)
-            json_data = json.dumps(data, indent=2)
-
-            # Write to temporary file first (atomic write pattern)
-            # Use same directory to ensure same filesystem for rename
-            fd, tmp_path = tempfile.mkstemp(dir=path.parent, prefix=".toin_", suffix=".tmp")
-            try:
-                with open(fd, "w") as f:
-                    f.write(json_data)
-
-                # Atomic rename (on POSIX systems)
-                Path(tmp_path).replace(path)
-
-            except Exception:
-                # Clean up temp file on failure
-                try:
-                    Path(tmp_path).unlink()
-                except OSError:
-                    pass
-                raise
+            self._backend.save(data)
 
             # Step 3: Update state under lock (fast)
             with self._lock:
                 self._dirty = False
                 self._last_save_time = time.time()
 
-        except OSError as e:
+        except Exception as e:
             # Log error but don't crash - TOIN should be resilient
-            logger.warning(f"Failed to save TOIN data: {e}")
+            logger.warning("Failed to save TOIN data: %s", e)
 
-    def _load_from_disk(self) -> None:
-        """Load TOIN data from disk."""
-        if not self._config.storage_path:
-            return
-
-        path = Path(self._config.storage_path)
-        if not path.exists():
+    def _load_from_backend(self) -> None:
+        """Load TOIN data from the storage backend."""
+        if self._backend is None:
             return
 
         try:
-            with open(path) as f:
-                data = json.load(f)
-            self.import_patterns(data)
-            self._dirty = False
-        except (json.JSONDecodeError, OSError):
-            pass  # Start fresh if corrupted
+            data = self._backend.load()
+            if data:
+                self.import_patterns(data)
+                self._dirty = False
+        except Exception as e:
+            logger.warning("Failed to load TOIN data from backend: %s", e)
 
     def _maybe_auto_save(self) -> None:
         """Auto-save if enough time has passed.
@@ -1543,7 +1527,7 @@ class ToolIntelligenceNetwork:
         The save() method already acquires the lock, and we use RLock so
         it's safe to hold the lock when calling save().
         """
-        if not self._config.storage_path or not self._config.auto_save_interval:
+        if self._backend is None or not self._config.auto_save_interval:
             return
 
         # Check under lock to prevent race conditions
@@ -1567,6 +1551,41 @@ class ToolIntelligenceNetwork:
 _toin_instance: ToolIntelligenceNetwork | None = None
 _toin_lock = threading.Lock()
 
+# Environment variable for custom TOIN backend
+TOIN_BACKEND_ENV_VAR = "HEADROOM_TOIN_BACKEND"
+
+
+def _create_default_toin_backend() -> Any:
+    """Create a TOIN backend from env (e.g. HEADROOM_TOIN_BACKEND=redis).
+
+    Loads adapters via setuptools entry point 'headroom.toin_backend'.
+    Returns None to use default FileSystemTOINBackend.
+    """
+    backend_type = (os.environ.get(TOIN_BACKEND_ENV_VAR) or "").strip().lower()
+    if not backend_type or backend_type == "filesystem":
+        return None
+    try:
+        from importlib.metadata import entry_points
+
+        all_eps = entry_points(group="headroom.toin_backend")
+        ep = next((e for e in all_eps if e.name == backend_type), None)
+        if ep is None:
+            logger.warning(
+                "HEADROOM_TOIN_BACKEND=%s but no entry point headroom.toin_backend[%s]",
+                backend_type,
+                backend_type,
+            )
+            return None
+        fn = ep.load()
+        kwargs = {
+            "url": os.environ.get("HEADROOM_TOIN_URL", ""),
+            "tenant_prefix": os.environ.get("HEADROOM_TOIN_TENANT_PREFIX", ""),
+        }
+        return fn(**kwargs)
+    except Exception as e:
+        logger.warning("Failed to load TOIN backend %s: %s", backend_type, e)
+        return None
+
 
 def get_toin(config: TOINConfig | None = None) -> ToolIntelligenceNetwork:
     """Get the global TOIN instance.
@@ -1574,6 +1593,10 @@ def get_toin(config: TOINConfig | None = None) -> ToolIntelligenceNetwork:
     Thread-safe singleton pattern. Always acquires lock to avoid subtle
     race conditions in double-checked locking on non-CPython implementations.
 
+    On first call, checks HEADROOM_TOIN_BACKEND env var. If set, loads the
+    backend via setuptools entry point 'headroom.toin_backend'. Otherwise
+    uses the default FileSystemTOINBackend.
+
     Args:
         config: Configuration (only used on first call). If the instance
             already exists, config is ignored and a warning is logged.
@@ -1587,7 +1610,8 @@ def get_toin(config: TOINConfig | None = None) -> ToolIntelligenceNetwork:
     # implementations. The overhead is negligible since we only construct once.
     with _toin_lock:
         if _toin_instance is None:
-            _toin_instance = ToolIntelligenceNetwork(config)
+            backend = _create_default_toin_backend()
+            _toin_instance = ToolIntelligenceNetwork(config, backend=backend)
         elif config is not None:
             # Warn when config is silently ignored
             logger.warning(

tests/test_adapter_hooks.py

@@ -0,0 +1,551 @@
+"""Tests for pluggable adapter hooks.
+
+Validates the extension points that allow SaaS packages to inject
+custom backends for CCR, Storage, and TOIN without forking OSS.
+
+Tests cover:
+1. TOIN backend protocol conformance (FileSystemTOINBackend)
+2. TOIN backend wiring (ToolIntelligenceNetwork with custom backend)
+3. TOIN entry_point loading (_create_default_toin_backend)
+4. CCR ContextVar scoping (set/clear/get request compression store)
+5. CCR entry_point loading (_create_default_ccr_backend)
+6. Storage entry_point loading (create_storage with custom scheme)
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from typing import Any
+
+import pytest
+
+from headroom.cache.backends import CompressionStoreBackend, InMemoryBackend
+from headroom.cache.compression_store import (
+    CompressionStore,
+    clear_request_compression_store,
+    get_compression_store,
+    reset_compression_store,
+    set_request_compression_store,
+)
+from headroom.storage import Storage, create_storage
+from headroom.telemetry.backends import FileSystemTOINBackend, TOINBackend
+from headroom.telemetry.models import ToolSignature
+from headroom.telemetry.toin import (
+    TOINConfig,
+    ToolIntelligenceNetwork,
+    _create_default_toin_backend,
+    get_toin,
+    reset_toin,
+)
+
+# =============================================================================
+# Fixtures
+# =============================================================================
+
+
+@pytest.fixture(autouse=True)
+def _clean_globals():
+    """Reset global singletons before and after each test."""
+    reset_toin()
+    reset_compression_store()
+    clear_request_compression_store()
+    yield
+    reset_toin()
+    reset_compression_store()
+    clear_request_compression_store()
+
+
+@pytest.fixture
+def tmp_toin_path(tmp_path):
+    """Temporary file path for TOIN storage."""
+    return str(tmp_path / "toin.json")
+
+
+@pytest.fixture
+def fs_backend(tmp_toin_path):
+    """FileSystemTOINBackend with a temp path."""
+    return FileSystemTOINBackend(tmp_toin_path)
+
+
+def _make_tool_signature() -> ToolSignature:
+    """Create a ToolSignature for testing."""
+    return ToolSignature.from_items(
+        [
+            {"id": 1, "name": "test", "score": 0.95},
+            {"id": 2, "name": "test2", "score": 0.80},
+        ]
+    )
+
+
+# =============================================================================
+# 1. TOIN Backend Protocol Conformance
+# =============================================================================
+
+
+class TestTOINBackendProtocol:
+    """Verify FileSystemTOINBackend satisfies the TOINBackend protocol."""
+
+    def test_filesystem_backend_is_toin_backend(self, fs_backend):
+        """FileSystemTOINBackend must satisfy the runtime-checkable TOINBackend protocol."""
+        assert isinstance(fs_backend, TOINBackend)
+
+    def test_load_empty(self, fs_backend):
+        """load() returns empty dict when no file exists."""
+        result = fs_backend.load()
+        assert result == {}
+
+    def test_save_and_load_roundtrip(self, fs_backend):
+        """Data survives a save/load cycle."""
+        data = {
+            "version": "1.0",
+            "patterns": {
+                "abc123": {
+                    "tool_signature_hash": "abc123",
+                    "total_compressions": 42,
+                }
+            },
+        }
+        fs_backend.save(data)
+        loaded = fs_backend.load()
+        assert loaded == data
+
+    def test_save_creates_parent_dirs(self, tmp_path):
+        """save() creates parent directories if they don't exist."""
+        deep_path = str(tmp_path / "a" / "b" / "c" / "toin.json")
+        backend = FileSystemTOINBackend(deep_path)
+        backend.save({"version": "1.0"})
+        loaded = backend.load()
+        assert loaded["version"] == "1.0"
+
+    def test_save_atomic_on_failure(self, tmp_path):
+        """If save fails mid-write, original data is preserved."""
+        path = str(tmp_path / "toin.json")
+        backend = FileSystemTOINBackend(path)
+
+        # Save initial data
+        backend.save({"version": "1.0", "state": "original"})
+
+        # Corrupt the temp dir to force failure (make dir read-only)
+        # This is OS-dependent; we verify the load still returns original
+        loaded = backend.load()
+        assert loaded["state"] == "original"
+
+    def test_load_corrupted_file(self, tmp_toin_path):
+        """load() returns empty dict on corrupted JSON."""
+        with open(tmp_toin_path, "w") as f:
+            f.write("{invalid json!!")
+
+        backend = FileSystemTOINBackend(tmp_toin_path)
+        result = backend.load()
+        assert result == {}
+
+    def test_save_overwrites(self, fs_backend):
+        """save() overwrites previous data completely."""
+        fs_backend.save({"version": "1.0", "old": True})
+        fs_backend.save({"version": "2.0", "new": True})
+        loaded = fs_backend.load()
+        assert loaded == {"version": "2.0", "new": True}
+        assert "old" not in loaded
+
+
+class TestCustomTOINBackend:
+    """Verify any dict-based backend satisfies the protocol."""
+
+    def test_dict_backend_satisfies_protocol(self):
+        """A minimal dict-backed implementation passes protocol check."""
+
+        class DictBackend:
+            def __init__(self):
+                self._data: dict[str, Any] = {}
+
+            def load(self) -> dict[str, Any]:
+                return dict(self._data)
+
+            def save(self, data: dict[str, Any]) -> None:
+                self._data = dict(data)
+
+        backend = DictBackend()
+        assert isinstance(backend, TOINBackend)
+
+        backend.save({"key": "value"})
+        assert backend.load() == {"key": "value"}
+
+
+# =============================================================================
+# 2. TOIN Backend Wiring
+# =============================================================================
+
+
+class TestTOINBackendWiring:
+    """Verify ToolIntelligenceNetwork correctly delegates to backends."""
+
+    def test_toin_with_custom_backend(self):
+        """TOIN uses custom backend for save/load."""
+        store: dict[str, Any] = {}
+
+        class MemBackend:
+            def load(self) -> dict[str, Any]:
+                return dict(store)
+
+            def save(self, data: dict[str, Any]) -> None:
+                store.clear()
+                store.update(data)
+
+        config = TOINConfig(enabled=True, storage_path="", auto_save_interval=0)
+        toin = ToolIntelligenceNetwork(config, backend=MemBackend())
+
+        sig = _make_tool_signature()
+        toin.record_compression(
+            tool_signature=sig,
+            original_count=10,
+            compressed_count=3,
+            original_tokens=500,
+            compressed_tokens=150,
+            strategy="smart_crusher",
+        )
+
+        # Save should go to our backend
+        toin.save()
+        assert "patterns" in store
+        assert len(store["patterns"]) == 1
+
+        # Create a new TOIN instance with same backend — should load patterns
+        toin2 = ToolIntelligenceNetwork(config, backend=MemBackend())
+        stats = toin2.get_stats()
+        assert stats["patterns_tracked"] == 1
+        assert stats["total_compressions"] == 1
+
+    def test_toin_with_none_backend_and_no_path(self):
+        """TOIN works without persistence when no backend and no path."""
+        config = TOINConfig(enabled=True, storage_path="")
+        toin = ToolIntelligenceNetwork(config, backend=None)
+
+        # save() should be a no-op
+        toin.save()
+        stats = toin.get_stats()
+        assert stats["patterns_tracked"] == 0
+
+    def test_toin_default_filesystem_backend(self, tmp_toin_path):
+        """TOIN creates FileSystemTOINBackend when storage_path is set and no backend given."""
+        config = TOINConfig(enabled=True, storage_path=tmp_toin_path)
+        toin = ToolIntelligenceNetwork(config)
+
+        sig = _make_tool_signature()
+        toin.record_compression(
+            tool_signature=sig,
+            original_count=5,
+            compressed_count=2,
+            original_tokens=100,
+            compressed_tokens=40,
+            strategy="default",
+        )
+        toin.save()
+
+        # Verify file was written
+        with open(tmp_toin_path) as f:
+            data = json.load(f)
+        assert "patterns" in data
+        assert len(data["patterns"]) == 1
+
+
+# =============================================================================
+# 3. TOIN Entry Point Loading
+# =============================================================================
+
+
+class TestTOINEntryPointLoading:
+    """Verify _create_default_toin_backend() env-based loading."""
+
+    def test_no_env_returns_none(self, monkeypatch):
+        """No HEADROOM_TOIN_BACKEND env → returns None (use default)."""
+        monkeypatch.delenv("HEADROOM_TOIN_BACKEND", raising=False)
+        assert _create_default_toin_backend() is None
+
+    def test_empty_env_returns_none(self, monkeypatch):
+        """Empty HEADROOM_TOIN_BACKEND → returns None."""
+        monkeypatch.setenv("HEADROOM_TOIN_BACKEND", "")
+        assert _create_default_toin_backend() is None
+
+    def test_filesystem_env_returns_none(self, monkeypatch):
+        """HEADROOM_TOIN_BACKEND=filesystem → returns None (use default)."""
+        monkeypatch.setenv("HEADROOM_TOIN_BACKEND", "filesystem")
+        assert _create_default_toin_backend() is None
+
+    def test_unknown_backend_returns_none(self, monkeypatch):
+        """Unknown backend name with no entry point → returns None with warning."""
+        monkeypatch.setenv("HEADROOM_TOIN_BACKEND", "nonexistent_backend_xyz")
+        result = _create_default_toin_backend()
+        assert result is None
+
+    def test_get_toin_respects_env_backend(self, monkeypatch, tmp_toin_path):
+        """get_toin() uses _create_default_toin_backend() on first call."""
+        monkeypatch.delenv("HEADROOM_TOIN_BACKEND", raising=False)
+        monkeypatch.setenv("HEADROOM_TOIN_PATH", tmp_toin_path)
+
+        toin = get_toin()
+        assert toin is not None
+        # Should use FileSystemTOINBackend since no HEADROOM_TOIN_BACKEND set
+        assert toin._backend is not None
+
+
+# =============================================================================
+# 4. CCR ContextVar Scoping
+# =============================================================================
+
+
+class TestCCRContextVarScoping:
+    """Verify request-scoped CCR stores work correctly."""
+
+    def test_default_returns_global(self):
+        """Without request scope, get_compression_store() returns global singleton."""
+        store1 = get_compression_store()
+        store2 = get_compression_store()
+        assert store1 is store2
+
+    def test_set_request_store_overrides_global(self):
+        """set_request_compression_store() makes get return the request store."""
+        global_store = get_compression_store()
+        request_store = CompressionStore(max_entries=10)
+
+        set_request_compression_store(request_store)
+        assert get_compression_store() is request_store
+        assert get_compression_store() is not global_store
+
+    def test_clear_request_store_restores_global(self):
+        """clear_request_compression_store() restores the global store."""
+        global_store = get_compression_store()
+        request_store = CompressionStore(max_entries=10)
+
+        set_request_compression_store(request_store)
+        assert get_compression_store() is request_store
+
+        clear_request_compression_store()
+        assert get_compression_store() is global_store
+
+    def test_request_store_isolated_per_thread(self):
+        """ContextVars are per-thread — each thread sees its own store."""
+        global_store = get_compression_store()
+        results: dict[str, CompressionStore | None] = {}
+
+        def worker(name: str, store: CompressionStore | None):
+            if store:
+                set_request_compression_store(store)
+            results[name] = get_compression_store()
+            if store:
+                clear_request_compression_store()
+
+        store_a = CompressionStore(max_entries=5)
+        store_b = CompressionStore(max_entries=7)
+
+        t1 = threading.Thread(target=worker, args=("t1", store_a))
+        t2 = threading.Thread(target=worker, args=("t2", store_b))
+        t3 = threading.Thread(target=worker, args=("t3", None))
+
+        t1.start()
+        t2.start()
+        t3.start()
+        t1.join()
+        t2.join()
+        t3.join()
+
+        assert results["t1"] is store_a
+        assert results["t2"] is store_b
+        assert results["t3"] is global_store
+
+    def test_request_store_data_isolation(self):
+        """Data stored in request-scoped store doesn't leak to global."""
+        global_store = get_compression_store()
+
+        request_store = CompressionStore(max_entries=10)
+        set_request_compression_store(request_store)
+
+        # Store data in request store
+        active_store = get_compression_store()
+        hash_key = active_store.store(
+            original='[{"id": 1}]',
+            compressed='[{"id": 1}]',
+            original_tokens=10,
+            compressed_tokens=10,
+        )
+
+        # Verify it's in request store
+        assert active_store.retrieve(hash_key) is not None
+        # Verify it's NOT in global store
+        assert global_store.retrieve(hash_key) is None
+
+        clear_request_compression_store()
+
+
+# =============================================================================
+# 5. CCR Entry Point Loading
+# =============================================================================
+
+
+class TestCCREntryPointLoading:
+    """Verify _create_default_ccr_backend() env-based loading."""
+
+    def test_no_env_returns_none(self, monkeypatch):
+        """No HEADROOM_CCR_BACKEND → returns None (use InMemoryBackend)."""
+        monkeypatch.delenv("HEADROOM_CCR_BACKEND", raising=False)
+        from headroom.cache.compression_store import _create_default_ccr_backend
+
+        assert _create_default_ccr_backend() is None
+
+    def test_memory_env_returns_none(self, monkeypatch):
+        """HEADROOM_CCR_BACKEND=memory → returns None (use default)."""
+        monkeypatch.setenv("HEADROOM_CCR_BACKEND", "memory")
+        from headroom.cache.compression_store import _create_default_ccr_backend
+
+        assert _create_default_ccr_backend() is None
+
+    def test_unknown_backend_returns_none(self, monkeypatch):
+        """Unknown backend with no entry point → returns None."""
+        monkeypatch.setenv("HEADROOM_CCR_BACKEND", "nonexistent_backend_xyz")
+        from headroom.cache.compression_store import _create_default_ccr_backend
+
+        assert _create_default_ccr_backend() is None
+
+    def test_inmemory_backend_satisfies_protocol(self):
+        """InMemoryBackend satisfies the CompressionStoreBackend protocol."""
+        backend = InMemoryBackend()
+        assert isinstance(backend, CompressionStoreBackend)
+
+
+# =============================================================================
+# 6. Storage Entry Point Loading
+# =============================================================================
+
+
+class TestStorageEntryPointLoading:
+    """Verify create_storage() scheme-based loading."""
+
+    def test_sqlite_scheme(self, tmp_path):
+        """sqlite:// scheme creates SQLiteStorage."""
+        from headroom.storage.sqlite import SQLiteStorage
+
+        store = create_storage(f"sqlite:///{tmp_path}/test.db")
+        assert isinstance(store, SQLiteStorage)
+
+    def test_jsonl_scheme(self, tmp_path):
+        """jsonl:// scheme creates JSONLStorage."""
+        from headroom.storage.jsonl import JSONLStorage
+
+        store = create_storage(f"jsonl:///{tmp_path}/test.jsonl")
+        assert isinstance(store, JSONLStorage)
+
+    def test_unknown_scheme_without_entry_point(self, tmp_path):
+        """Unknown scheme without entry point falls back to SQLiteStorage."""
+        from headroom.storage.sqlite import SQLiteStorage
+
+        # This should fall back to SQLite (legacy behavior)
+        store = create_storage(str(tmp_path / "test.db"))
+        assert isinstance(store, SQLiteStorage)
+
+    def test_storage_base_is_abstract(self):
+        """Storage ABC requires all methods to be implemented."""
+        assert hasattr(Storage, "save")
+        assert hasattr(Storage, "get")
+        assert hasattr(Storage, "query")
+        assert hasattr(Storage, "count")
+        assert hasattr(Storage, "iter_all")
+        assert hasattr(Storage, "get_summary_stats")
+
+
+# =============================================================================
+# 7. Integration: Full Adapter Lifecycle
+# =============================================================================
+
+
+class TestAdapterLifecycle:
+    """End-to-end test of the adapter pattern."""
+
+    def test_ccr_with_custom_backend(self):
+        """CompressionStore works with a custom backend implementation."""
+
+        class ListBackend:
+            """Minimal backend that tracks all operations."""
+
+            def __init__(self):
+                self._store: dict[str, Any] = {}
+                self.ops: list[str] = []
+
+            def get(self, hash_key):
+                self.ops.append(f"get:{hash_key[:8]}")
+                return self._store.get(hash_key)
+
+            def set(self, hash_key, entry):
+                self.ops.append(f"set:{hash_key[:8]}")
+                self._store[hash_key] = entry
+
+            def delete(self, hash_key):
+                self.ops.append(f"delete:{hash_key[:8]}")
+                if hash_key in self._store:
+                    del self._store[hash_key]
+                    return True
+                return False
+
+            def exists(self, hash_key):
+                return hash_key in self._store
+
+            def clear(self):
+                self._store.clear()
+
+            def count(self):
+                return len(self._store)
+
+            def keys(self):
+                return list(self._store.keys())
+
+            def items(self):
+                return list(self._store.items())
+
+            def get_stats(self):
+                return {"backend_type": "list", "entry_count": len(self._store)}
+
+        backend = ListBackend()
+        store = CompressionStore(backend=backend)
+
+        # Store and retrieve
+        hash_key = store.store(
+            original='[{"id": 1, "name": "test"}]',
+            compressed='[{"id": 1}]',
+            original_tokens=50,
+            compressed_tokens=20,
+        )
+        entry = store.retrieve(hash_key)
+
+        assert entry is not None
+        assert entry.original_tokens == 50
+        assert any("set:" in op for op in backend.ops)
+        assert any("get:" in op for op in backend.ops)
+
+    def test_toin_save_load_preserves_patterns(self, tmp_toin_path):
+        """Patterns survive save/load via backend."""
+        config = TOINConfig(storage_path=tmp_toin_path)
+        toin = ToolIntelligenceNetwork(config)
+
+        sig = _make_tool_signature()
+
+        # Record multiple events
+        for _i in range(15):
+            toin.record_compression(
+                tool_signature=sig,
+                original_count=50,
+                compressed_count=10,
+                original_tokens=1000,
+                compressed_tokens=200,
+                strategy="smart_crusher",
+            )
+
+        toin.save()
+
+        # New instance loads from same backend
+        toin2 = ToolIntelligenceNetwork(TOINConfig(storage_path=tmp_toin_path))
+        stats = toin2.get_stats()
+        assert stats["patterns_tracked"] >= 1
+        assert stats["total_compressions"] >= 15
+
+        # Recommendations should work
+        hint = toin2.get_recommendation(sig)
+        assert hint.based_on_samples >= 15