diff --git a/packages/mizan-django/src/mizan/cache/__init__.py b/packages/mizan-django/src/mizan/cache/__init__.py
index b06a364..1aeb18a 100644
--- a/packages/mizan-django/src/mizan/cache/__init__.py
+++ b/packages/mizan-django/src/mizan/cache/__init__.py
@@ -1,21 +1,12 @@
 """
 mizan.cache — Origin-side cache implementing the Mizan cache protocol.
 
-This module provides the same cache semantics as the Edge layer:
-- HMAC-SHA256 key derivation (JSON-canonical form)
-- Scoped invalidation (purge by context + params)
-- Reverse indexes for efficient purge lookups
+Simple key-value cache with HMAC-derived keys. No reverse indexes.
+Scoped purge recomputes the key and deletes directly.
+Broad purge uses key-prefix scan (rare operation).
 
 Usage:
     from mizan.cache import get_cache, cache_get, cache_put, cache_purge
-
-Backends:
-    - MemoryCache: for testing (no dependencies)
-    - RedisCache: for production (requires redis-py)
-
-Configuration (Django settings):
-    MIZAN_CACHE_SECRET = "your-signing-secret"
-    MIZAN_CACHE_REDIS_URL = "redis://localhost:6379/0"
 """
 
 from __future__ import annotations
@@ -25,7 +16,7 @@ import threading
 from typing import Any
 
 from .backend import CacheBackend, MemoryCache, RedisCache
-from .keys import derive_cache_key, build_index_keys
+from .keys import derive_cache_key, CONTEXT_KEY_PREFIX
 
 logger = logging.getLogger("mizan.cache")
 
@@ -37,10 +28,7 @@ _init_lock = threading.Lock()
 def get_cache() -> CacheBackend | None:
     """
     Get the configured cache backend, or None if caching is disabled.
-
-    Returns RedisCache if MIZAN_CACHE_SECRET and MIZAN_CACHE_REDIS_URL are
-    both set. Returns None otherwise. The instance is cached for the process
-    lifetime. Thread-safe.
+    Thread-safe.
     """
     global _cache_instance, _initialized
     if _initialized:
@@ -76,18 +64,14 @@ def get_cache() -> CacheBackend | None:
 
 
 def set_cache(backend: CacheBackend | None) -> None:
-    """
-    Override the cache backend. Primarily for testing.
-
-    Pass None to disable caching. Pass a MemoryCache instance for tests.
-    """
+    """Override the cache backend. For testing."""
     global _cache_instance, _initialized
     _cache_instance = backend
     _initialized = True
 
 
 def reset_cache() -> None:
-    """Reset the cache to uninitialized state. For testing teardown."""
+    """Reset to uninitialized state. For testing teardown."""
     global _cache_instance, _initialized
     _cache_instance = None
     _initialized = False
@@ -101,11 +85,7 @@ def cache_get(
     user_id: str | None = None,
     rev: int = 0,
 ) -> bytes | None:
-    """
-    Look up a cached context response.
-
-    Returns the cached response bytes, or None on miss.
-    """
+    """Look up a cached context response."""
     key = derive_cache_key(secret, context, params, user_id, rev)
     return backend.get(key)
 
@@ -119,65 +99,34 @@ def cache_put(
     user_id: str | None = None,
     rev: int = 0,
 ) -> None:
-    """
-    Store a context response in the cache with reverse indexes.
-    """
+    """Store a context response in the cache."""
     key = derive_cache_key(secret, context, params, user_id, rev)
-    indexes = build_index_keys(context, params)
-    backend.put(key, value, indexes)
+    backend.set(key, value)
 
 
 def cache_purge(
     backend: CacheBackend,
     context: str,
     params: dict[str, Any] | None = None,
+    secret: str | None = None,
+    user_id: str | None = None,
+    rev: int = 0,
 ) -> int:
     """
-    Purge cached entries for a context, optionally scoped by params.
+    Purge cached entries for a context.
 
-    If params is None, purges ALL entries for the context (broad invalidation).
-    If params is provided, purges only entries matching those specific params
-    (scoped invalidation via the reverse index).
+    Scoped purge (params provided): recomputes the HMAC key and deletes
+    it directly. One DELETE, no index needed.
 
-    Returns the number of entries purged.
+    Broad purge (no params): scans by key prefix "ctx:{context}:*".
+    This is a rare operation (Tier 3 fallback in invalidation).
     """
-    if params:
-        # Scoped purge — intersect index lookups (AND semantics)
-        # Entry must match ALL params, not just any one.
-        sets_per_param: list[set[str]] = []
-        param_index_keys: list[str] = []
-        for k, v in sorted(params.items()):
-            index_key = f"mizan:idx:{context}:{k}={str(v)}"
-            param_index_keys.append(index_key)
-            sets_per_param.append(backend.get_index(index_key))
-
-        if sets_per_param:
-            keys_to_delete = sets_per_param[0]
-            for s in sets_per_param[1:]:
-                keys_to_delete = keys_to_delete & s
-        else:
-            keys_to_delete = set()
-
-        if keys_to_delete:
-            # Clean up per-param indexes
-            for idx_key in param_index_keys:
-                backend.remove_from_index(idx_key, keys_to_delete)
-            # Remove from the broad context index too
-            backend.remove_from_index(f"mizan:idx:{context}", keys_to_delete)
-            return backend.delete_many(list(keys_to_delete))
-        return 0
+    if params and secret:
+        key = derive_cache_key(secret, context, params, user_id, rev)
+        return 1 if backend.delete(key) else 0
     else:
