diff --git a/changes/298.bugfix.md b/changes/298.bugfix.md
new file mode 100644
index 0000000000..e45736b209
--- /dev/null
+++ b/changes/298.bugfix.md
@@ -0,0 +1,5 @@
+Fixed several ways `zarr.experimental.cache_store.CacheStore` could serve data the source store no longer holds. `set_if_not_exists`, `_set_many`, `delete_dir` and `clear` were not overridden, so they were forwarded straight to the source and the cache went on serving the superseded value — most visibly, creating an array with `overwrite=True` through a `CacheStore` left the *old* array readable through the cache. Each now invalidates the keys it affects. `delete` also failed to reclaim its entry's bytes, so every delete permanently inflated `current_size` and ate into the `max_size` budget, and a value too large to cache was left in the backing cache untracked — uncounted against `max_size`, never eviction-eligible, but still served. Finally, every backing-store mutation is now published in the same locked section as its tracking update, so a concurrent write can no longer be observed half-applied.
+
+`CacheStore.open()` now works: it inherited `WrapperStore.open()`, which builds the wrapped store from a `store_cls` argument and so could not supply the required `cache_store`.
+
+`CacheStore` now requires its `cache_store` to support listing (in addition to deletes), since prefix deletions need it. This is checked in the constructor rather than failing later mid-write.
diff --git a/src/zarr/experimental/cache_store.py b/src/zarr/experimental/cache_store.py
index 20cb4d4c0f..7cbdaee3c0 100644
--- a/src/zarr/experimental/cache_store.py
+++ b/src/zarr/experimental/cache_store.py
@@ -13,6 +13,8 @@
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     from zarr.core.buffer.core import Buffer, BufferPrototype
 
 # A cache entry identifier.  Plain ``str`` for full-key entries that live in
@@ -53,7 +55,8 @@ class CacheStore(WrapperStore[Store]):
         The underlying store to wrap with caching
     cache_store : Store
         The store to use for caching (can be any Store implementation that
-        supports deletes)
+        supports deletes and listing). Listing is required so that ``delete_dir``
+        and ``clear`` can drop the cached keys they invalidate.
     max_age_seconds : int or "infinity", optional
         Maximum age of cached entries in seconds. The string "infinity" means
         entries never expire. Default is "infinity".
@@ -112,6 +115,14 @@ def __init__(
             )
             raise ValueError(msg)
 
+        if not cache_store.supports_listing:
+            msg = (
+                f"The provided cache store {cache_store} does not support listing. "
+                "The cache_store must support listing so that prefix deletions "
+                "(delete_dir, clear) can invalidate the keys they remove."
+            )
+            raise ValueError(msg)
+
         self._cache = cache_store
         # Validate and set max_age_seconds
         if isinstance(max_age_seconds, str):
