diff --git a/hypothesis-python/RELEASE.rst b/hypothesis-python/RELEASE.rst
new file mode 100644
index 0000000000..c8739170f4
--- /dev/null
+++ b/hypothesis-python/RELEASE.rst
@@ -0,0 +1,7 @@
+RELEASE_TYPE: minor
+
+This release changes :class:`hypothesis.stateful.Bundle` to use the internals of
+:func:`~hypothesis.strategies.sampled_from`, improving the `filter` and `map` methods.
+In addition to performance improvements, you can now ``consumes(some_bundle).filter(...)``!
+
+Thanks to Reagan Lee for this feature (:issue:`3944`).
\ No newline at end of file
diff --git a/hypothesis-python/src/hypothesis/stateful.py b/hypothesis-python/src/hypothesis/stateful.py
index 2103fdcb2c..44aa29a824 100644
--- a/hypothesis-python/src/hypothesis/stateful.py
+++ b/hypothesis-python/src/hypothesis/stateful.py
@@ -37,6 +37,7 @@
     InvalidDefinition,
 )
 from hypothesis.internal.compat import add_note, batched
+from hypothesis.internal.conjecture.data import ConjectureData
 from hypothesis.internal.conjecture.engine import BUFFER_SIZE
 from hypothesis.internal.conjecture.junkdrawer import gc_cumulative_time
 from hypothesis.internal.conjecture.utils import calc_label_from_name
@@ -54,8 +55,11 @@
 from hypothesis.strategies._internal.strategies import (
     Ex,
     OneOfStrategy,
+    SampledFromStrategy,
+    SampledFromTransformationsT,
     SearchStrategy,
     check_strategy,
+    filter_not_satisfied,
 )
 from hypothesis.utils.deprecation import note_deprecation
 from hypothesis.vendor.pretty import RepresentationPrinter
@@ -198,11 +202,11 @@ def output(s):
                     data = dict(data)
                     for k, v in list(data.items()):
                         if isinstance(v, VarReference):
-                            data[k] = machine.names_to_values[v.name]
+                            data[k] = v.value
                         elif isinstance(v, list) and all(
                             isinstance(item, VarReference) for item in v
                         ):
-                            data[k] = [machine.names_to_values[item.name] for item in v]
+                            data[k] = [item.value for item in v]
 
                     label = f"execute:rule:{rule.function.__name__}"
                     start = perf_counter()
@@ -330,7 +334,7 @@ def __init__(self) -> None:
         # copy since we pop from this as we run initialize rules.
         self._initialize_rules_to_run = setup_state.initializers.copy()
 
-        self.bundles: dict[str, list] = {}
+        self.bundles: dict[str, list[str]] = {}
         self.names_counters: collections.Counter = collections.Counter()
         self.names_list: list[str] = []
         self.names_to_values: dict[str, Any] = {}
@@ -459,7 +463,7 @@ def printer(obj, p, cycle, name=name):
                 if not _is_singleton(result):
                     self.__printer.singleton_pprinters.setdefault(id(result), printer)
                 self.names_to_values[name] = result
-                self.bundles.setdefault(target, []).append(VarReference(name))
+                self.bundles.setdefault(target, []).append(name)
 
     def check_invariants(self, settings, output, runtimes):
         for invar in self.invariants:
@@ -529,8 +533,12 @@ def __post_init__(self):
             assert not isinstance(v, BundleReferenceStrategy)
             if isinstance(v, Bundle):
                 bundles.append(v)
-                consume = isinstance(v, BundleConsumer)
-                v = BundleReferenceStrategy(v.name, consume=consume)
+                v = BundleReferenceStrategy(
+                    v.name,
+                    consume=v.consume,
+                    force_repr=v.force_repr,
+                    transformations=v._transformations,
+                )
             self.arguments_strategies[k] = v
         self.bundles = tuple(bundles)
 
@@ -564,25 +572,60 @@ def __hash__(self):
 self_strategy = st.runner()
 
 
-class BundleReferenceStrategy(SearchStrategy):
-    def __init__(self, name: str, *, consume: bool = False):
-        super().__init__()
+class BundleReferenceStrategy(SampledFromStrategy[Ex]):
+
+    def __init__(
+        self,
+        name: str,
+        *,
+        consume: bool = False,
+        force_repr: str | None = None,
+        transformations: SampledFromTransformationsT = (),
+    ):
+        super().__init__(
+            [...],
+            force_repr=force_repr,
+            transformations=transformations,
+        )  # Some random items that'll get replaced in do_draw
         self.name = name
         self.consume = consume
 
