diff --git a/mongodol/base.py b/mongodol/base.py
index bd884fa..6b99f68 100644
--- a/mongodol/base.py
+++ b/mongodol/base.py
@@ -8,7 +8,7 @@
 
 from pymongo import MongoClient
 
-from dol import KvReader, Collection as DolCollection, BaseValuesView, BaseItemsView
+from dol import KvReader, Collection as DolCollection
 
 from mongodol.constants import ID, PyMongoCollectionSpec, end_of_cursor, DFLT_TEST_DB
 from mongodol.util import (
@@ -17,6 +17,12 @@
     projection_union,
     get_mongo_collection_pymongo_obj,
 )
+from mongodol.views import (
+    MongoItemsView,
+    MongoValuesView,
+    bulk_items,
+    bulk_values,
+)
 
 
 # TODO: mgc type annotation
@@ -123,7 +129,8 @@ class MongoCollectionReader(MongoCollectionCollection, KvReader):
     >>> assert v != {'the': 'default'}
 
     ``s.keys()``, ``s.values()``, and ``s.items()`` are ``collections.abc.MappingViews`` instances
-    (specialized for mongo).
+    (specialized for mongo -- see :mod:`mongodol.views`: they fetch the whole collection in
+    a single query, and keep doing so, correctly, when the store is wrapped by ``dol``).
 
     >>> assert type(s.keys()) == s.KeysView
     >>> assert type(s.values()) == s.ValuesView
@@ -160,19 +167,10 @@ class MongoCollectionReader(MongoCollectionCollection, KvReader):
 
     _projections_are_flattened = False
 