@@ -124,6 +135,20 @@ def __init__(
         self.cache_set_data = cache_set_data
         self._state = _CacheState()
 
+    @classmethod
+    async def open(cls, *args: Any, **kwargs: Any) -> Self:
+        """Create and open a ``CacheStore``.
+
+        ``WrapperStore.open`` builds the wrapped store from a ``store_cls``
+        argument and then wraps it, which cannot supply the required
+        ``cache_store``; inheriting it makes ``CacheStore.open`` raise. Take
+        ``Store.open``'s behaviour instead: construct from the arguments this
+        class actually accepts, then open.
+        """
+        store = cls(*args, **kwargs)
+        await store._open()
+        return store
+
     def _with_store(self, store: Store) -> Self:
         # Cannot support this operation because it would share a cache, but have a new store
         # So cache keys would conflict
@@ -175,22 +200,11 @@ async def _evict_key(self, entry_key: _CacheEntryKey) -> None:
         For ``(str, ByteRequest)`` keys the entry is removed from the
         in-memory range cache.
         """
-        key_size = self._state.key_sizes.get(entry_key, 0)
-
         if isinstance(entry_key, str):
             await self._cache.delete(entry_key)
+            self._remove_from_tracking(entry_key)
         else:
-            base_key, byte_range = entry_key
-            per_key = self._state.range_cache.get(base_key)
-            if per_key is not None:
-                per_key.pop(byte_range, None)
-                if not per_key:
-                    del self._state.range_cache[base_key]
-
-        self._state.cache_order.pop(entry_key, None)
-        self._state.key_insert_times.pop(entry_key, None)
-        self._state.key_sizes.pop(entry_key, None)
-        self._state.current_size = max(0, self._state.current_size - key_size)
+            self._drop_range(*entry_key)
         self._state.evictions += 1
 
     async def _track_entry(self, entry_key: _CacheEntryKey, value: Buffer) -> bool:
@@ -200,45 +214,60 @@ async def _track_entry(self, entry_key: _CacheEntryKey, value: Buffer) -> bool:
         exceeds ``max_size`` and was skipped.  Callers should roll back any
         data they already stored when this returns ``False``.
 
-        This method holds the lock for the entire operation to ensure atomicity.
+        Must be called while holding ``self._state.lock``, which the callers hold
+        across the backing-store write as well, so the value and its tracking
+        entry are published together.
         """
         value_size = len(value)
 
+        # Drop any prior tracking for this entry first, reclaiming its bytes.
+        # This also removes it from the eviction candidates, so
+        # ``_accommodate_value`` cannot select the very entry being (re)tracked
+        # -- which would double-subtract its size, stop the eviction loop early,
+        # and delete the value the caller just wrote to the backing store.
+        self._remove_from_tracking(entry_key)
+
         # Check if value exceeds max size
         if self.max_size is not None and value_size > self.max_size:
             return False
 
-        async with self._state.lock:
-            # If key already exists, subtract old size first
-            if entry_key in self._state.key_sizes:
-                old_size = self._state.key_sizes[entry_key]
-                self._state.current_size -= old_size
-
-            # Make room for the new value
-            await self._accommodate_value(value_size)
-
-            # Update tracking atomically
-            self._state.cache_order[entry_key] = None
-            self._state.current_size += value_size
-            self._state.key_sizes[entry_key] = value_size
-            self._state.key_insert_times[entry_key] = time.monotonic()
-
+        # Make room for the new value, then track it (appended as most-recent).
+        await self._accommodate_value(value_size)
+        self._state.cache_order[entry_key] = None
+        self._state.current_size += value_size
+        self._state.key_sizes[entry_key] = value_size
+        self._state.key_insert_times[entry_key] = time.monotonic()
         return True
 
     async def _update_access_order(self, entry_key: _CacheEntryKey) -> None:
         """Update the access order for LRU tracking."""
-        if entry_key in self._state.cache_order:
-            async with self._state.lock:
+        async with self._state.lock:
+            # Re-check membership under the lock: the entry may have been evicted
+            # by a concurrent operation between the call and acquiring the lock.
+            if entry_key in self._state.cache_order:
                 self._state.cache_order.move_to_end(entry_key)
 
     def _remove_from_tracking(self, entry_key: _CacheEntryKey) -> None:
-        """Remove an entry from all tracking structures.
+        """Remove an entry from all tracking structures, reclaiming its bytes.
 
         Must be called while holding self._state.lock.
         """
         self._state.cache_order.pop(entry_key, None)
         self._state.key_insert_times.pop(entry_key, None)
-        self._state.key_sizes.pop(entry_key, None)
+        entry_size = self._state.key_sizes.pop(entry_key, 0)
+        self._state.current_size = max(0, self._state.current_size - entry_size)
+
+    def _drop_range(self, key: str, byte_range: ByteRequest) -> None:
+        """Remove one byte-range entry from the range cache and tracking.
+
+        Must be called while holding self._state.lock.
+        """
+        per_key = self._state.range_cache.get(key)
+        if per_key is not None:
+            per_key.pop(byte_range, None)
+            if not per_key:
+                del self._state.range_cache[key]
+        self._remove_from_tracking((key, byte_range))
 
     def _invalidate_range_entries(self, key: str) -> None:
         """Remove all byte-range entries for *key* from the range cache and tracking.
@@ -248,11 +277,7 @@ def _invalidate_range_entries(self, key: str) -> None:
         per_key = self._state.range_cache.pop(key, None)
         if per_key is not None:
             for byte_range in per_key:
-                entry_key: _CacheEntryKey = (key, byte_range)
-                entry_size = self._state.key_sizes.pop(entry_key, 0)
-                self._state.cache_order.pop(entry_key, None)
-                self._state.key_insert_times.pop(entry_key, None)
-                self._state.current_size = max(0, self._state.current_size - entry_size)
+                self._remove_from_tracking((key, byte_range))
 
     # ------------------------------------------------------------------
     # get helpers
@@ -261,36 +286,34 @@ def _invalidate_range_entries(self, key: str) -> None:
     async def _cache_miss(
         self, key: str, byte_range: ByteRequest | None, result: Buffer | None
     ) -> None:
-        """Handle a cache miss by storing or cleaning up after a source-store fetch."""
+        """Handle a cache miss by storing or cleaning up after a source-store fetch.
+
+        Each branch takes ``self._state.lock`` for the whole of its backing-store
+        mutation *and* the matching tracking mutation, so the two can never be
+        observed out of step by a concurrent operation.
+        """
         if result is None:
             if byte_range is None:
-                await self._cache.delete(key)
                 async with self._state.lock:
+                    await self._cache.delete(key)
                     self._remove_from_tracking(key)
             else:
-                entry_key: _CacheEntryKey = (key, byte_range)
                 async with self._state.lock:
-                    per_key = self._state.range_cache.get(key)
-                    if per_key is not None:
-                        per_key.pop(byte_range, None)
-                        if not per_key:
-                            del self._state.range_cache[key]
-                    self._remove_from_tracking(entry_key)
+                    self._drop_range(key, byte_range)
         else:
             if byte_range is None:
-                await self._cache.set(key, result)
-                await self._track_entry(key, result)
+                async with self._state.lock:
+                    await self._cache.set(key, result)
+                    if not await self._track_entry(key, result):
+                        # Value too large for the cache — roll back so the backing
+                        # cache holds no untracked (uncounted, unevictable) orphan.
+                        await self._cache.delete(key)
             else:
-                entry_key = (key, byte_range)
-                self._state.range_cache.setdefault(key, {})[byte_range] = result
-                tracked = await self._track_entry(entry_key, result)
-                if not tracked:
-                    # Value too large for the cache — roll back the insertion
-                    per_key = self._state.range_cache.get(key)
-                    if per_key is not None:
-                        per_key.pop(byte_range, None)
-                        if not per_key:
-                            del self._state.range_cache[key]
+                async with self._state.lock:
+                    self._state.range_cache.setdefault(key, {})[byte_range] = result
+                    if not await self._track_entry((key, byte_range), result):
+                        # Value too large for the cache — roll back the insertion
+                        self._drop_range(key, byte_range)
 
     async def _get_try_cache(
         self, key: str, prototype: BufferPrototype, byte_range: ByteRequest | None = None
@@ -380,17 +403,49 @@ async def set(self, key: str, value: Buffer) -> None:
             The data to store
         """
         await super().set(key, value)
-        # Invalidate all cached byte-range entries (source data changed)
         async with self._state.lock:
+            # The value just written supersedes any cached byte ranges for the key.
             self._invalidate_range_entries(key)
-        if self.cache_set_data:
-            await self._cache.set(key, value)
-            await self._track_entry(key, value)
-        else:
-            await self._cache.delete(key)
-            async with self._state.lock:
+            if self.cache_set_data:
+                await self._cache.set(key, value)
+                if not await self._track_entry(key, value):
+                    # Value too large for the cache — roll back so the backing cache
+                    # holds no untracked (uncounted, unevictable) orphan.
+                    await self._cache.delete(key)
+            else:
+                await self._cache.delete(key)
                 self._remove_from_tracking(key)
 
+    async def set_if_not_exists(self, key: str, value: Buffer) -> None:
+        """
+        Store data only if the key does not already exist in the source store.
+
+        Parameters
+        ----------
+        key : str
+            The key to store under
+        value : Buffer
+            The data to store
+        """
+        await super().set_if_not_exists(key, value)
+        # Whether or not the write happened, any cached state for this key may now
+        # be stale (we may have just written a new value, or it already existed).
+        # Drop all of it so the next read reflects the source. Invalidating
+        # unconditionally is always safe. We do not populate the cache here: there
+        # is no guaranteed-fresh value to store (the write may have been a no-op).
+        async with self._state.lock:
+            self._invalidate_range_entries(key)
+            await self._cache.delete(key)
+            self._remove_from_tracking(key)
+
+    async def _set_many(self, values: Iterable[tuple[str, Buffer]]) -> None:
+        """Bulk writes routed through ``self.set``, so they invalidate the cache.
+
+        ``WrapperStore._set_many`` forwards straight to the source store, so
+        without this override a key written in bulk keeps its stale cached value.
+        """
+        await Store._set_many(self, values)
+
     async def delete(self, key: str) -> None:
         """
         Delete data from both the underlying store and cache.
@@ -401,13 +456,42 @@ async def delete(self, key: str) -> None:
             The key to delete
         """
         await super().delete(key)
-        # Invalidate all cached byte-range entries
         async with self._state.lock:
             self._invalidate_range_entries(key)
-        await self._cache.delete(key)
-        async with self._state.lock:
+            await self._cache.delete(key)
             self._remove_from_tracking(key)
 
+    async def delete_dir(self, prefix: str) -> None:
+        """
+        Delete a prefix from the underlying store and drop its cached keys.
+
+        ``WrapperStore.delete_dir`` delegates straight to the source store, so
+        without this override no ``delete`` runs for the keys under *prefix* and
+        the cache keeps serving them after e.g. an ``overwrite=True`` array
+        creation.
+        """
+        await super().delete_dir(prefix)
+        if prefix != "" and not prefix.endswith("/"):
+            prefix += "/"
+        async with self._state.lock:
+            await self._cache.delete_dir(prefix)
+            for base_key in [k for k in self._state.range_cache if k.startswith(prefix)]:
+                self._invalidate_range_entries(base_key)
+            for entry_key in [
+                k for k in self._state.cache_order if isinstance(k, str) and k.startswith(prefix)
+            ]:
+                self._remove_from_tracking(entry_key)
+
+    async def clear(self) -> None:
+        """Clear the underlying store, and with it everything cached from it.
+
+        Same bypass as ``delete_dir``: ``WrapperStore.clear`` delegates to the
+        source store, which would leave the cache serving values for keys that no
+        longer exist anywhere.
+        """
+        await super().clear()
+        await self.clear_cache()
+
     def cache_info(self) -> dict[str, Any]:
         """Return information about the cache state."""
         return {
@@ -436,12 +520,12 @@ def cache_stats(self) -> dict[str, Any]:
 
     async def clear_cache(self) -> None:
         """Clear all cached data and tracking information."""
-        # Clear the cache store if it supports clear
-        if hasattr(self._cache, "clear"):
-            await self._cache.clear()
-
-        # Reset tracking
+        # The backing-store wipe and the tracking reset share one locked section:
+        # a ``set`` landing between them would otherwise leave a value in the
+        # backing cache with no tracking entry — uncounted against ``max_size``,
+        # never eviction-eligible, and served as a hit indefinitely.
         async with self._state.lock:
+            await self._cache.clear()
             self._state.key_insert_times.clear()
             self._state.cache_order.clear()
             self._state.key_sizes.clear()
diff --git a/tests/test_experimental/test_cache_store.py b/tests/test_experimental/test_cache_store.py
index f688a6ca02..b01e3b25b1 100644
--- a/tests/test_experimental/test_cache_store.py
+++ b/tests/test_experimental/test_cache_store.py
@@ -4,14 +4,17 @@
 
 import asyncio
 import time
+from typing import Any
 
 import pytest
 
 from zarr.abc.store import RangeByteRequest, Store, SuffixByteRequest
-from zarr.core.buffer.core import default_buffer_prototype
+from zarr.core.buffer.core import Buffer, default_buffer_prototype
 from zarr.core.buffer.cpu import Buffer as CPUBuffer
 from zarr.experimental.cache_store import CacheStore
 from zarr.storage import MemoryStore
+from zarr.storage._wrapper import WrapperStore
+from zarr.testing.store import StoreTests
 
 
 class TestCacheStore:
@@ -1038,6 +1041,322 @@ async def test_delete_invalidates_cached_byte_ranges(self) -> None:
         assert result is None
 
 
+class _GatedCacheStore(WrapperStore[Store]):
+    """Cache backend whose first mutating call blocks until it is released.
+
+    Gives the tests a deterministic interleaving point *inside* a backing-store
+    mutation, which is where the tracking state and the backing store can be
+    observed out of step if they are not mutated under one lock.
+    """
+
+    def __init__(self, store: Store) -> None:
+        super().__init__(store)
+        self.entered = asyncio.Event()
+        self.release = asyncio.Event()
+        self.armed = False
+
+    async def _gate(self) -> None:
+        if self.armed:
+            self.armed = False
+            self.entered.set()
+            await self.release.wait()
+
+    async def delete(self, key: str) -> None:
+        await self._gate()
+        await self._store.delete(key)
+
+    async def clear(self) -> None:
+        # Gate *after* the wipe: the orphan window is between the backing store
+        # emptying and the tracking reset, not before.
+        await self._store.clear()
+        await self._gate()
+
+
+class TestCacheStoreWriteCoherence:
+    """Writes through the cache must never leave a stale or orphaned entry.
+
+    Each of these pins a bug that `CacheStore` had: a write path that bypassed
+    the cache entirely, or a backing-store mutation that was not published in
+    the same locked section as its tracking mutation.
+    """
+
+    @staticmethod
+    def _counted_size(cached_store: CacheStore) -> int:
+        """The size the tracking entries actually add up to."""
+        return sum(cached_store._state.key_sizes.values())
+
+    @pytest.fixture
+    def cached_store(self) -> CacheStore:
+        return CacheStore(MemoryStore(), cache_store=MemoryStore())
+
+    @pytest.mark.parametrize(
+        ("operation", "expected"),
+        [
+            ("set", b"NEW"),
+            ("set_if_not_exists", b"NEW"),
+            ("_set_many", b"NEW"),
+            ("delete", None),
+            ("delete_dir", None),
+            ("clear", None),
+        ],
+    )
+    async def test_write_operations_invalidate_the_cache(
+        self, cached_store: CacheStore, operation: str, expected: bytes | None
+    ) -> None:
+        """After any write through the cache, a read must agree with the source.
+
+        `set_if_not_exists`, `_set_many`, `delete_dir` and `clear` were not
+        overridden, so `WrapperStore` forwarded them straight to the source and
+        the cache went on serving the superseded value indefinitely.
+        """
+        proto = default_buffer_prototype()
+        key = "pfx/key"
+        await cached_store.set(key, CPUBuffer.from_bytes(b"OLD"))
+        assert (await cached_store.get(key, proto)).to_bytes() == b"OLD"  # type: ignore[union-attr]
+
+        new = CPUBuffer.from_bytes(b"NEW")
+        if operation == "set":
+            await cached_store.set(key, new)
+        elif operation == "set_if_not_exists":
+            # The key exists at the source, so this is a no-op there; it must
+            # still not leave a cached value the source disagrees with.
+            await cached_store._store.delete(key)
+            await cached_store.set_if_not_exists(key, new)
+        elif operation == "_set_many":
+            await cached_store._set_many([(key, new)])
+        elif operation == "delete":
+            await cached_store.delete(key)
+        elif operation == "delete_dir":
+            await cached_store.delete_dir("pfx")
+        else:
+            await cached_store.clear()
+
+        source = await cached_store._store.get(key, proto)
+        assert (source.to_bytes() if source is not None else None) == expected
+        served = await cached_store.get(key, proto)
+        assert (served.to_bytes() if served is not None else None) == expected
+
+    async def test_overwriting_an_array_does_not_serve_the_old_one(self) -> None:
+        """`overwrite=True` goes through `delete_dir`; the old array must not survive it."""
+        import numpy as np
+
+        import zarr
+
+        cached_store = CacheStore(MemoryStore(), cache_store=MemoryStore())
+        arr = zarr.create_array(cached_store, name="a", shape=(4,), chunks=(4,), dtype="i4")
+        arr[:] = np.array([1, 2, 3, 4], dtype="i4")
+        np.testing.assert_array_equal(zarr.open_array(cached_store, path="a")[:], [1, 2, 3, 4])
+
+        zarr.create_array(
+            cached_store, name="a", shape=(4,), chunks=(4,), dtype="i4", overwrite=True
+        )
+        # Fresh array, so every chunk is the fill value — both through the cache
+        # and at the source.
+        np.testing.assert_array_equal(zarr.open_array(cached_store, path="a")[:], [0, 0, 0, 0])
+        np.testing.assert_array_equal(
+            zarr.open_array(cached_store._store, path="a")[:], [0, 0, 0, 0]
+        )
+
+    async def test_size_accounting_is_exact_across_operations(
+        self, cached_store: CacheStore
+    ) -> None:
+        """`current_size` must always equal the tracked entries' sizes.
+
+        `delete` used to drop the tracking entries without reclaiming their
+        bytes, so every delete permanently inflated `current_size` and ate into
+        the `max_size` budget.
+        """
+        proto = default_buffer_prototype()
+        await cached_store.set("a", CPUBuffer.from_bytes(b"x" * 100))
+        assert cached_store._state.current_size == self._counted_size(cached_store) == 100
+
+        await cached_store.set("a", CPUBuffer.from_bytes(b"x" * 40))  # overwrite
+        assert cached_store._state.current_size == self._counted_size(cached_store) == 40
+
+        await cached_store.set("b", CPUBuffer.from_bytes(b"y" * 10))
+        await cached_store.delete("a")
+        assert cached_store._state.current_size == self._counted_size(cached_store) == 10
+
+        await cached_store.get("missing", proto)  # miss on an absent key
+        await cached_store.clear()
+        assert cached_store._state.current_size == self._counted_size(cached_store) == 0
+
+    @pytest.mark.parametrize("populate_via", ["set", "read"])
+    async def test_oversized_value_leaves_no_orphan(self, populate_via: str) -> None:
+        """A value too large to track must not be left in the backing cache.
+
+        It would be uncounted against `max_size` and never eviction-eligible,
+        yet still served as a hit — a permanent, unbounded leak.
+        """
+        proto = default_buffer_prototype()
+        source = MemoryStore()
+        backing = MemoryStore()
+        cached_store = CacheStore(source, cache_store=backing, max_size=10)
+        value = CPUBuffer.from_bytes(b"z" * 50)
+
+        if populate_via == "set":
+            await cached_store.set("k", value)
+        else:
+            await source.set("k", value)
+            assert (await cached_store.get("k", proto)).to_bytes() == b"z" * 50  # type: ignore[union-attr]
+
+        assert await backing.get("k", proto) is None, "orphan left in the backing cache"
+        assert "k" not in cached_store._state.key_sizes
+        assert cached_store._state.current_size == 0
+        # The value is still readable — it just comes from the source every time.
+        assert (await cached_store.get("k", proto)).to_bytes() == b"z" * 50  # type: ignore[union-attr]
+
+    async def test_delete_racing_a_set_keeps_the_cache_consistent(self) -> None:
+        """A `set` landing inside `delete`'s backing-store call must not be clobbered.
+
+        `delete` used to drop its tracking and then delete from the backing
+        store outside the lock, so a concurrent `set` could be published in that
+        window and have its backing value deleted underneath it — leaving a
+        tracking entry that claimed bytes the backing store did not hold.
+        """
+        proto = default_buffer_prototype()
+        backing = _GatedCacheStore(MemoryStore())
+        cached_store = CacheStore(MemoryStore(), cache_store=backing)
+        await cached_store.set("k", CPUBuffer.from_bytes(b"AAAA"))
+
+        backing.armed = True
+        deleting = asyncio.create_task(cached_store.delete("k"))
+        await backing.entered.wait()  # inside the backing delete, tracking already dropped
+        setting = asyncio.create_task(cached_store.set("k", CPUBuffer.from_bytes(b"BBBBBBBB")))
+        await asyncio.sleep(0)
+        backing.release.set()
+        await asyncio.gather(deleting, setting)
+
+        assert cached_store._state.current_size == self._counted_size(cached_store)
+        tracked = "k" in cached_store._state.key_sizes
+        assert (await backing.get("k", proto) is not None) == tracked, (
+            "tracking and backing store disagree"
+        )
+        source = await cached_store._store.get("k", proto)
+        served = await cached_store.get("k", proto)
+        assert (served.to_bytes() if served else None) == (source.to_bytes() if source else None), (
+            "cache disagrees with the source"
+        )
+
+    async def test_clear_cache_racing_a_set_leaves_no_orphan(self) -> None:
+        """A `set` landing inside `clear_cache`'s backing wipe must not be orphaned.
+
+        `clear_cache` used to wipe the backing store before taking the lock, so
+        a `set` published in that window kept its backing value while its
+        tracking entry was wiped by the reset that followed — an untracked
+        entry served as a hit forever.
+        """
+        backing = _GatedCacheStore(MemoryStore())
+        cached_store = CacheStore(MemoryStore(), cache_store=backing)
+        await cached_store.set("k", CPUBuffer.from_bytes(b"AAAA"))
+
+        backing.armed = True
+        clearing = asyncio.create_task(cached_store.clear_cache())
+        await backing.entered.wait()
+        setting = asyncio.create_task(cached_store.set("k", CPUBuffer.from_bytes(b"BBBBBBBB")))
+        await asyncio.sleep(0)
+        backing.release.set()
+        await asyncio.gather(clearing, setting)
+
+        assert cached_store._state.current_size == self._counted_size(cached_store)
+        for cached_key in [k async for k in backing.list()]:
+            assert cached_key in cached_store._state.key_sizes, (
+                f"untracked value for {cached_key} left in the backing cache"
+            )
+
+    async def test_open_constructs_a_cache_store(self) -> None:
+        """`CacheStore.open` must build a `CacheStore`, not raise.
+
+        The inherited `WrapperStore.open` builds the *wrapped* store from a
+        `store_cls` argument, so it cannot supply `cache_store`.
+        """
+        cached_store = await CacheStore.open(MemoryStore(), cache_store=MemoryStore())
+        assert isinstance(cached_store, CacheStore)
+        await cached_store.set("k", CPUBuffer.from_bytes(b"v"))
+        assert (await cached_store.get("k", default_buffer_prototype())).to_bytes() == b"v"  # type: ignore[union-attr]
+
+
+class TestCacheStoreConformance(StoreTests[CacheStore, CPUBuffer]):
+    """Run `CacheStore` through the shared `Store` conformance suite.
+
+    `CacheStore` is a `Store` that users hand to `zarr.open`, so it owes the
+    same contract as every other store. The write paths this module fixes were
+    exactly the ones the shared suite exercises and nothing else did.
+    """
+
+    store_cls = CacheStore
+    buffer_cls = CPUBuffer
+
+    @pytest.fixture
+    def store_kwargs(self) -> dict[str, Any]:
+        return {"store": MemoryStore(), "cache_store": MemoryStore()}
+
+    async def get(self, store: CacheStore, key: str) -> Buffer:
+        # Read the source store, not the cache: the harness asks "what did the
+        # store actually persist?", which is a question about the source.
+        return await store._store.get(key, prototype=default_buffer_prototype())  # type: ignore[return-value]
+
+    async def set(self, store: CacheStore, key: str, value: Buffer) -> None:
+        await store._store.set(key, value)
+
+    def test_store_repr(self, store: CacheStore) -> None:
+        assert "CacheStore" in repr(store)
+
+    def test_store_supports_writes(self, store: CacheStore) -> None:
+        assert store.supports_writes
+
+    def test_store_supports_listing(self, store: CacheStore) -> None:
+        assert store.supports_listing
+
+    @pytest.mark.skip(
+        reason="CacheStore deliberately opts out of sync IO (_supports_sync_io is False), "
+        "but the inherited WrapperStore sync methods still satisfy the protocol "
+        "structurally, so the harness does not skip this itself."
+    )
+    async def test_delete_sync_visible_to_async_get(self, store: CacheStore) -> None: ...
+
+    # The four tests below are skipped because of pre-existing gaps in
+    # CacheStore's construction API, not because of anything this module
+    # changes. They are left visible rather than dropped so the gaps are
+    # recorded somewhere other than a review comment.
+
+    @pytest.mark.skip(
+        reason="CacheStore.__init__ takes no read_only kwarg; it derives read_only from the "
+        "wrapped source store. Use CacheStore.with_read_only instead (covered by "
+        "TestCacheStore::test_with_read_only_round_trip)."
+    )
+    async def test_store_open_read_only(
+        self, open_kwargs: dict[str, Any], read_only: bool
+    ) -> None: ...
+
+    @pytest.mark.skip(reason="See test_store_open_read_only: no read_only constructor kwarg.")
+    async def test_read_only_store_raises(self, open_kwargs: dict[str, Any]) -> None: ...
+
+    @pytest.mark.skip(reason="See test_store_open_read_only: no read_only constructor kwarg.")
+    async def test_with_read_only_store(self, open_kwargs: dict[str, Any]) -> None: ...
+
+    @pytest.mark.skip(
+        reason="CacheStore._with_store raises by design: a copy wrapping a different source "
+        "store would share this store's cache, so the cached keys would collide. That makes "
+        "the __enter__/__exit__ protocol unsupported."
+    )
+    def test_store_context_manager(self, open_kwargs: dict[str, Any]) -> None: ...
+
+
+async def test_cache_store_requires_listing_support() -> None:
+    """A cache store that cannot list cannot back `delete_dir`/`clear`.
+
+    Rejecting it up front beats failing later, mid-write, with a
+    `NotImplementedError` from the backing store.
+    """
+
+    no_listing = MemoryStore()
+    no_listing.supports_listing = False
+
+    with pytest.raises(ValueError, match="does not support listing"):
+        CacheStore(MemoryStore(), cache_store=no_listing)
+
+
 def test_cache_store_opts_out_of_sync_io() -> None:
     """`CacheStore` must not advertise sync IO capability.