-    def do_draw(self, data):
-        machine = data.draw(self_strategy)
-        bundle = machine.bundle(self.name)
-        if not bundle:
+    def get_transformed_value(self, name: str) -> Ex:
+        value = self.machine.names_to_values[name]
+        return self._transform(value)
+
+    def get_element(self, i: int) -> int:
+        idx = self.elements[i]
+        name = self.bundle[idx]
+        value = self.get_transformed_value(name)
+        if value is filter_not_satisfied:
+            return filter_not_satisfied
+        return idx
+
+    def do_draw(self, data: ConjectureData) -> Ex:
+        self.machine = data.draw(self_strategy)
+        self.bundle = self.machine.bundle(self.name)
+        if not self.bundle:
             data.mark_invalid(f"Cannot draw from empty bundle {self.name!r}")
+
+        # We use both self.bundle and self.elements to make sure an index is
+        # used to safely pop.
+
         # Shrink towards the right rather than the left. This makes it easier
         # to delete data generated earlier, as when the error is towards the
         # end there can be a lot of hard to remove padding.
-        position = data.draw_integer(0, len(bundle) - 1, shrink_towards=len(bundle))
+        self.elements = range(len(self.bundle))[::-1]
+
+        position = super().do_draw(data)
+        name = self.bundle[position]
         if self.consume:
-            return bundle.pop(position)  # pragma: no cover  # coverage is flaky here
-        else:
-            return bundle[position]
+            self.bundle.pop(position)  # pragma: no cover  # coverage is flaky here
+
+        value = self.get_transformed_value(name)
+
+        # We need both reference and the value itself to pretty-print deterministically
+        # and maintain any transformations that is bundle-specific
+        return VarReference(name, value)
 
 
 class Bundle(SearchStrategy[Ex]):
@@ -605,26 +648,80 @@ class MyStateMachine(RuleBasedStateMachine):
 
     If the ``consume`` argument is set to True, then all values that are
     drawn from this bundle will be consumed (as above) when requested.
+
+    Bundles can be combined with |.map| and |.filter|:
+
+    .. code-block:: python
+
+        class Machine(RuleBasedStateMachine):
+            values = Bundle("values")
+
+            @initialize(target=values)
+            def populate_values(self):
+                return multiple(1, 2)
+
+            @rule(n=buns.map(lambda x: -x))
+            def use_map(self, n):
+                pass
+
+            @rule(n=buns.filter(lambda x: x > 1))
+            def use_filter(self, n):
+                pass
     """
 
     def __init__(
-        self, name: str, *, consume: bool = False, draw_references: bool = True
+        self,
+        name: str,
+        *,
+        consume: bool = False,
+        force_repr: str | None = None,
+        transformations: SampledFromTransformationsT = (),
     ) -> None:
         super().__init__()
         self.name = name
-        self.__reference_strategy = BundleReferenceStrategy(name, consume=consume)
-        self.draw_references = draw_references
+        self.__reference_strategy = BundleReferenceStrategy(
+            name,
+            consume=consume,
+            force_repr=force_repr,
+            transformations=transformations,
+        )
 
-    def do_draw(self, data):
-        machine = data.draw(self_strategy)
-        reference = data.draw(self.__reference_strategy)
-        return machine.names_to_values[reference.name]
+    @property
+    def consume(self):
+        return self.__reference_strategy.consume
+
+    @property
+    def force_repr(self):
+        return self.__reference_strategy.force_repr
+
+    @property
+    def _transformations(self):
+        return self.__reference_strategy._transformations
+
+    def do_draw(self, data: ConjectureData) -> Ex:
+        self.machine = data.draw(self_strategy)
+        var_reference = data.draw(self.__reference_strategy)
+        assert isinstance(var_reference, VarReference)
+        return var_reference.value
+
+    def __with_transform(self, method, fn):
+        return Bundle(
+            self.name,
+            consume=self.consume,
+            force_repr=self.force_repr,
+            transformations=(*self._transformations, (method, fn)),
+        )
+
+    def filter(self, condition):
+        return self.__with_transform("filter", condition)
+
+    def map(self, pack):
+        return self.__with_transform("map", pack)
 
     def __repr__(self):
-        consume = self.__reference_strategy.consume
-        if consume is False:
+        if self.consume is False:
             return f"Bundle(name={self.name!r})"
-        return f"Bundle(name={self.name!r}, {consume=})"
+        return f"Bundle(name={self.name!r}, {self.consume=})"
 
     def calc_is_empty(self, recur):
         # We assume that a bundle will grow over time
@@ -635,16 +732,7 @@ def is_currently_empty(self, data):
         # Hence drawing from it only returns the current state machine without
         # modifying the underlying choice sequence.
         machine = data.draw(self_strategy)
-        return not bool(machine.bundle(self.name))
-
-    def flatmap(self, expand):
-        if self.draw_references:
-            return type(self)(
-                self.name,
-                consume=self.__reference_strategy.consume,
-                draw_references=False,
-            ).flatmap(expand)
-        return super().flatmap(expand)
+        return bool(machine.bundle(self.name))
 
     def __hash__(self):
         # Making this hashable means we hit the fast path of "everything is
@@ -655,11 +743,6 @@ def __hash__(self):
         return hash(("Bundle", self.name))
 
 
-class BundleConsumer(Bundle[Ex]):
-    def __init__(self, bundle: Bundle[Ex]) -> None:
-        super().__init__(bundle.name, consume=True)
-
-
 def consumes(bundle: Bundle[Ex]) -> SearchStrategy[Ex]:
     """When introducing a rule in a RuleBasedStateMachine, this function can
     be used to mark bundles from which each value used in a step with the