-    class ValuesView(BaseValuesView):
-        def __contains__(self, v):
-            return self._mapping.contains_value(v)
-
-        def __iter__(self):
-            return self._mapping.iter_values()
-
-    class ItemsView(BaseItemsView):
-        def __contains__(self, item):
-            return self._mapping.contains_item(item)
-
-        def __iter__(self):
-            return self._mapping.iter_items()
+    #: Views that resolve the bulk-read fast path through any ``dol`` wrapper chain,
+    #: rather than through blind attribute delegation. See :mod:`mongodol.views`.
+    ValuesView = MongoValuesView
+    ItemsView = MongoItemsView
 
     def __init__(
         self,
@@ -538,20 +536,30 @@ def __getitem__(self, k):
 
 
 class MongoBaseStore(Store):
+    """A ``Store`` that forwards the mongo bulk-read protocol through its transforms.
+
+    Historically this was the *only* way to get ``values()``/``items()`` to honour a
+    wrapper's transforms -- hence ``mongodol.trans.wrap_kvs``, which uses it as the
+    wrapper class. It is no longer needed for that: :mod:`mongodol.views` resolves the
+    bulk path through any wrapper chain, so plain ``dol.wrap_kvs`` now works too. It is
+    kept because it also forwards the write-side bulk methods (``append``/``extend``),
+    and because code may call ``iter_values()``/``contains_value()`` directly.
+    """
+
     def contains_value(self, v):
         return self.store.contains_value(self._data_of_obj(v))
 
     def iter_values(self):
-        return map(self._obj_of_data, self.store.iter_values())
+        return map(self._obj_of_data, bulk_values(self.store))
 
     def contains_item(self, item):
         k, v = item
         return self.store.contains_item((self._id_of_key(k), self._data_of_obj(v)))
 
     def iter_items(self):
-        yield from (
+        return (
             (self._key_of_id(key), self._obj_of_data(doc))
-            for key, doc in self.store.iter_items()
+            for key, doc in bulk_items(self.store)
         )
 
     def append(self, v):
diff --git a/mongodol/stores.py b/mongodol/stores.py
index 33eef2d..dd8c594 100644
--- a/mongodol/stores.py
+++ b/mongodol/stores.py
@@ -14,6 +14,7 @@
     MongoCollectionPersister,
 )
 from mongodol.trans import PostGet, ObjOfData, normalize_result
+from mongodol.views import disable_bulk_read
 
 single_value_fetch_with_unicity_validation = partial(
     wrap_kvs, postget=PostGet.single_value_fetch_with_unicity_validation
@@ -66,8 +67,12 @@ class MongoCollectionFirstDocReader(MongoCollectionReader):
     """
 
 
+# ``disable_bulk_read``: ``s[key]`` collects *all* docs matching the key into a list,
+# whereas the inherited one-query bulk stream yields single docs -- so ``values()`` and
+# ``items()`` must take the per-key path here to stay equal to ``s[key]``.
+@disable_bulk_read
 @wrap_kvs(
-    postget=partial(ObjOfData.all_docs_fetch, doc_collector=list)
+    postget=partial(PostGet.all_docs_fetch, doc_collector=list)
 )  # list is default but explicit here to show that other choices possible
 class MongoCollectionMultipleDocsReader(MongoCollectionReader):
     """A mongo collection (kv-)reader where s[key] will return the list of all key-matching docs.
@@ -96,8 +101,10 @@ class MongoCollectionFirstDocPersister(MongoCollectionPersisterWithResultMapping
     """
 
 
+# See the ``disable_bulk_read`` note on MongoCollectionMultipleDocsReader.
+@disable_bulk_read
 @wrap_kvs(
-    postget=partial(ObjOfData.all_docs_fetch, doc_collector=list)
+    postget=partial(PostGet.all_docs_fetch, doc_collector=list)
 )  # list is default but explicit here to show that other choices possible
 class MongoCollectionMultipleDocsPersister(MongoCollectionPersisterWithResultMapping):
     """A mongo collection (kv-)reader where s[key] will return the list of all key-matching docs.
@@ -113,9 +120,11 @@ def __setitem__(self, k, v):
         ), (
             f"v (value) must be mappings (often dictionaries) or a collection of mappings. Were:\n\tk={k}\n\tv={v}"
         )
-        self._mgc.delete_many(self._merge_with_filt(k))
-        _v = v if isinstance(v, Collection) else [v]
-        return self._mgc.insert_many([self._build_doc(k, vi) for vi in _v])
+        self.mgc.delete_many(self._merge_with_filt(k))
+        # A Mapping is itself a Collection, so it must be tested for first, or a single
+        # doc would be "iterated" into its field names.
+        docs = [v] if isinstance(v, Mapping) else list(v)
+        return self.mgc.insert_many([self._build_doc(k, doc) for doc in docs])
 
 
 class MongoStore(Store):
diff --git a/mongodol/tests/not_working.py b/mongodol/tests/not_working.py
deleted file mode 100644
index a7c61b6..0000000
--- a/mongodol/tests/not_working.py
+++ /dev/null
@@ -1,26 +0,0 @@
-import pytest
-from mongodol.tests.util import populated_pymongo_collection
-from mongodol.base import MongoCollectionReader
-
-
-@pytest.mark.xfail(reason='TDD')
-def test_mongo_values_view_when_wrapping():
-    s = MongoCollectionReader(mgc=populated_pymongo_collection())
-
-    assert type(s.values()).__name__ == 'MongoValuesView'
-
-    from dol import wrap_kvs
-
-    ss = wrap_kvs(s)
-    assert type(ss.values()).__name__ == 'MongoValuesView'
-
-    from dol import wrap_kvs
-
-    ss = wrap_kvs(s)
-    ss.values()
-
-    # TODO: But we want this to NOT raise an error. Need to use factories
-    with pytest.raises(AttributeError) as excinfo:
-        list(ss.values())  # BOOM
-
-    assert "'ValuesView' object has no attribute 'mgc'" in str(excinfo.value)
diff --git a/mongodol/tests/views_test.py b/mongodol/tests/views_test.py
new file mode 100644
index 0000000..8482e42
--- /dev/null
+++ b/mongodol/tests/views_test.py
@@ -0,0 +1,236 @@
+"""Tests for the ``Mapping``-contract invariants of mongo store views.
+
+The invariants pinned here are the ones every ``Mapping`` owes its user::
+
+    list(store.values()) == [store[k] for k in store]
+    list(store.items())  == [(k, store[k]) for k in store]
+
+They are easy to break in mongodol because its views take a *bulk-read* fast path
+(one ``find`` for the whole collection instead of one per key) and, before
+:mod:`mongodol.views`, that fast path punched through any ``dol`` wrapper -- see
+`i2mint/mongodol#7 <https://github.com/i2mint/mongodol/issues/7>`_.
+"""
+
+from operator import itemgetter
+
+import pytest
+from dol import filt_iter, wrap_kvs
+
+from mongodol.base import MongoBaseStore, MongoCollectionReader
+from mongodol.stores import (
+    MongoCollectionFirstDocPersister,
+    MongoCollectionMultipleDocsPersister,
+)
+from mongodol.tests.util import get_test_collection_object
+from mongodol.views import (
+    NoBulkReadPath,
+    bulk_values,
+    is_crossable,
+    provides_bulk_read,
+    resolve_bulk_source,
+)
+
+#: Collection used by this module. Its own, so parallel modules can't disturb it.
+TEST_COLLECTION_NAME = "views_test"
+
+#: Docs written to the test collection before each test.
+TEST_DOCS = (
+    {"_id": "123", "name": "Matthew", "age": 42},
+    {"_id": "456", "name": "Mark", "age": 43},
+)
+
+
+def assert_mapping_view_invariants(store):
+    """Assert that ``store``'s views agree with its ``__iter__``/``__getitem__``."""
+    keys = list(store)
+    assert list(store.keys()) == keys
+    assert list(store.values()) == [store[k] for k in keys]
+    assert list(store.items()) == list(zip(keys, (store[k] for k in keys)))
+
+
+@pytest.fixture
+def store():
+    """A populated ``MongoCollectionFirstDocPersister`` over a dedicated collection."""
+    mgc = get_test_collection_object(collection_name=TEST_COLLECTION_NAME)
+    s = MongoCollectionFirstDocPersister(mgc)
+    for k in list(s):
+        del s[k]
+    for doc in TEST_DOCS:
+        s[{"_id": doc["_id"]}] = {k: v for k, v in doc.items() if k != "_id"}
+    return s
+
+
+# --------------------------------------------------------------------------------------
+# The issue #7 reproduction
+
+
+def test_wrap_kvs_value_trans_reaches_values_and_items(store):
+    """``dol.wrap_kvs`` value transforms must show up in ``values()``/``items()``.
+
+    This is the i2mint/mongodol#7 reproduction: before the fix, ``list(ss.values())``
+    returned the raw, untransformed mongo documents.
+    """
+    ss = wrap_kvs(store, obj_of_data=itemgetter("name", "age"))
+
+    assert [ss[k] for k in ss] == [("Matthew", 42), ("Mark", 43)]
+    assert list(ss.values()) == [("Matthew", 42), ("Mark", 43)]
+    assert [v for _, v in ss.items()] == [("Matthew", 42), ("Mark", 43)]
+    assert_mapping_view_invariants(ss)
+
+
+def test_wrap_kvs_key_trans_reaches_items(store):
+    """``dol.wrap_kvs`` key transforms must show up in ``keys()``/``items()``.
+
+    Only the *keys* are checked against ``__iter__`` here: with no
+    ``getitem_projection``, ``items()`` values still differ from ``store[k]`` by the
+    key fields -- see ``test_items_values_equal_getitem_values_when_no_getitem_projection``.
+    """
+    ss = wrap_kvs(store, key_of_id=itemgetter("_id"), id_of_key=lambda k: {"_id": k})
+
+    assert list(ss) == list(ss.keys()) == ["123", "456"]
+    assert [k for k, _ in ss.items()] == ["123", "456"]
+    assert list(ss.values()) == [ss[k] for k in ss]
+
+
+def test_stacked_wrappers(store):
+    """Every layer of a wrapper stack must contribute to the bulk stream."""
+    ss = wrap_kvs(store, obj_of_data=itemgetter("name", "age"))
+    sss = wrap_kvs(ss, obj_of_data=lambda name_and_age: name_and_age[0].upper())
+
+    assert list(sss.values()) == ["MATTHEW", "MARK"]
+    assert_mapping_view_invariants(sss)
+
+
+def test_mongodol_wrap_kvs_still_honours_transforms(store):
+    """The historical ``MongoBaseStore``-based wrapper keeps working."""
+    ss = wrap_kvs(store, wrapper=MongoBaseStore, obj_of_data=itemgetter("name", "age"))
+
+    assert list(ss.values()) == [("Matthew", 42), ("Mark", 43)]
+    assert_mapping_view_invariants(ss)
+
+
+# --------------------------------------------------------------------------------------
+# Fallback: layers whose transforms can't be pushed onto a bulk stream
+
+
+def test_filtered_store_falls_back_to_the_per_key_path(store):
+    """``filt_iter`` changes the key set, so the bulk stream must not be used."""
+    ss = filt_iter(store, filt=lambda k: k["_id"] == "123")
+
+    with pytest.raises(NoBulkReadPath):
+        bulk_values(ss)
+    assert list(ss.values()) == [{"_id": "123", "name": "Matthew", "age": 42}]
+    assert_mapping_view_invariants(ss)
+
+
+def test_postget_wrapper_falls_back_to_the_per_key_path(store):
+    """A user-supplied ``postget`` isn't expressible on the bulk stream."""
+    ss = wrap_kvs(store, postget=lambda k, v: (k["_id"], v["name"]))
+
+    with pytest.raises(NoBulkReadPath):
+        bulk_values(ss)
+    assert list(ss.values()) == [("123", "Matthew"), ("456", "Mark")]
+    assert_mapping_view_invariants(ss)
+
+
+def test_multiple_docs_store_uses_the_per_key_path():
+    """``MongoCollectionMultipleDocsPersister`` values are *lists* of docs.
+
+    Its inherited bulk stream yields single docs, so it declares itself
+    bulk-unfaithful (``disable_bulk_read``) and the views take the per-key path.
+    """
+    mgc = get_test_collection_object(collection_name=TEST_COLLECTION_NAME)
+    s = MongoCollectionMultipleDocsPersister(mgc)
+    for k in list(s):
+        del s[k]
+    s[{"_id": "123"}] = {"name": "Matthew", "age": 42}
+    s[{"_id": "456"}] = {"name": "Mark", "age": 43}
+
+    assert s[{"_id": "123"}] == [{"_id": "123", "name": "Matthew", "age": 42}]
+    with pytest.raises(NoBulkReadPath):
+        bulk_values(s)
+    assert_mapping_view_invariants(s)
+
+
+# --------------------------------------------------------------------------------------
+# Containment (``v in store.values()``, ``item in store.items()``)
+
+
+def test_containment_through_a_non_invertible_value_trans(store):
+    """A value transform with no declared inverse must not be pushed down to mongo.
+
+    Before the fix this raised a pymongo ``OperationFailure`` (a tuple was handed
+    to ``find`` as a filter).
+    """
+    ss = wrap_kvs(store, obj_of_data=itemgetter("name", "age"))
+
+    assert ("Matthew", 42) in ss.values()
+    assert ("Nobody", 0) not in ss.values()
+    assert ({"_id": "123"}, ("Matthew", 42)) in ss.items()
+    assert ({"_id": "123"}, ("Nobody", 0)) not in ss.items()
+
+
+def test_containment_uses_the_bulk_path_when_the_trans_is_invertible(store):
+    """With both directions declared, containment stays a single mongo query."""
+    ss = wrap_kvs(
+        store,
+        obj_of_data=lambda d: dict(d, name=d["name"].upper()),
+        data_of_obj=lambda d: dict(d, name=d["name"].capitalize()),
+    )
+
+    assert {"_id": "123", "name": "MATTHEW", "age": 42} in ss.values()
+    assert {"_id": "123", "name": "NOBODY", "age": 42} not in ss.values()
+
+
+# --------------------------------------------------------------------------------------
+# The resolver's own contract
+
+
+def test_unwrapped_reader_is_its_own_bulk_source():
+    """No wrapper: the reader itself provides the fast path, nothing to cross."""
+    mgc = get_test_collection_object(collection_name=TEST_COLLECTION_NAME)
+    s = MongoCollectionReader(mgc)
+
+    source, layers = resolve_bulk_source(s, "iter_values")
+    assert source is s
+    assert layers == []
+    assert provides_bulk_read(s, "iter_values")
+    assert not is_crossable(s)
+
+
+def test_transform_only_wrapper_is_crossed_not_used_as_a_source(store):
+    """A plain ``wrap_kvs`` layer is crossed on the way to the backend fast path."""
+    ss = wrap_kvs(store, obj_of_data=itemgetter("name"))
+
+    assert is_crossable(ss)
+    assert not provides_bulk_read(ss, "iter_values")
+    source, layers = resolve_bulk_source(ss, "iter_values")
+    assert source is store
+    assert layers[0] is ss  # dol may insert further pass-through layers behind it
+    assert all(map(is_crossable, layers))
+
+
+def test_bulk_source_lookup_ignores_store_attribute_delegation(store):
+    """``hasattr`` lies on a ``Store``; ``provides_bulk_read`` must not.
+
+    ``Store.__getattr__`` forwards to the wrapped store, so a wrapper *looks* like
+    it implements the bulk protocol. This is precisely what made issue #7 silent.
+    """
+    ss = wrap_kvs(store, obj_of_data=itemgetter("name"))
+
+    assert hasattr(ss, "iter_values")  # ...only because of delegation
+    assert not provides_bulk_read(ss, "iter_values")
+
+
+@pytest.mark.xfail(
+    reason=(
+        "Separate, pre-existing bug: MongoCollectionReader.iter_items pops the key "
+        "fields out of the value even when getitem_projection is None, so items() "
+        "values lack '_id' while store[k] has it. Fixing it changes behaviour that "
+        "tests/int_tests/base_int_test.py explicitly encodes."
+    ),
+    strict=True,
+)
+def test_items_values_equal_getitem_values_when_no_getitem_projection(store):
+    """``items()`` values must equal ``store[k]``, key fields included."""
+    assert list(store.items()) == [(k, store[k]) for k in store]
diff --git a/mongodol/trans.py b/mongodol/trans.py
index 68e91ee..c257973 100644
--- a/mongodol/trans.py
+++ b/mongodol/trans.py
@@ -234,6 +234,16 @@ def single_value_fetch_without_unicity_validation(store, k, cursor):
         else:
             raise KeyError(f"No document found for query: {k}")
 
+    @staticmethod
+    def all_docs_fetch(k, cursor, doc_collector=list):
+        """Collect every doc matching ``k``, so ``s[k]`` is a collection of docs.
+
+        The key-aware (``postget``) counterpart of :meth:`ObjOfData.all_docs_fetch`.
+        ``wrap_kvs`` calls ``obj_of_data`` with the value alone and ``postget`` with
+        ``(key, value)``, so a store wired through ``postget`` needs this signature.
+        """
+        return doc_collector(cursor)
+
 
 class ObjOfData:
     @staticmethod
diff --git a/mongodol/views.py b/mongodol/views.py
new file mode 100644
index 0000000..ed223a5
--- /dev/null
+++ b/mongodol/views.py
@@ -0,0 +1,352 @@
+"""Mapping views that keep working when a mongo store is wrapped by ``dol``.
+
+A mongo collection can serve a store's whole ``(key, value)`` stream in a single
+``find`` round trip, so :class:`~mongodol.base.MongoCollectionReader` implements a
+**bulk-read protocol** -- ``iter_values``, ``iter_items``, ``contains_value`` and
+``contains_item`` -- and exposes it through the ``values()``/``items()`` views
+defined here. One query instead of N is the whole point of these views.
+
+The catch is *composition*. A ``dol`` :class:`~dol.base.Store` wrapper (what
+``wrap_kvs`` builds) forwards every attribute it doesn't define to the store it
+wraps. A view that simply calls ``self._mapping.iter_values()`` therefore punches
+straight through the wrappers and yields raw backend documents, silently skipping
+the value transforms the user asked for -- breaking the ``Mapping`` contract::
+
+    list(store.values()) == [store[k] for k in store]
+
+(see `i2mint/mongodol#7 <https://github.com/i2mint/mongodol/issues/7>`_).
+
+This module resolves the bulk stream **explicitly** instead of relying on
+attribute delegation. Given the store a view was built on, :func:`bulk_values`
+and :func:`bulk_items` walk the wrapper chain inward, remembering each layer they
+cross, until they reach a store that actually implements the bulk-read protocol.
+The bulk stream is then re-transformed by the crossed layers, innermost first, so
+that it lands in exactly the same space as ``store[k]``.
+
+A layer may only be crossed if its read path is *plain transform composition* --
+"read from the inner store, then apply ``_key_of_id``/``_obj_of_data``", which is
+what :class:`~dol.base.Store` does. A layer that redefines ``__getitem__`` or
+``__iter__`` (``wrap_kvs(postget=...)``, ``filt_iter``, ``cached_keys``, ...)
+changes values or key sets in ways that cannot be pushed onto a bulk stream, so
+the resolver refuses to guess: it raises :class:`NoBulkReadPath` and the views
+fall back to the generic per-key behaviour. That fallback is correct, just one
+round trip per key -- correctness first, efficiency when it is provable.
+
+Simple use is invisible: build a mongo store, wrap it however you like, and
+``values()``/``items()`` agree with ``__getitem__``. The knobs, for store authors:
+
+- Implement the bulk-read methods to *provide* the fast path.
+- Set the :data:`BULK_READ_IS_FAITHFUL_ATTR` class attribute to ``False`` (see
+  :func:`disable_bulk_read`) when a class inherits bulk-read methods that no
+  longer agree with its own ``__getitem__``.
+
+Known limitation. :class:`~mongodol.base.MongoCollectionReader` is deliberately a
+*cursor*-level store: ``s[k]`` is a pymongo ``Cursor``, while its bulk stream
+already yields *documents* -- one per key. The two only line up once a single-doc
+layer (``MongoCollectionFirstDocReader`` and friends) has turned cursors into
+docs, which is why those are the stores you are meant to wrap. Hanging an
+``obj_of_data`` that expects a cursor directly off the raw reader is outside the
+protocol: such a transform cannot be pushed onto a doc-level bulk stream, and is
+not detectable from here.
+
+Nothing here is mongo-specific; it is a general answer to "how does a store with
+a bulk-read fast path compose with ``dol`` wrappers?", and would be a reasonable
+thing for ``dol`` itself to own one day.
+"""
+
+from typing import Any, Callable, Iterable, Iterator, Tuple
+
+from dol import BaseItemsView, BaseValuesView
+from dol.base import Store
+
+#: Bulk-read method yielding a store's values in one backend round trip.
+ITER_VALUES_METHOD = "iter_values"
+#: Bulk-read method yielding a store's ``(key, value)`` pairs in one backend round trip.
+ITER_ITEMS_METHOD = "iter_items"
+#: Bulk-read method answering "is this value in the store?" in one backend round trip.
+CONTAINS_VALUE_METHOD = "contains_value"
+#: Bulk-read method answering "is this item in the store?" in one backend round trip.
+CONTAINS_ITEM_METHOD = "contains_item"
+
+#: Class attribute through which a store declares whether its bulk-read methods are
+#: value-equivalent to its own ``__getitem__``. It defaults to ``True`` (a class that
+#: implements the protocol is trusted to implement it faithfully). It exists because
+#: ``dol``'s class-decorator wrapping *copies* the wrapped class's extra methods onto
+#: the wrapper, so a wrapper that redefines value semantics -- ``wrap_kvs(postget=...)``
+#: -- silently inherits bulk-read methods that no longer match it. Such a class sets
+#: this to ``False``; see :func:`disable_bulk_read`.
+BULK_READ_IS_FAITHFUL_ATTR = "_bulk_read_is_faithful"
+
+#: The ``dol`` :class:`~dol.base.Store` attribute holding the store a wrapper wraps.
+INNER_STORE_ATTR = "store"
+
+KeyValStream = Iterator[Tuple[Any, Any]]
+
+
+class NoBulkReadPath(Exception):
+    """No bulk-read stream can be *proven* equivalent to the store's per-key reads.
+
+    Raised by the resolvers of this module, and caught by the views, which then
+    fall back to the generic (correct, one-round-trip-per-key) ``Mapping``
+    behaviour. It is a control-flow signal, not a user-facing error.
+    """
+
+
+# -------------------------------------------------------------------------------------
+# Inspecting a store layer
+#
+# Everything here looks attributes up on ``type(store)``, never on the instance:
+# ``dol``'s ``Store.__getattr__`` forwards *instance* lookups to the wrapped store, so
+# ``hasattr(store, 'iter_values')`` is True even for a wrapper that has no idea what a
+# bulk read is. Class lookup does not delegate, so it tells the truth.
+
+
+def _class_attr(store, attr: str, default=None):
+    """Look ``attr`` up on ``type(store)``, bypassing ``Store.__getattr__`` delegation."""
+    return getattr(type(store), attr, default)
+
+
+def provides_bulk_read(store, method_name: str) -> bool:
+    """Whether ``store``'s own class implements bulk-read ``method_name``, faithfully.
+
+    "Faithfully" means the store has not declared, via
+    :data:`BULK_READ_IS_FAITHFUL_ATTR`, that its bulk-read methods disagree with
+    its ``__getitem__``.
+    """
+    if not _class_attr(store, BULK_READ_IS_FAITHFUL_ATTR, True):
+        return False
+    return _class_attr(store, method_name) is not None
+
+
+def is_crossable(store) -> bool:
+    """Whether ``store`` is a ``Store`` layer whose read path is plain transform composition.
+
+    Such a layer reads from the store it wraps and applies ``_key_of_id`` to keys
+    and ``_obj_of_data`` to values -- and nothing else. Those two transforms can be
+    mapped over a bulk stream, so the layer can be "crossed" on the way to the
+    backend's fast path. A layer that redefines ``__getitem__`` (``postget``) or
+    ``__iter__`` (key filtering/caching) cannot.
+    """
+    if not isinstance(store, Store):
+        return False
+    cls = type(store)
+    return cls.__getitem__ is Store.__getitem__ and cls.__iter__ is Store.__iter__
+
+
+def store_layers(store) -> Iterator:
+    """Yield ``store`` then each store it wraps, outermost first, innermost last.
+
+    The chain ends at the first non-``Store`` -- the actual backend. Note that
+    ``dol`` is free to insert pass-through ``Store`` layers of its own, so never
+    assume one ``wrap_kvs`` call means exactly one layer.
+
+    >>> from dol import wrap_kvs
+    >>> layers = list(store_layers(wrap_kvs({'a': 1}, obj_of_data=str)))
+    >>> type(layers[0]).__name__, type(layers[-1]).__name__
+    ('Store', 'dict')
+    >>> all(isinstance(x, Store) for x in layers[:-1])
+    True
+    """
+    yield store
+    while isinstance(store, Store):
+        inner = getattr(store, INNER_STORE_ATTR, None)
+        if inner is None:  # a Store that never got one: nothing further to walk
+            return
+        store = inner
+        yield store
+
+
+def resolve_bulk_source(store, method_name: str):
+    """Find the store providing bulk-read ``method_name``, and the layers crossed to reach it.
+
+    :return: ``(source, layers)`` where ``layers`` are the crossed
+        :class:`~dol.base.Store` wrappers, outermost first.
+    :raises NoBulkReadPath: if a layer that cannot be crossed is met before a
+        provider is found.
+    """
+    layers = []
+    for layer in store_layers(store):
+        if provides_bulk_read(layer, method_name):
+            return layer, layers
+        if not is_crossable(layer):
+            raise NoBulkReadPath(
+                f"No bulk {method_name!r} path: {type(layer).__name__} neither provides "
+                "it nor is a plain transform-composition layer that can be crossed."
+            )
+        layers.append(layer)
+    raise NoBulkReadPath(f"No store in the chain provides {method_name!r}")
+
+
+# -------------------------------------------------------------------------------------
+# Pushing a layer's transforms onto a bulk stream
+
+
+def _outgoing_trans(layer, trans_attr: str) -> Callable[[Any], Any]:
+    """The layer's outgoing (backend -> user) transform named ``trans_attr``."""
+    return getattr(layer, trans_attr)
+
+
+def _ingoing_trans(
+    layer, *, outgoing_attr: str, ingoing_attr: str
+) -> Callable[[Any], Any]:
+    """The layer's ingoing (user -> backend) transform, if it is usable as an inverse.
+
+    A layer that transforms outgoing values/keys but declares no ingoing transform
+    has no inverse, so a user-space value cannot be pushed down to the backend.
+    """
+    cls = type(layer)
+    transforms_outgoing = getattr(cls, outgoing_attr) is not getattr(
+        Store, outgoing_attr
+    )
+    has_inverse = getattr(cls, ingoing_attr) is not getattr(Store, ingoing_attr)
+    if transforms_outgoing and not has_inverse:
+        raise NoBulkReadPath(
+            f"{type(layer).__name__} transforms outgoing values via {outgoing_attr!r} "
+            f"but declares no {ingoing_attr!r} inverse, so nothing can be pushed down "
+            "to the backend."
+        )
+    return getattr(layer, ingoing_attr)
+
+
+def _map_outward(stream: Iterable, layers, trans_attr: str) -> Iterator:
+    """Apply each crossed layer's ``trans_attr`` to ``stream``, innermost layer first."""
+    for layer in reversed(layers):
+        stream = map(_outgoing_trans(layer, trans_attr), stream)
+    return iter(stream)
+
+
+def _trans_items(items: KeyValStream, key_of_id, obj_of_data) -> KeyValStream:
+    """Apply one layer's key and value transforms to an item stream.
+
+    A function -- not an inlined generator expression in :func:`_map_items_outward` --
+    so that each layer's transforms are captured in their own scope. A genexpr would
+    look them up lazily, in a scope the next loop iteration has already overwritten.
+    """
+    return ((key_of_id(k), obj_of_data(v)) for k, v in items)
+
+
+def _map_items_outward(items: KeyValStream, layers) -> KeyValStream:
+    """Apply each crossed layer's key *and* value transforms to an item stream."""
+    for layer in reversed(layers):
+        items = _trans_items(items, layer._key_of_id, layer._obj_of_data)
+    return items
+
+
+def _push_inward(x, layers, *, outgoing_attr: str, ingoing_attr: str):
+    """Push a user-space ``x`` down to backend space through ``layers``, outermost first."""
+    for layer in layers:
+        x = _ingoing_trans(
+            layer, outgoing_attr=outgoing_attr, ingoing_attr=ingoing_attr
+        )(x)
+    return x
+
+
+def _push_value_inward(v, layers):
+    return _push_inward(
+        v, layers, outgoing_attr="_obj_of_data", ingoing_attr="_data_of_obj"
+    )
+
+
+def _push_key_inward(k, layers):
+    return _push_inward(
+        k, layers, outgoing_attr="_key_of_id", ingoing_attr="_id_of_key"
+    )
+
+
+# -------------------------------------------------------------------------------------
+# The bulk-read facade: what the views (and store authors) call
+
+
+def bulk_values(store) -> Iterator:
+    """Iterate ``store``'s values via the backend's bulk-read path, transforms honoured.
+
+    :raises NoBulkReadPath: when the bulk stream cannot be proven equivalent to
+        ``(store[k] for k in store)``.
+    """
+    source, layers = resolve_bulk_source(store, ITER_VALUES_METHOD)
+    return _map_outward(source.iter_values(), layers, "_obj_of_data")
+
+
+def bulk_items(store) -> KeyValStream:
+    """Iterate ``store``'s ``(key, value)`` pairs via the backend's bulk-read path.
+
+    :raises NoBulkReadPath: when the bulk stream cannot be proven equivalent to
+        ``((k, store[k]) for k in store)``.
+    """
+    source, layers = resolve_bulk_source(store, ITER_ITEMS_METHOD)
+    return _map_items_outward(source.iter_items(), layers)
+
+
+def bulk_contains_value(store, v) -> bool:
+    """Ask the backend whether ``v`` is one of ``store``'s values, in one round trip.
+
+    :raises NoBulkReadPath: when ``v`` cannot be pushed down to backend space.
+    """
+    source, layers = resolve_bulk_source(store, CONTAINS_VALUE_METHOD)
+    return source.contains_value(_push_value_inward(v, layers))
+
+
+def bulk_contains_item(store, item) -> bool:
+    """Ask the backend whether ``item`` is one of ``store``'s items, in one round trip.
+
+    :raises NoBulkReadPath: when ``item`` cannot be pushed down to backend space.
+    """
+    source, layers = resolve_bulk_source(store, CONTAINS_ITEM_METHOD)
+    k, v = item
+    return source.contains_item(
+        (_push_key_inward(k, layers), _push_value_inward(v, layers))
+    )
+
+
+def disable_bulk_read(store_cls: type) -> type:
+    """Class decorator declaring that inherited bulk-read methods are not to be trusted.
+
+    Use it on a class that changes what ``__getitem__`` returns (typically via
+    ``wrap_kvs(postget=...)``) while inheriting -- or being handed, by ``dol``'s
+    class-decorator wrapping -- bulk-read methods written for the *un*-changed
+    semantics. Views then take the correct per-key path instead.
+    """
+    setattr(store_cls, BULK_READ_IS_FAITHFUL_ATTR, False)
+    return store_cls
+
+
+# -------------------------------------------------------------------------------------
+# The views themselves
+
+
+class MongoValuesView(BaseValuesView):
+    """A ``values()`` view that uses the backend's bulk read when -- and only when --
+    that stream provably equals ``(store[k] for k in store)``."""
+
+    def __iter__(self):
+        try:
+            return bulk_values(self._mapping)
+        except NoBulkReadPath:
+            return (self._mapping[k] for k in self._mapping)
+
+    def __contains__(self, v):
+        try:
+            return bulk_contains_value(self._mapping, v)
+        except NoBulkReadPath:
+            return any(v == value for value in self)
+
+
+class MongoItemsView(BaseItemsView):
+    """An ``items()`` view that uses the backend's bulk read when -- and only when --
+    that stream provably equals ``((k, store[k]) for k in store)``."""
+
+    def __iter__(self):
+        try:
+            return bulk_items(self._mapping)
+        except NoBulkReadPath:
+            return ((k, self._mapping[k]) for k in self._mapping)
+
+    def __contains__(self, item):
+        try:
+            return bulk_contains_item(self._mapping, item)
+        except NoBulkReadPath:
+            k, v = item
+            try:
+                return self._mapping[k] == v
+            except KeyError:
+                return False