-        # Broad purge — delete everything in the context index
-        index_key = f"mizan:idx:{context}"
-        keys_to_delete = backend.get_index(index_key)
-        backend.delete_index(index_key)
-
-        # Clean up per-param sub-indexes (e.g., mizan:idx:user:user_id=5)
-        backend.delete_indexes_by_prefix(f"mizan:idx:{context}:")
-
-        if keys_to_delete:
-            return backend.delete_many(list(keys_to_delete))
-        return 0
+        prefix = f"{CONTEXT_KEY_PREFIX}{context}:"
+        return backend.delete_by_prefix(prefix)
 
 
 __all__ = [
diff --git a/packages/mizan-django/src/mizan/cache/backend.py b/packages/mizan-django/src/mizan/cache/backend.py
index d531b9c..fd95f2c 100644
--- a/packages/mizan-django/src/mizan/cache/backend.py
+++ b/packages/mizan-django/src/mizan/cache/backend.py
@@ -1,9 +1,9 @@
 """
 Cache backends — MemoryCache (testing) and RedisCache (production).
 
-Both implement the same interface. MemoryCache requires no dependencies and
-is used in the test suite. RedisCache requires `redis-py` and is used in
-production when MIZAN_CACHE_REDIS_URL is configured.
+Simple key-value stores. No reverse indexes. Cache keys are derived
+from HMAC, so scoped purge just recomputes the key and deletes it.
+Broad purge uses key-prefix scan (rare operation).
 """
 
 from __future__ import annotations
@@ -15,12 +15,9 @@ class CacheBackend(Protocol):
     """Interface that all Mizan cache backends implement."""
 
     def get(self, key: str) -> bytes | None: ...
-    def put(self, key: str, value: bytes, indexes: list[str]) -> None: ...
-    def delete_many(self, keys: list[str]) -> int: ...
-    def get_index(self, index_key: str) -> set[str]: ...
-    def remove_from_index(self, index_key: str, members: set[str]) -> None: ...
-    def delete_index(self, index_key: str) -> None: ...
-    def delete_indexes_by_prefix(self, prefix: str) -> None: ...
+    def set(self, key: str, value: bytes) -> None: ...
+    def delete(self, key: str) -> bool: ...
+    def delete_by_prefix(self, prefix: str) -> int: ...
     def clear(self) -> None: ...
 
 
@@ -28,83 +25,64 @@ class MemoryCache:
     """
     In-memory cache backend for testing.
 
-    Uses Python dicts. Same API as RedisCache. No persistence, no
-    cross-process sharing. Perfect for unit tests.
+    Uses a Python dict. No persistence, no cross-process sharing.
     """
 
     def __init__(self) -> None:
         self._store: dict[str, bytes] = {}
-        self._indexes: dict[str, set[str]] = {}
 
     def get(self, key: str) -> bytes | None:
         return self._store.get(key)
 
-    def put(self, key: str, value: bytes, indexes: list[str]) -> None:
+    def set(self, key: str, value: bytes) -> None:
         self._store[key] = value
-        for idx in indexes:
-            if idx not in self._indexes:
-                self._indexes[idx] = set()
-            self._indexes[idx].add(key)
 
-    def delete_many(self, keys: list[str]) -> int:
-        count = 0
-        for key in keys:
-            if key in self._store:
-                del self._store[key]
-                count += 1
-        return count
+    def delete(self, key: str) -> bool:
+        if key in self._store:
+            del self._store[key]
+            return True
+        return False
 
-    def get_index(self, index_key: str) -> set[str]:
-        return self._indexes.get(index_key, set()).copy()
-
-    def remove_from_index(self, index_key: str, members: set[str]) -> None:
-        if index_key in self._indexes:
-            self._indexes[index_key] -= members
-            if not self._indexes[index_key]:
-                del self._indexes[index_key]
-
-    def delete_index(self, index_key: str) -> None:
-        self._indexes.pop(index_key, None)
-
-    def delete_indexes_by_prefix(self, prefix: str) -> None:
-        to_delete = [k for k in self._indexes if k.startswith(prefix)]
+    def delete_by_prefix(self, prefix: str) -> int:
+        to_delete = [k for k in self._store if k.startswith(prefix)]
         for k in to_delete:
-            del self._indexes[k]
+            del self._store[k]
+        return len(to_delete)
 
     def clear(self) -> None:
         self._store.clear()
-        self._indexes.clear()
 
 
 class RedisCache:
     """
     Redis-backed cache backend for production.
 
-    Uses Redis strings for cache entries and Redis sets for reverse indexes.
-    Requires `redis-py` (pip install mizan[cache]).
+    Simple GET/SET/DEL. No reverse indexes. Scoped purge recomputes
+    the HMAC key and deletes directly. Broad purge uses SCAN.
     """
 
-    # Safety-net TTL: entries expire even if purge fails (24 hours)
-    DEFAULT_TTL = 86400
+    DEFAULT_TTL = 86400  # 24h safety-net
 
     def __init__(
         self,
         redis_url: str,
-        prefix: str = "mizan:cache:",
+        prefix: str = "mizan:",
         ttl: int | None = None,
     ) -> None:
         try:
-            import redis
+            import redis as redis_lib
         except ImportError:
             raise ImportError(
                 "Redis is required for Mizan's cache backend. "
                 "Install it with: pip install mizan[cache]"
             )
-        self._client = redis.from_url(
+        self._client = redis_lib.from_url(
             redis_url,
             socket_connect_timeout=5,
             socket_timeout=5,
             health_check_interval=30,
+            retry_on_timeout=True,
+            max_connections=50,
         )
         self._prefix = prefix
         self._ttl = ttl if ttl is not None else self.DEFAULT_TTL
@@ -113,57 +91,25 @@ class RedisCache:
         return f"{self._prefix}{key}"
 
     def get(self, key: str) -> bytes | None:
-        result = self._client.get(self._key(key))
-        return result
+        return self._client.get(self._key(key))
 
-    def put(self, key: str, value: bytes, indexes: list[str]) -> None:
-        prefixed_key = self._key(key)
-        pipe = self._client.pipeline(transaction=True)
-        pipe.set(prefixed_key, value, ex=self._ttl)
-        for idx in indexes:
-            pipe.sadd(self._key(idx), key)
-            pipe.expire(self._key(idx), self._ttl)
-        pipe.execute()
+    def set(self, key: str, value: bytes) -> None:
+        self._client.set(self._key(key), value, ex=self._ttl)
 
-    def delete_many(self, keys: list[str]) -> int:
-        if not keys:
-            return 0
-        prefixed = [self._key(k) for k in keys]
-        return self._client.delete(*prefixed)
+    def delete(self, key: str) -> bool:
+        return self._client.unlink(self._key(key)) > 0
 
-    def get_index(self, index_key: str) -> set[str]:
-        members = self._client.smembers(self._key(index_key))
-        return {m.decode("utf-8") if isinstance(m, bytes) else m for m in members}
-
-    def remove_from_index(self, index_key: str, members: set[str]) -> None:
-        if members:
-            self._client.srem(self._key(index_key), *members)
-
-    def delete_index(self, index_key: str) -> None:
-        self._client.delete(self._key(index_key))
-
-    def delete_indexes_by_prefix(self, prefix: str) -> None:
+    def delete_by_prefix(self, prefix: str) -> int:
         pattern = f"{self._prefix}{prefix}*"
+        count = 0
         cursor = 0
         while True:
-            cursor, keys = self._client.scan(cursor, match=pattern, count=100)
+            cursor, keys = self._client.scan(cursor, match=pattern, count=1000)
             if keys:
-                pipe = self._client.pipeline()
-                for key in keys:
-                    pipe.unlink(key)
-                pipe.execute()
+                count += self._client.unlink(*keys)
             if cursor == 0:
                 break
+        return count
 
     def clear(self) -> None:
-        pattern = f"{self._prefix}*"
-        cursor = 0
-        while True:
-            cursor, keys = self._client.scan(cursor, match=pattern, count=100)
-            if keys:
-                pipe = self._client.pipeline()
-                for key in keys:
-                    pipe.unlink(key)
-                pipe.execute()
-            if cursor == 0:
-                break
+        self.delete_by_prefix("")
diff --git a/packages/mizan-django/src/mizan/cache/keys.py b/packages/mizan-django/src/mizan/cache/keys.py
index 537bdce..ce7632c 100644
--- a/packages/mizan-django/src/mizan/cache/keys.py
+++ b/packages/mizan-django/src/mizan/cache/keys.py
@@ -1,18 +1,11 @@
 """
 Cache key derivation — HMAC-SHA256 over JSON-canonical form.
 
-This is the protocol-critical function. Every Mizan adapter (Python, TypeScript,
-future languages) must produce identical output for identical inputs. The
-conformance test suite verifies this.
+Protocol-critical: every Mizan adapter must produce identical output
+for identical inputs. Cross-language conformance verified by pin tests.
 
-Wire format:
-    HMAC-SHA256(secret, JSON.stringify({"c": ctx, "p": params, "r": rev}, sort_keys))
-
-    - Keys are sorted (JSON canonical form)
-    - No whitespace (separators=(",", ":"))
-    - "u" key included only for user-scoped content
-    - Params are sorted by key name
-    - All param values are stringified
+Scoped purge recomputes the key directly — no reverse index needed.
+Broad purge uses a context prefix scan.
 """
 
 from __future__ import annotations
@@ -22,6 +15,9 @@ import hmac
 import json
 from typing import Any
 
+# Context prefix for broad purge (SCAN pattern)
+CONTEXT_KEY_PREFIX = "ctx:"
+
 
 def derive_cache_key(
     secret: str,
@@ -33,17 +29,9 @@ def derive_cache_key(
     """
     Derive a deterministic HMAC-SHA256 cache key.
 
-    Args:
-        secret: The MIZAN_CACHE_SECRET signing key.
-        context: Context name (e.g., "user", "products").
-        params: Query parameters, sorted internally by key.
-        user_id: User ID for user-scoped content. None for public.
-        rev: Revision number from @client(rev=N). Defaults to 0.
-
-    Returns:
-        Hex-encoded HMAC-SHA256 digest (64 characters).
+    Returns a prefixed key: "ctx:{context}:{hmac_hex}" so that
+    broad purge can SCAN by prefix "ctx:{context}:*".
     """
-    # Stringify all param values for cross-language determinism
     sorted_params = {k: str(v) for k, v in sorted(params.items())}
 
     key_data: dict[str, Any] = {"c": context, "p": sorted_params, "r": rev}
@@ -51,29 +39,10 @@ def derive_cache_key(
         key_data["u"] = str(user_id)
 
     message = json.dumps(key_data, sort_keys=True, separators=(",", ":"))
-    return hmac.new(
+    hmac_hex = hmac.new(
         secret.encode("utf-8"),
         message.encode("utf-8"),
         hashlib.sha256,
     ).hexdigest()
 
-
-def build_index_keys(context: str, params: dict[str, Any]) -> list[str]:
-    """
-    Build reverse index keys for a cache entry.
-
-    Returns a list of index keys that this entry should be added to:
-    - "mizan:idx:{context}" (broad — for purging entire context)
-    - "mizan:idx:{context}:{key}={value}" (scoped — for per-param purging)
-
-    Args:
-        context: Context name.
-        params: Query parameters.
-
-    Returns:
-        List of index key strings.
-    """
-    keys = [f"mizan:idx:{context}"]
-    for k, v in sorted(params.items()):
-        keys.append(f"mizan:idx:{context}:{k}={str(v)}")
-    return keys
+    return f"{CONTEXT_KEY_PREFIX}{context}:{hmac_hex}"
diff --git a/packages/mizan-django/src/mizan/client/executor.py b/packages/mizan-django/src/mizan/client/executor.py
index 771baf2..bde0967 100644
--- a/packages/mizan-django/src/mizan/client/executor.py
+++ b/packages/mizan-django/src/mizan/client/executor.py
@@ -708,13 +708,19 @@ def function_call_view(request: HttpRequest) -> JsonResponse:
         # Purge origin-side cache for invalidated contexts
         _cache_log = logging.getLogger("mizan.cache")
         cache = get_cache()
+        cache_settings = get_settings()
         if cache is not None:
             try:
                 for entry in invalidate_contexts:
                     if isinstance(entry, str):
+                        # Broad purge (no params) — prefix scan
                         cache_purge(cache, entry)
                     elif isinstance(entry, dict):
-                        cache_purge(cache, entry["context"], entry.get("params"))
+                        # Scoped purge — recompute key and delete directly
+                        cache_purge(
+                            cache, entry["context"], entry.get("params"),
+                            secret=cache_settings.cache_secret,
+                        )
             except Exception:
                 _cache_log.warning("Cache purge failed", exc_info=True)
 
diff --git a/packages/mizan-django/src/mizan/tests/test_core.py b/packages/mizan-django/src/mizan/tests/test_core.py
index 7f7ce74..3bb0106 100644
--- a/packages/mizan-django/src/mizan/tests/test_core.py
+++ b/packages/mizan-django/src/mizan/tests/test_core.py
@@ -2798,7 +2798,8 @@ class CacheKeyDerivationTests(TestCase):
         key1 = derive_cache_key(self.SECRET, "user", {"user_id": "5"})
         key2 = derive_cache_key(self.SECRET, "user", {"user_id": "5"})
         self.assertEqual(key1, key2)
-        self.assertEqual(len(key1), 64)  # SHA-256 hex digest
+        self.assertTrue(key1.startswith("ctx:user:"))
+        self.assertEqual(len(key1), len("ctx:user:") + 64)  # prefix + SHA-256 hex
 
     def test_param_order_irrelevant(self):
         """Parameter ordering does not affect the key."""
@@ -2850,7 +2851,7 @@ class CacheKeyDerivationTests(TestCase):
         public_key = derive_cache_key(pin_secret, "user", {"user_id": "5"}, rev=0)
         self.assertEqual(
             public_key,
-            "605a1ca5ad5994e9b765c8d1b330474c2a0d51a7b8fbbdc402f992da7ba902f6",
+            "ctx:user:605a1ca5ad5994e9b765c8d1b330474c2a0d51a7b8fbbdc402f992da7ba902f6",
         )
 
         user_scoped_key = derive_cache_key(
@@ -2858,7 +2859,7 @@ class CacheKeyDerivationTests(TestCase):
         )
         self.assertEqual(
             user_scoped_key,
-            "30fc08eb46ee4ff2cf7d317e97dca90fd616511e0587304416f71dc863338dc2",
+            "ctx:user:30fc08eb46ee4ff2cf7d317e97dca90fd616511e0587304416f71dc863338dc2",
         )
 
 
@@ -2873,33 +2874,33 @@ class CacheBackendTests(TestCase):
         """Empty cache returns None."""
         self.assertIsNone(self.cache.get("nonexistent"))
 
-    def test_put_then_get(self):
+    def test_set_then_get(self):
         """Store and retrieve a value."""
-        self.cache.put("key1", b'{"data": true}', ["mizan:idx:ctx"])
+        self.cache.set("key1", b'{"data": true}')
         result = self.cache.get("key1")
         self.assertEqual(result, b'{"data": true}')
 
-    def test_index_tracking(self):
-        """Put adds the key to specified indexes."""
-        self.cache.put("key1", b"v1", ["mizan:idx:user", "mizan:idx:user:user_id=5"])
-        self.assertIn("key1", self.cache.get_index("mizan:idx:user"))
-        self.assertIn("key1", self.cache.get_index("mizan:idx:user:user_id=5"))
-
-    def test_delete_many(self):
-        """Delete multiple keys at once."""
-        self.cache.put("k1", b"v1", [])
-        self.cache.put("k2", b"v2", [])
-        count = self.cache.delete_many(["k1", "k2"])
-        self.assertEqual(count, 2)
+    def test_delete(self):
+        """Delete a key."""
+        self.cache.set("k1", b"v1")
+        self.assertTrue(self.cache.delete("k1"))
         self.assertIsNone(self.cache.get("k1"))
-        self.assertIsNone(self.cache.get("k2"))
+
+    def test_delete_by_prefix(self):
+        """Delete by prefix removes matching keys only."""
+        self.cache.set("ctx:user:abc", b"v1")
+        self.cache.set("ctx:user:def", b"v2")
+        self.cache.set("ctx:products:ghi", b"v3")
+        count = self.cache.delete_by_prefix("ctx:user:")
+        self.assertEqual(count, 2)
+        self.assertIsNone(self.cache.get("ctx:user:abc"))
+        self.assertIsNotNone(self.cache.get("ctx:products:ghi"))
 
     def test_clear(self):
         """Clear removes everything."""
-        self.cache.put("k1", b"v1", ["idx1"])
+        self.cache.set("k1", b"v1")
         self.cache.clear()
         self.assertIsNone(self.cache.get("k1"))
-        self.assertEqual(self.cache.get_index("idx1"), set())
 
 
 class CachePurgeTests(TestCase):
@@ -2923,10 +2924,10 @@ class CachePurgeTests(TestCase):
         reset_cache()
 
     def test_scoped_purge(self):
-        """Purging user;user_id=5 removes only that entry."""
+        """Purging user;user_id=5 recomputes key and deletes directly."""
         from mizan.cache import cache_purge, cache_get
 
-        count = cache_purge(self.cache, "user", {"user_id": "5"})
+        count = cache_purge(self.cache, "user", {"user_id": "5"}, secret=self.SECRET)
         self.assertEqual(count, 1)
 
         # user_id=5 is gone
@@ -3515,80 +3516,62 @@ class RedisCacheBackendTests(TestCase):
         """Empty cache returns None."""
         self.assertIsNone(self.cache.get("nonexistent"))
 
-    def test_put_then_get(self):
+    def test_set_then_get(self):
         """Store and retrieve a value."""
-        self.cache.put("key1", b'{"data": true}', ["mizan:idx:ctx"])
+        self.cache.set("key1", b'{"data": true}')
         result = self.cache.get("key1")
         self.assertEqual(result, b'{"data": true}')
 
-    def test_index_tracking(self):
-        """Put adds the key to specified indexes."""
-        self.cache.put("key1", b"v1", ["mizan:idx:user", "mizan:idx:user:user_id=5"])
-        self.assertIn("key1", self.cache.get_index("mizan:idx:user"))
-        self.assertIn("key1", self.cache.get_index("mizan:idx:user:user_id=5"))
-
-    def test_delete_many(self):
-        """Delete multiple keys at once."""
-        self.cache.put("k1", b"v1", [])
-        self.cache.put("k2", b"v2", [])
-        count = self.cache.delete_many(["k1", "k2"])
-        self.assertEqual(count, 2)
+    def test_delete(self):
+        """Delete a key."""
+        self.cache.set("k1", b"v1")
+        self.assertTrue(self.cache.delete("k1"))
         self.assertIsNone(self.cache.get("k1"))
-        self.assertIsNone(self.cache.get("k2"))
 
-    def test_remove_from_index(self):
-        """Remove specific members from an index."""
-        self.cache.put("k1", b"v1", ["mizan:idx:ctx"])
-        self.cache.put("k2", b"v2", ["mizan:idx:ctx"])
-        self.cache.remove_from_index("mizan:idx:ctx", {"k1"})
-        idx = self.cache.get_index("mizan:idx:ctx")
-        self.assertNotIn("k1", idx)
-        self.assertIn("k2", idx)
+    def test_delete_nonexistent(self):
+        """Delete a nonexistent key returns False."""
+        self.assertFalse(self.cache.delete("ghost"))
 
-    def test_delete_indexes_by_prefix(self):
-        """Delete all indexes matching a prefix."""
-        self.cache.put("k1", b"v1", ["mizan:idx:user:user_id=5"])
-        self.cache.put("k2", b"v2", ["mizan:idx:user:user_id=6"])
-        self.cache.delete_indexes_by_prefix("mizan:idx:user:")
-        self.assertEqual(self.cache.get_index("mizan:idx:user:user_id=5"), set())
-        self.assertEqual(self.cache.get_index("mizan:idx:user:user_id=6"), set())
+    def test_delete_by_prefix(self):
+        """Delete all keys matching a prefix."""
+        self.cache.set("ctx:user:abc", b"v1")
+        self.cache.set("ctx:user:def", b"v2")
+        self.cache.set("ctx:products:ghi", b"v3")
+        count = self.cache.delete_by_prefix("ctx:user:")
+        self.assertEqual(count, 2)
+        self.assertIsNone(self.cache.get("ctx:user:abc"))
+        self.assertIsNone(self.cache.get("ctx:user:def"))
+        self.assertIsNotNone(self.cache.get("ctx:products:ghi"))
 
     def test_ttl_is_set(self):
-        """Put sets a TTL on the cache key."""
+        """Set applies a TTL on the cache key."""
         import redis
-        self.cache.put("ttl_key", b"value", [])
+        self.cache.set("ttl_key", b"value")
         client = redis.from_url(REDIS_URL)
-        ttl = client.ttl(f"mizan:test:ttl_key")
+        ttl = client.ttl("mizan:test:ttl_key")
+        client.close()
         self.assertGreater(ttl, 0)
         self.assertLessEqual(ttl, self.cache._ttl)
 
-    def test_index_ttl_is_set(self):
-        """Put sets a TTL on the index keys too."""
-        import redis
-        self.cache.put("k1", b"v1", ["mizan:idx:ctx"])
-        client = redis.from_url(REDIS_URL)
-        ttl = client.ttl(f"mizan:test:mizan:idx:ctx")
-        self.assertGreater(ttl, 0)
-
     def test_clear(self):
-        """Clear removes all mizan keys."""
-        self.cache.put("k1", b"v1", ["mizan:idx:ctx"])
-        self.cache.put("k2", b"v2", ["mizan:idx:ctx2"])
+        """Clear removes all keys with our prefix."""
+        self.cache.set("k1", b"v1")
+        self.cache.set("k2", b"v2")
         self.cache.clear()
         self.assertIsNone(self.cache.get("k1"))
         self.assertIsNone(self.cache.get("k2"))
-        self.assertEqual(self.cache.get_index("mizan:idx:ctx"), set())
 
-    def test_pipeline_atomicity(self):
-        """Put writes value and indexes in a single pipeline."""
-        self.cache.put("atomic_key", b"atomic_val", [
-            "mizan:idx:ctx",
-            "mizan:idx:ctx:id=1",
-        ])
-        # All should be present (no partial writes)
-        self.assertEqual(self.cache.get("atomic_key"), b"atomic_val")
-        self.assertIn("atomic_key", self.cache.get_index("mizan:idx:ctx"))
-        self.assertIn("atomic_key", self.cache.get_index("mizan:idx:ctx:id=1"))
+    def test_clear_preserves_other_prefixes(self):
+        """Clear only removes keys with our prefix, not others."""
+        import redis
+        client = redis.from_url(REDIS_URL)
+        client.set("other:key", "should_survive")
+        self.cache.set("k1", b"v1")
+        self.cache.clear()
+        self.assertIsNone(self.cache.get("k1"))
+        self.assertEqual(client.get("other:key"), b"should_survive")
+        client.delete("other:key")
+        client.close()
 
 
 class RedisCachePurgeTests(TestCase):
@@ -3612,27 +3595,21 @@ class RedisCachePurgeTests(TestCase):
             self.cache.clear()
 
     def test_scoped_purge(self):
-        """Scoped purge removes only matching entries."""
+        """Scoped purge recomputes key and deletes directly."""
         from mizan.cache import cache_purge, cache_get
 
-        count = cache_purge(self.cache, "user", {"user_id": "5"})
+        count = cache_purge(
+            self.cache, "user", {"user_id": "5"}, secret=self.SECRET,
+        )
         self.assertEqual(count, 1)
         self.assertIsNone(cache_get(self.SECRET, self.cache, "user", {"user_id": "5"}))
         self.assertIsNotNone(cache_get(self.SECRET, self.cache, "user", {"user_id": "6"}))
 
     def test_broad_purge(self):
-        """Broad purge removes all entries in context."""
+        """Broad purge uses prefix scan to remove all entries in context."""
         from mizan.cache import cache_purge, cache_get
 
         count = cache_purge(self.cache, "user")
         self.assertEqual(count, 2)
         self.assertIsNone(cache_get(self.SECRET, self.cache, "user", {"user_id": "5"}))
         self.assertIsNone(cache_get(self.SECRET, self.cache, "user", {"user_id": "6"}))
-
-    def test_broad_purge_cleans_sub_indexes(self):
-        """Broad purge removes per-param sub-indexes."""
-        from mizan.cache import cache_purge
-
-        cache_purge(self.cache, "user")
-        self.assertEqual(self.cache.get_index("mizan:idx:user:user_id=5"), set())
-        self.assertEqual(self.cache.get_index("mizan:idx:user:user_id=6"), set())
diff --git a/packages/mizan-ts/src/cache/backend.ts b/packages/mizan-ts/src/cache/backend.ts
index 6d1fd9a..1dcac45 100644
--- a/packages/mizan-ts/src/cache/backend.ts
+++ b/packages/mizan-ts/src/cache/backend.ts
@@ -1,70 +1,44 @@
 /**
  * Cache backends — MemoryCache for testing.
  *
- * Same API as Python's MemoryCache. RedisCache is implemented
- * per-adapter for production (not included here).
+ * Simple key-value store. No reverse indexes.
  */
 
 export interface CacheBackend {
     get(key: string): string | null
-    put(key: string, value: string, indexes: string[]): void
-    deleteMany(keys: string[]): number
-    getIndex(indexKey: string): Set<string>
-    removeFromIndex(indexKey: string, members: Set<string>): void
-    deleteIndex(indexKey: string): void
-    deleteIndexesByPrefix(prefix: string): void
+    set(key: string, value: string): void
+    delete(key: string): boolean
+    deleteByPrefix(prefix: string): number
     clear(): void
 }
 
 export class MemoryCache implements CacheBackend {
     private _store = new Map<string, string>()
-    private _indexes = new Map<string, Set<string>>()
 
     get(key: string): string | null {
         return this._store.get(key) ?? null
     }
 
-    put(key: string, value: string, indexes: string[]): void {
+    set(key: string, value: string): void {
         this._store.set(key, value)
-        for (const idx of indexes) {
-            if (!this._indexes.has(idx)) {
-                this._indexes.set(idx, new Set())
-            }
-            this._indexes.get(idx)!.add(key)
-        }
     }
 
-    deleteMany(keys: string[]): number {
+    delete(key: string): boolean {
+        return this._store.delete(key)
+    }
+
+    deleteByPrefix(prefix: string): number {
         let count = 0
-        for (const key of keys) {
-            if (this._store.delete(key)) count++
+        for (const key of [...this._store.keys()]) {
+            if (key.startsWith(prefix)) {
+                this._store.delete(key)
+                count++
+            }
         }
         return count
     }
 
-    getIndex(indexKey: string): Set<string> {
-        return new Set(this._indexes.get(indexKey) ?? [])
-    }
-
-    removeFromIndex(indexKey: string, members: Set<string>): void {
-        const idx = this._indexes.get(indexKey)
-        if (!idx) return
-        for (const m of members) idx.delete(m)
-        if (idx.size === 0) this._indexes.delete(indexKey)
-    }
-
-    deleteIndex(indexKey: string): void {
-        this._indexes.delete(indexKey)
-    }
-
-    deleteIndexesByPrefix(prefix: string): void {
-        for (const key of [...this._indexes.keys()]) {
-            if (key.startsWith(prefix)) this._indexes.delete(key)
-        }
-    }
-
     clear(): void {
         this._store.clear()
-        this._indexes.clear()
     }
 }
diff --git a/packages/mizan-ts/src/cache/index.ts b/packages/mizan-ts/src/cache/index.ts
index cbf7635..7e0c4f9 100644
--- a/packages/mizan-ts/src/cache/index.ts
+++ b/packages/mizan-ts/src/cache/index.ts
@@ -2,15 +2,16 @@
  * mizan cache — TypeScript adapter.
  *
  * Same protocol as Python's mizan.cache. Cross-language conformance
- * verified by pin tests.
+ * verified by pin tests. No reverse indexes — scoped purge recomputes
+ * the key directly, broad purge uses prefix scan.
  */
 
 export { MemoryCache } from './backend'
 export type { CacheBackend } from './backend'
-export { deriveCacheKey, buildIndexKeys } from './keys'
+export { deriveCacheKey, CONTEXT_KEY_PREFIX } from './keys'
 
 import type { CacheBackend } from './backend'
-import { deriveCacheKey, buildIndexKeys } from './keys'
+import { deriveCacheKey, CONTEXT_KEY_PREFIX } from './keys'
 
 let _cacheInstance: CacheBackend | null = null
 
@@ -48,53 +49,24 @@ export function cachePut(
     rev: number = 0,
 ): void {
     const key = deriveCacheKey(secret, context, params, userId, rev)
-    const indexes = buildIndexKeys(context, params)
-    backend.put(key, value, indexes)
+    backend.set(key, value)
 }
 
 export function cachePurge(
     backend: CacheBackend,
     context: string,
     params?: Record<string, any> | null,
+    secret?: string | null,
+    userId?: string,
+    rev: number = 0,
 ): number {
-    if (params) {
-        // Scoped purge — AND semantics (intersection)
-        const setsPerParam: Set<string>[] = []
-        const paramIndexKeys: string[] = []
-        for (const [k, v] of Object.entries(params).sort(([a], [b]) => a.localeCompare(b))) {
-            const indexKey = `mizan:idx:${context}:${k}=${String(v)}`
-            paramIndexKeys.push(indexKey)
-            setsPerParam.push(backend.getIndex(indexKey))
-        }
-
-        let keysToDelete: Set<string>
-        if (setsPerParam.length > 0) {
-            keysToDelete = setsPerParam[0]
-            for (let i = 1; i < setsPerParam.length; i++) {
-                keysToDelete = new Set([...keysToDelete].filter(k => setsPerParam[i].has(k)))
-            }
-        } else {
-            keysToDelete = new Set()
-        }
-
-        if (keysToDelete.size > 0) {
-            for (const idxKey of paramIndexKeys) {
-                backend.removeFromIndex(idxKey, keysToDelete)
-            }
-            backend.removeFromIndex(`mizan:idx:${context}`, keysToDelete)
-            return backend.deleteMany([...keysToDelete])
-        }
-        return 0
+    if (params && secret) {
+        // Scoped purge — recompute key and delete directly
+        const key = deriveCacheKey(secret, context, params, userId, rev)
+        return backend.delete(key) ? 1 : 0
     } else {
-        // Broad purge
-        const indexKey = `mizan:idx:${context}`
-        const keysToDelete = backend.getIndex(indexKey)
-        backend.deleteIndex(indexKey)
-        backend.deleteIndexesByPrefix(`mizan:idx:${context}:`)
-
-        if (keysToDelete.size > 0) {
-            return backend.deleteMany([...keysToDelete])
-        }
-        return 0
+        // Broad purge — prefix scan
+        const prefix = `${CONTEXT_KEY_PREFIX}${context}:`
+        return backend.deleteByPrefix(prefix)
     }
 }
diff --git a/packages/mizan-ts/src/cache/keys.ts b/packages/mizan-ts/src/cache/keys.ts
index 1a64dc2..bd223d9 100644
--- a/packages/mizan-ts/src/cache/keys.ts
+++ b/packages/mizan-ts/src/cache/keys.ts
@@ -1,19 +1,16 @@
 /**
  * Cache key derivation — HMAC-SHA256 over JSON-canonical form.
  *
- * Protocol-critical: must produce identical output to Python's derive_cache_key
- * for the same inputs. Cross-language conformance verified by pin tests.
+ * Protocol-critical: must produce identical output to Python's derive_cache_key.
+ * Cross-language conformance verified by pin tests.
  *
- * Wire format:
- *   HMAC-SHA256(secret, stableStringify({"c": ctx, "p": params, "r": rev}))
- *   - All keys sorted recursively
- *   - No whitespace (compact JSON)
- *   - "u" key included only for user-scoped content
- *   - All param values stringified
+ * Key format: "ctx:{context}:{hmac_hex}" — enables broad purge by prefix scan.
  */
 
 import { createHmac } from 'crypto'
 
+const CONTEXT_KEY_PREFIX = 'ctx:'
+
 /**
  * JSON.stringify with recursively sorted keys and no whitespace.
  * Equivalent to Python's json.dumps(obj, sort_keys=True, separators=(",", ":"))
@@ -33,7 +30,7 @@ function stableStringify(obj: any): string {
 /**
  * Derive a deterministic HMAC-SHA256 cache key.
  *
- * Must produce identical output to Python's derive_cache_key for the same inputs.
+ * Returns "ctx:{context}:{hmac_hex}".
  */
 export function deriveCacheKey(
     secret: string,
@@ -42,7 +39,6 @@ export function deriveCacheKey(
     userId?: string,
     rev: number = 0,
 ): string {
-    // Stringify all param values for cross-language determinism
     const sortedParams: Record<string, string> = {}
     for (const [k, v] of Object.entries(params).sort(([a], [b]) => a.localeCompare(b))) {
         sortedParams[k] = String(v)
@@ -54,16 +50,8 @@ export function deriveCacheKey(
     }
 
     const message = stableStringify(keyData)
-    return createHmac('sha256', secret).update(message).digest('hex')
+    const hmacHex = createHmac('sha256', secret).update(message).digest('hex')
+    return `${CONTEXT_KEY_PREFIX}${context}:${hmacHex}`
 }
 
-/**
- * Build reverse index keys for a cache entry.
- */
-export function buildIndexKeys(context: string, params: Record<string, any>): string[] {
-    const keys = [`mizan:idx:${context}`]
-    for (const [k, v] of Object.entries(params).sort(([a], [b]) => a.localeCompare(b))) {
-        keys.push(`mizan:idx:${context}:${k}=${String(v)}`)
-    }
-    return keys
-}
+export { CONTEXT_KEY_PREFIX }
diff --git a/packages/mizan-ts/src/dispatch.ts b/packages/mizan-ts/src/dispatch.ts
index 80239ad..b1c092b 100644
--- a/packages/mizan-ts/src/dispatch.ts
+++ b/packages/mizan-ts/src/dispatch.ts
@@ -212,7 +212,7 @@ export async function handleMutationCall(
                         if (typeof entry === 'string') {
                             cachePurge(cb, entry)
                         } else {
-                            cachePurge(cb, entry.context, entry.params)
+                            cachePurge(cb, entry.context, entry.params, _cacheSecret)
                         }
                     }
                 } catch { /* purge failure is non-fatal */ }
diff --git a/packages/mizan-ts/tests/edge-compat.test.ts b/packages/mizan-ts/tests/edge-compat.test.ts
index e917238..f949c2e 100644
--- a/packages/mizan-ts/tests/edge-compat.test.ts
+++ b/packages/mizan-ts/tests/edge-compat.test.ts
@@ -285,7 +285,8 @@ describe('Cache Conformance', () => {
         const k1 = deriveCacheKey(SECRET, 'user', { user_id: '5' })
         const k2 = deriveCacheKey(SECRET, 'user', { user_id: '5' })
         expect(k1).toBe(k2)
-        expect(k1).toHaveLength(64)
+        expect(k1).toStartWith('ctx:user:')
+        expect(k1).toHaveLength('ctx:user:'.length + 64)
     })
 
     test('deriveCacheKey param order irrelevant', () => {
@@ -298,29 +299,29 @@ describe('Cache Conformance', () => {
         // These exact values are pinned from Python's derive_cache_key output.
         // If this test fails, cross-language cache key compatibility is broken.
         const publicKey = deriveCacheKey(SECRET, 'user', { user_id: '5' }, undefined, 0)
-        expect(publicKey).toBe('605a1ca5ad5994e9b765c8d1b330474c2a0d51a7b8fbbdc402f992da7ba902f6')
+        expect(publicKey).toBe('ctx:user:605a1ca5ad5994e9b765c8d1b330474c2a0d51a7b8fbbdc402f992da7ba902f6')
 
         const userScopedKey = deriveCacheKey(SECRET, 'user', { user_id: '5' }, '5', 0)
-        expect(userScopedKey).toBe('30fc08eb46ee4ff2cf7d317e97dca90fd616511e0587304416f71dc863338dc2')
+        expect(userScopedKey).toBe('ctx:user:30fc08eb46ee4ff2cf7d317e97dca90fd616511e0587304416f71dc863338dc2')
     })
 
-    test('MemoryCache get/put/clear', () => {
+    test('MemoryCache get/set/clear', () => {
         const cache = new MemoryCache()
         expect(cache.get('k1')).toBeNull()
 
-        cache.put('k1', '{"data":true}', ['mizan:idx:ctx'])
+        cache.set('k1', '{"data":true}')
         expect(cache.get('k1')).toBe('{"data":true}')
 
         cache.clear()
         expect(cache.get('k1')).toBeNull()
     })
 
-    test('scoped purge AND semantics', () => {
+    test('scoped purge recomputes key directly', () => {
         const cache = new MemoryCache()
         cachePut(SECRET, cache, 'user', { user_id: '5' }, '{"u5":true}')
         cachePut(SECRET, cache, 'user', { user_id: '6' }, '{"u6":true}')
 
-        const count = cachePurge(cache, 'user', { user_id: '5' })
+        const count = cachePurge(cache, 'user', { user_id: '5' }, SECRET)
         expect(count).toBe(1)
 
         expect(cacheGet(SECRET, cache, 'user', { user_id: '5' })).toBeNull()