@@ -675,7 +758,12 @@ def consumes(bundle: Bundle[Ex]) -> SearchStrategy[Ex]:
     """
     if not isinstance(bundle, Bundle):
         raise TypeError("Argument to be consumed must be a bundle.")
-    return BundleConsumer(bundle)
+    return Bundle(
+        bundle.name,
+        consume=True,
+        transformations=bundle._transformations,
+        force_repr=bundle.force_repr,
+    )
 
 
 @dataclass(slots=True, frozen=True)
@@ -720,7 +808,7 @@ def _convert_targets(targets, target):
                 )
             raise InvalidArgument(msg % (t, type(t)))
         while isinstance(t, Bundle):
-            if isinstance(t, BundleConsumer):
+            if t.consume:
                 note_deprecation(
                     f"Using consumes({t.name}) doesn't makes sense in this context.  "
                     "This will be an error in a future version of Hypothesis.",
@@ -964,6 +1052,7 @@ def rule_wrapper(*args, **kwargs):
 @dataclass(slots=True, frozen=True)
 class VarReference:
     name: str
+    value: Any
 
 
 # There are multiple alternatives for annotating the `precond` type, all of them
diff --git a/hypothesis-python/src/hypothesis/strategies/_internal/strategies.py b/hypothesis-python/src/hypothesis/strategies/_internal/strategies.py
index a9bf739386..7ff2a1d1cb 100644
--- a/hypothesis-python/src/hypothesis/strategies/_internal/strategies.py
+++ b/hypothesis-python/src/hypothesis/strategies/_internal/strategies.py
@@ -559,6 +559,12 @@ def is_hashable(value: object) -> bool:
     return _is_hashable(value)[0]
 
 
+SampledFromTransformationsT: "TypeAlias" = tuple[
+    tuple[Literal["filter", "map"], Callable[[Ex], Any]],
+    ...,
+]
+
+
 class SampledFromStrategy(SearchStrategy[Ex]):
     """A strategy which samples from a set of elements. This is essentially
     equivalent to using a OneOfStrategy over Just strategies but may be more
@@ -573,10 +579,7 @@ def __init__(
         *,
         force_repr: str | None = None,
         force_repr_braces: tuple[str, str] | None = None,
-        transformations: tuple[
-            tuple[Literal["filter", "map"], Callable[[Ex], Any]],
-            ...,
-        ] = (),
+        transformations: SampledFromTransformationsT = (),
     ):
         super().__init__()
         self.elements = cu.check_sample(elements, "sampled_from")
@@ -693,7 +696,7 @@ def _transform(
         # conservative than necessary
         element: Ex,  # type: ignore
     ) -> Ex | UniqueIdentifier:
-        # Used in UniqueSampledListStrategy
+        # Used in UniqueSampledListStrategy and BundleStrategy
         for name, f in self._transformations:
             if name == "map":
                 result = f(element)
diff --git a/hypothesis-python/tests/cover/test_stateful.py b/hypothesis-python/tests/cover/test_stateful.py
index d5f75aee84..61eacaea72 100644
--- a/hypothesis-python/tests/cover/test_stateful.py
+++ b/hypothesis-python/tests/cover/test_stateful.py
@@ -30,6 +30,7 @@
 from hypothesis.database import InMemoryExampleDatabase
 from hypothesis.errors import (
     DidNotReproduce,
+    FailedHealthCheck,
     Flaky,
     FlakyStrategyDefinition,
     InvalidArgument,
@@ -1475,24 +1476,141 @@ def test_lots_of_entropy():
     run_state_machine_as_test(LotsOfEntropyPerStepMachine)
 
 
-def test_flatmap():
+def test_filter():
     class Machine(RuleBasedStateMachine):
-        buns = Bundle("buns")
+        val = Bundle("val")
 
-        @initialize(target=buns)
-        def create_bun(self):
-            return 0
+        @initialize(target=val)
+        def initialize(self):
+            return multiple(1, 2, 3)
 
-        @rule(target=buns, bun=buns.flatmap(lambda x: just(x + 1)))
-        def use_flatmap(self, bun):
-            assert isinstance(bun, int)
-            return bun
+        @rule(
+            a1=val.filter(lambda x: x < 2),
+            a2=val.filter(lambda x: x > 2),
+            a3=val,
+        )
+        def fail_fast(self, a1, a2, a3):
+            raise AssertionError
 
-        @rule(bun=buns)
-        def use_directly(self, bun):
-            assert isinstance(bun, int)
+    Machine.TestCase.settings = NO_BLOB_SETTINGS
+    with pytest.raises(AssertionError) as err:
+        run_state_machine_as_test(Machine)
+
+    result = "\n".join(err.value.__notes__)
+    assert (
+        result
+        == """
+Falsifying example:
+state = Machine()
+val_0, val_1, val_2 = state.initialize()
+state.fail_fast(a1=val_0, a2=val_2, a3=val_2)
+state.teardown()
+""".strip()
+    )
+
+
+def test_consumes_filter():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def initialize(self):
+            return multiple(1, 2, 3)
+
+        @rule(
+            v1=consumes(values).filter(lambda x: x < 2),
+            v2=consumes(values).filter(lambda x: x > 2),
+            v3=consumes(values),
+        )
+        def fail_fast(self, v1, v2, v3):
+            raise AssertionError
+
+    Machine.TestCase.settings = NO_BLOB_SETTINGS
+    with pytest.raises(AssertionError) as err:
+        run_state_machine_as_test(Machine)
+
+    result = "\n".join(err.value.__notes__)
+    assert (
+        result
+        == """
+Falsifying example:
+state = Machine()
+values_0, values_1, values_2 = state.initialize()
+state.fail_fast(v1=values_0, v2=values_2, v3=values_1)
+state.teardown()
+""".strip()
+    )
+
+
+def test_consumes_map_with_filter():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def initialize(self):
+            return multiple(2, 4)
+
+        @rule(
+            v1=values.map(lambda x: x**2).filter(lambda x: x < 3**2),
+            v2=consumes(values).map(lambda x: x**2).filter(lambda x: x > 3**2),
+            v3=consumes(values),
+        )
+        def fail_fast(self, v1, v2, v3):
+            raise AssertionError
+
+    Machine.TestCase.settings = NO_BLOB_SETTINGS
+    with pytest.raises(AssertionError) as err:
+        run_state_machine_as_test(Machine)
+
+    result = "\n".join(err.value.__notes__)
+    assert (
+        result
+        == """
+Falsifying example:
+state = Machine()
+values_0, values_1 = state.initialize()
+state.fail_fast(v1=values_0, v2=values_1, v3=values_0)
+state.teardown()
+""".strip()
+    )
+
+
+def test_flatmap_with_combinations():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def create_values(self):
+            return 1
+
+        @rule(target=values, n=values.flatmap(lambda x: just(x + 1)))
+        def use_flatmap(self, n):
+            assert isinstance(n, int)
+            return n
+
+        @rule(
+            target=values,
+            n=values.flatmap(lambda x: just(-x)).filter(lambda x: x < -1),
+        )
+        def use_flatmap_filtered(self, n):
+            assert isinstance(n, int)
+            assert n < -1
+            return -n
+
+        @rule(
+            target=values,
+            n=values.flatmap(lambda x: just(x + 1)).map(lambda x: -x),
+        )
+        def use_flatmap_mapped(self, n):
+            assert isinstance(n, int)
+            assert n < 0
+            return -n
+
+        @rule(n=values)
+        def use_directly(self, n):
+            assert isinstance(n, int)
+            assert n >= 0
 
-    Machine.TestCase.settings = Settings(stateful_step_count=5, max_examples=10)
     run_state_machine_as_test(Machine)
 
 
@@ -1517,6 +1635,129 @@ def check(self, instance):
     run_state_machine_as_test(Machine)
 
 
+def test_map_with_combinations():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def create_values(self):
+            return multiple(1, 2)
+
+        @rule(n=values.map(lambda x: -x))
+        def use_map_base(self, n):
+            assert isinstance(n, int)
+            assert n < 0
+
+        @rule(
+            n=values.map(lambda x: -x).filter(lambda x: x < -1),
+        )
+        def use_flatmap_filtered(self, n):
+            assert isinstance(n, int)
+            assert n < -1
+
+        @rule(
+            n=values.map(lambda x: -x).flatmap(lambda x: just(abs(x) + 1)),
+        )
+        def use_flatmap_mapped(self, n):
+            assert isinstance(n, int)
+            assert n > 0
+
+        @rule(n=values)
+        def use_directly(self, n):
+            assert isinstance(n, int)
+            assert n > 0
+
+    run_state_machine_as_test(Machine)
+
+
+def test_filter_with_combinations():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def create_values(self):
+            return multiple(0, -1, -2)
+
+        @rule(n=values.filter(lambda x: x > 0))
+        def use_filter_base(self, n):
+            assert isinstance(n, int)
+            assert n > 0
+
+        @rule(
+            n=values.filter(lambda x: x > 0).flatmap(lambda x: just(-x)),
+        )
+        def use_filter_flatmapped(self, n):
+            assert isinstance(n, int)
+            assert n < 0
+
+        @rule(
+            n=values.filter(lambda x: x < 0).map(lambda x: -x),
+        )
+        def use_flatmap_mapped(self, n):
+            assert isinstance(n, int)
+            assert n > 0
+
+        @rule(n=values)
+        def use_directly(self, n):
+            assert isinstance(n, int)
+
+    run_state_machine_as_test(Machine)
+
+
+def test_filter_not_satisfied_healthcheck():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def create_values(self):
+            return 0
+
+        @rule(n=values.filter(lambda x: False))
+        def use_filter(self, n):
+            raise ValueError("This shouldn't be raised.")
+
+    Machine.TestCase.settings = Settings(
+        stateful_step_count=2,
+        max_examples=2,
+    )
+    with pytest.raises(FailedHealthCheck, match="HealthCheck.filter_too_much"):
+        run_state_machine_as_test(Machine)
+
+
+def test_mapped_values_assigned_properly():
+    class Machine(RuleBasedStateMachine):
+        values = Bundle("values")
+
+        @initialize(target=values)
+        def initialize(self):
+            return multiple("ret1", "ret2")
+
+        @rule(
+            v1=values,
+            v2=values.map(lambda x: x + x),
+            v3=consumes(values).map(lambda x: x + x),
+            v4=values,
+        )
+        def fail_fast(self, v1, v2, v3, v4):
+            raise AssertionError
+
+    Machine.TestCase.settings = NO_BLOB_SETTINGS
+    with pytest.raises(AssertionError) as err:
+        run_state_machine_as_test(Machine)
+
+    result = "\n".join(err.value.__notes__)
+    assert (
+        result
+        == """
+Falsifying example:
+state = Machine()
+values_0, values_1 = state.initialize()
+state.fail_fast(v1=values_1, v2=values_1, v3=values_1, v4=values_0)
+state.teardown()
+""".strip()
+    )
+
+
 def test_precondition_cannot_be_used_without_rule():
     class BadStateMachine(RuleBasedStateMachine):
         @precondition(lambda: True)