diff --git a/include/beman/execution/detail/get_completion_domain.hpp b/include/beman/execution/detail/get_completion_domain.hpp
index a150ba98..743c53aa 100644
--- a/include/beman/execution/detail/get_completion_domain.hpp
+++ b/include/beman/execution/detail/get_completion_domain.hpp
@@ -55,12 +55,10 @@ struct get_completion_domain_t : ::beman::execution::forwarding_query_t {
                                  ::beman::execution::detail::try_query(
                                      ::beman::execution::get_completion_scheduler<Tag>(q, e...),
                                      get_completion_domain_t<::beman::execution::set_value_t>{},
-                                     ::std::forward<Q>(q),
                                      ::std::forward<E>(e)...);
                              }) {
             return ::beman::execution::detail::try_query(::beman::execution::get_completion_scheduler<Tag>(q, e...),
                                                          get_completion_domain_t<::beman::execution::set_value_t>{},
-                                                         ::std::forward<Q>(q),
                                                          ::std::forward<E>(e)...);
         } else if constexpr (::beman::execution::scheduler<Q> && 0u != sizeof...(E)) {
             return ::beman::execution::default_domain{};
diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp
new file mode 100644
index 00000000..c1c2b492
--- /dev/null
+++ b/include/beman/execution/detail/parallel_scheduler.hpp
@@ -0,0 +1,470 @@
+// include/beman/execution/detail/parallel_scheduler.hpp            -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER
+
+#include <beman/execution/detail/common.hpp>
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <cstddef>
+#include <concepts>
+#include <exception>
+#include <memory>
+#include <optional>
+#include <span>
+#include <type_traits>
+#include <utility>
+#include <variant>
+#endif
+#ifdef BEMAN_HAS_MODULES
+import beman.execution.detail.bulk;
+import beman.execution.detail.connect;
+import beman.execution.detail.connect_result_t;
+import beman.execution.detail.completion_signatures;
+import beman.execution.detail.decayed_tuple;
+import beman.execution.detail.env_of_t;
+import beman.execution.detail.execution_policy;
+import beman.execution.detail.forward_like;
+import beman.execution.detail.get_completion_domain;
+import beman.execution.detail.get_completion_scheduler;
+import beman.execution.detail.get_completion_signatures;
+import beman.execution.detail.get_env;
+import beman.execution.detail.get_forward_progress_guarantee;
+import beman.execution.detail.get_stop_token;
+import beman.execution.detail.inplace_stop_source;
+import beman.execution.detail.meta.combine;
+import beman.execution.detail.meta.prepend;
+import beman.execution.detail.meta.unique;
+import beman.execution.detail.operation_state;
+import beman.execution.detail.receiver;
+import beman.execution.detail.scheduler;
+import beman.execution.detail.scheduler_tag;
+import beman.execution.detail.sender;
+import beman.execution.detail.start;
+import beman.execution.detail.tag_of_t;
+import beman.execution.detail.set_error;
+import beman.execution.detail.set_stopped;
+import beman.execution.detail.set_value;
+import beman.execution.detail.stop_propagator;
+import beman.execution.detail.stop_token_of_t;
+import beman.execution.detail.unreachable;
+import beman.execution.detail.value_types_of_t;
+#else
+#include <beman/execution/detail/bulk.hpp>
+#include <beman/execution/detail/connect.hpp>
+#include <beman/execution/detail/connect_result_t.hpp>
+#include <beman/execution/detail/completion_signatures.hpp>
+#include <beman/execution/detail/decayed_tuple.hpp>
+#include <beman/execution/detail/env_of_t.hpp>
+#include <beman/execution/detail/execution_policy.hpp>
+#include <beman/execution/detail/forward_like.hpp>
+#include <beman/execution/detail/get_completion_domain.hpp>
+#include <beman/execution/detail/get_completion_scheduler.hpp>
+#include <beman/execution/detail/get_completion_signatures.hpp>
+#include <beman/execution/detail/get_env.hpp>
+#include <beman/execution/detail/get_forward_progress_guarantee.hpp>
+#include <beman/execution/detail/get_stop_token.hpp>
+#include <beman/execution/detail/inplace_stop_source.hpp>
+#include <beman/execution/detail/meta_combine.hpp>
+#include <beman/execution/detail/meta_prepend.hpp>
+#include <beman/execution/detail/meta_unique.hpp>
+#include <beman/execution/detail/operation_state.hpp>
+#include <beman/execution/detail/receiver.hpp>
+#include <beman/execution/detail/scheduler.hpp>
+#include <beman/execution/detail/scheduler_tag.hpp>
+#include <beman/execution/detail/sender.hpp>
+#include <beman/execution/detail/set_error.hpp>
+#include <beman/execution/detail/set_stopped.hpp>
+#include <beman/execution/detail/set_value.hpp>
+#include <beman/execution/detail/start.hpp>
+#include <beman/execution/detail/stop_propagator.hpp>
+#include <beman/execution/detail/stop_token_of_t.hpp>
+#include <beman/execution/detail/tag_of_t.hpp>
+#include <beman/execution/detail/unreachable.hpp>
+#include <beman/execution/detail/value_types_of_t.hpp>
+#endif
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution::parallel_scheduler_replacement {
+
+struct receiver_proxy {
+    virtual ~receiver_proxy() = default;
+
+    virtual auto set_value() noexcept -> void                     = 0;
+    virtual auto set_error(::std::exception_ptr) noexcept -> void = 0;
+    virtual auto set_stopped() noexcept -> void                   = 0;
+
+    template <class P, class Query>
+        requires(::std::same_as<P, ::std::remove_cv_t<P>> && ::std::is_object_v<P> && !::std::is_array_v<P>)
+    auto try_query(Query) const noexcept -> ::std::optional<P> {
+        if constexpr (::std::same_as<Query, ::beman::execution::get_stop_token_t> &&
+                      ::std::same_as<P, ::beman::execution::inplace_stop_token>) {
+            return query_stop_token();
+        }
+        // TODO: support more queries
+        return ::std::nullopt;
+    }
+
+  private:
+    virtual auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> {
+        return std::nullopt;
+    }
+};
+
+struct bulk_item_receiver_proxy : receiver_proxy {
+    virtual auto execute(::std::size_t, ::std::size_t) noexcept -> void = 0;
+};
+
+struct parallel_scheduler_backend {
+    virtual ~parallel_scheduler_backend() = default;
+
+    virtual auto schedule(receiver_proxy&, ::std::span<::std::byte>) noexcept -> void = 0;
+    virtual auto schedule_bulk_chunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept
+        -> void = 0;
+    virtual auto schedule_bulk_unchunked(::std::size_t, bulk_item_receiver_proxy&, ::std::span<::std::byte>) noexcept
+        -> void = 0;
+};
+
+// TODO(P2079R10): provide the project-supported link-time replaceability hook.
+auto query_parallel_scheduler_backend() -> ::std::shared_ptr<parallel_scheduler_backend>;
+
+} // namespace beman::execution::parallel_scheduler_replacement
+
+namespace beman::execution::detail {
+template <bool IsChunked, typename Policy, typename Shape, typename Fn, typename Child>
+struct psched_bulk_sender {
+    using sender_concept = ::beman::execution::sender_tag;
+
+    static constexpr bool is_parallel_policy = ::std::same_as<Policy, ::beman::execution::parallel_policy> ||
+                                               ::std::same_as<Policy, ::beman::execution::parallel_unsequenced_policy>;
+    template <typename Rcvr>
+    struct rcvr_proxy : ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy {
+        using result_type = ::beman::execution::detail::meta::
+            prepend<::std::monostate, ::beman::execution::value_types_of_t<Child, ::beman::execution::env_of_t<Rcvr>>>;
+        using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t<Rcvr>>;
+
+        rcvr_proxy(Rcvr rcvr, Policy policy, Shape shape, Fn fn) noexcept
+            : sp(::beman::execution::get_stop_token(::beman::execution::get_env(rcvr))),
+              rcvr(::std::move(rcvr)),
+              policy(::std::move(policy)),
+              shape(::std::move(shape)),
+              fn(::std::move(fn)) {}
+
+        auto get_env() const noexcept { return ::beman::execution::get_env(rcvr); }
+
+        auto execute(::std::size_t begin, ::std::size_t end) noexcept -> void final {
+            const Shape first   = is_parallel_policy ? static_cast<Shape>(begin) : Shape(0);
+            const Shape last    = is_parallel_policy ? static_cast<Shape>(end) : shape;
+            const auto  call_fn = [=, this]<typename... Args>(const Args&... args) {
+                if constexpr (IsChunked) {
+                    fn(first, last, args...);
+                } else {
+                    for (Shape i = first; i < last; ++i) {
+                        fn(i, args...);
+                    }
+                }
+            };
+            std::visit(
+                [&]<typename T>(T& tpl) {
+                    if constexpr (::std::same_as<T, ::std::monostate>) {
+                        ::beman::execution::detail::unreachable();
+                    } else {
+                        ::std::apply(call_fn, ::std::move(tpl));
+                    }
+                },
+                result);
+        }
+
+        auto set_value() noexcept -> void final {
+            const auto call_set_value = [this]<typename... Args>(Args&&... args) {
+                ::beman::execution::set_value(::std::move(rcvr), ::std::forward<Args>(args)...);
+            };
+            try {
+                std::visit(
+                    [&]<typename T>(T tpl) {
+                        if constexpr (::std::same_as<T, std::monostate>) {
+                            ::beman::execution::detail::unreachable();
+                        } else {
+                            ::std::apply(call_set_value, ::std::move(tpl));
+                        }
+                    },
+                    ::std::move(result));
+            } catch (...) {
+                this->set_error(::std::current_exception());
+            }
+        }
+
+        template <typename E>
+        auto set_error(E e) noexcept -> void {
+            ::beman::execution::set_error(::std::move(rcvr), ::std::move(e));
+        }
+
+        auto set_error(::std::exception_ptr e) noexcept -> void final { this->set_error<>(::std::move(e)); }
+
+        auto set_stopped() noexcept -> void final { ::beman::execution::set_stopped(::std::move(rcvr)); }
+
+      private:
+        auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> final {
+            return sp.get_token();
+        }
+
+      public:
+        result_type                                                                     result;
+        [[no_unique_address]] ::beman::execution::detail::stop_propagator<stop_token_t> sp;
+        Rcvr                                                                            rcvr;
+        Policy                                                                          policy;
+        Shape                                                                           shape;
+        Fn                                                                              fn;
+    };
+
+    template <typename Rcvr>
+    struct state {
+        using operation_state_concept = ::beman::execution::operation_state_tag;
+        using result_type             = ::beman::execution::detail::meta::
+            prepend<::std::monostate, ::beman::execution::value_types_of_t<Child, ::beman::execution::env_of_t<Rcvr>>>;
+
+        struct receiver_ref {
+            using receiver_concept = ::beman::execution::receiver_tag;
+
+            template <typename... Args>
+            auto set_value(Args&&... args) noexcept -> void try {
+                using arg_t = ::beman::execution::detail::decayed_tuple<Args...>;
+                proxy->result.template emplace<arg_t>(::std::forward<Args>(args)...);
+                auto backend = ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend();
+                if (backend == nullptr) [[unlikely]] {
+                    ::std::terminate();
+                }
+                const ::std::size_t       s = is_parallel_policy ? proxy->shape : 1uz;
+                alignas(void*)::std::byte storage[sizeof(void*) * 4uz];
+                if constexpr (IsChunked) {
+                    backend->schedule_bulk_chunked(s, *proxy, storage);
+                } else {
+                    backend->schedule_bulk_unchunked(s, *proxy, storage);
+                }
+            } catch (...) {
+                proxy->set_error(::std::current_exception());
+            }
+
+            template <typename E>
+            auto set_error(E e) noexcept -> void {
+                proxy->set_error(::std::move(e));
+            }
+
+            auto set_stopped() noexcept -> void { proxy->set_stopped(); }
+
+            auto get_env() const noexcept { return proxy->get_env(); }
+
+            rcvr_proxy<Rcvr>* proxy;
+        };
+        using sub_state_t = ::beman::execution::connect_result_t<Child, receiver_ref>;
+
+        state(Child child, Rcvr rcvr, Policy policy, Shape shape, Fn fn)
+            : proxy(::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)),
+              sub_state(::beman::execution::connect(::std::move(child), receiver_ref{&proxy})) {}
+
+        auto start() & noexcept -> void { ::beman::execution::start(sub_state); }
+
+        rcvr_proxy<Rcvr> proxy;
+        sub_state_t      sub_state;
+    };
+
+    template <typename, typename... Env>
+    static consteval auto get_completion_signatures() {
+        constexpr auto compl_sigs = ::beman::execution::get_completion_signatures<Child, Env...>();
+        return ::beman::execution::detail::meta::unique<::beman::execution::detail::meta::combine<
+            ::std::remove_cvref_t<decltype(compl_sigs)>,
+            ::beman::execution::completion_signatures<::beman::execution::set_error_t(::std::exception_ptr)>>>{};
+    }
+
+    template <typename Rcvr>
+    auto connect(Rcvr rcvr) && noexcept {
+        return state<Rcvr>{
+            ::std::move(child), ::std::move(rcvr), ::std::move(policy), ::std::move(shape), ::std::move(fn)};
+    }
+
+    auto get_env() const noexcept { return ::beman::execution::get_env(child); }
+
+    [[no_unique_address]] ::std::bool_constant<IsChunked> _;
+    Policy                                                policy;
+    Shape                                                 shape;
+    Fn                                                    fn;
+    Child                                                 child;
+};
+
+struct parallel_scheduler_domain {
+    template <typename Sndr>
+        requires ::std::same_as<::beman::execution::tag_of_t<Sndr>, ::beman::execution::bulk_chunked_t> ||
+                 ::std::same_as<::beman::execution::tag_of_t<Sndr>, ::beman::execution::bulk_unchunked_t>
+    static auto transform_sender(::beman::execution::set_value_t, Sndr&& sndr, const auto&) noexcept {
+        auto&& [_, data, child]  = ::std::forward<Sndr>(sndr);
+        auto [policy, shape, fn] = ::beman::execution::detail::forward_like<Sndr>(data);
+        return ::beman::execution::detail::psched_bulk_sender{
+            ::std::bool_constant<
+                ::std::same_as<::beman::execution::tag_of_t<Sndr>, ::beman::execution::bulk_chunked_t>>{},
+            ::std::move(policy),
+            ::std::move(shape),
+            ::std::move(fn),
+            ::beman::execution::detail::forward_like<Sndr>(child)};
+    }
+};
+} // namespace beman::execution::detail
+
+namespace beman::execution {
+
+class parallel_scheduler {
+    using backend_type = ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend;
+
+  public:
+    using scheduler_concept = ::beman::execution::scheduler_tag;
+
+    class sender;
+
+    parallel_scheduler()  = delete;
+    ~parallel_scheduler() = default;
+
+    parallel_scheduler(const parallel_scheduler&) noexcept                    = default;
+    parallel_scheduler(parallel_scheduler&&) noexcept                         = default;
+    auto operator=(const parallel_scheduler&) noexcept -> parallel_scheduler& = default;
+    auto operator=(parallel_scheduler&&) noexcept -> parallel_scheduler&      = default;
+
+    auto operator==(const parallel_scheduler& other) const noexcept -> bool {
+        return this->backend_ == other.backend_;
+    }
+
+    static constexpr auto query(::beman::execution::get_forward_progress_guarantee_t) noexcept
+        -> ::beman::execution::forward_progress_guarantee {
+        return ::beman::execution::forward_progress_guarantee::parallel;
+    }
+
+    static constexpr auto query(::beman::execution::get_completion_domain_t<::beman::execution::set_value_t>) noexcept
+        -> ::beman::execution::detail::parallel_scheduler_domain {
+        return {};
+    }
+
+    auto schedule() const noexcept -> sender;
+
+  private:
+    explicit parallel_scheduler(::std::shared_ptr<backend_type> backend) noexcept : backend_(::std::move(backend)) {}
+
+    ::std::shared_ptr<backend_type> backend_;
+
+    friend auto get_parallel_scheduler() -> parallel_scheduler;
+};
+
+class parallel_scheduler::sender {
+    using backend_type = ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend;
+
+  public:
+    using sender_concept = ::beman::execution::sender_tag;
+    using completion_signatures =
+        ::beman::execution::completion_signatures<::beman::execution::set_value_t(),
+                                                  ::beman::execution::set_error_t(::std::exception_ptr),
+                                                  ::beman::execution::set_stopped_t()>;
+
+    class env {
+      public:
+        explicit env(::std::shared_ptr<backend_type> backend) noexcept : backend_(::std::move(backend)) {}
+
+        auto
+        query(const ::beman::execution::get_completion_scheduler_t<::beman::execution::set_value_t>&) const noexcept
+            -> ::beman::execution::parallel_scheduler {
+            return ::beman::execution::parallel_scheduler{this->backend_};
+        }
+
+      private:
+        ::std::shared_ptr<backend_type> backend_;
+    };
+
+    template <typename Rcvr>
+    struct rcvr_proxy : ::beman::execution::parallel_scheduler_replacement::receiver_proxy {
+        using stop_token_t = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t<Rcvr>>;
+
+        explicit rcvr_proxy(Rcvr rcvr) noexcept
+            : sp_(::beman::execution::get_stop_token(::beman::execution::get_env(rcvr))), rcvr_(::std::move(rcvr)) {}
+
+        auto set_value() noexcept -> void final { ::beman::execution::set_value(::std::move(this->rcvr_)); }
+
+        auto set_error(::std::exception_ptr error) noexcept -> void final {
+            ::beman::execution::set_error(::std::move(this->rcvr_), ::std::move(error));
+        }
+
+        auto set_stopped() noexcept -> void final { ::beman::execution::set_stopped(::std::move(this->rcvr_)); }
+
+      private:
+        auto query_stop_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> final {
+            return sp_.get_token();
+        }
+
+        [[no_unique_address]] ::beman::execution::detail::stop_propagator<stop_token_t> sp_;
+        Rcvr                                                                            rcvr_;
+    };
+
+    template <::beman::execution::receiver Rcvr>
+    class operation {
+      public:
+        using operation_state_concept = ::beman::execution::operation_state_tag;
+        using stop_token_t            = ::beman::execution::stop_token_of_t<::beman::execution::env_of_t<Rcvr>>;
+
+        operation(::std::shared_ptr<backend_type> backend,
+                  Rcvr&& rcvr) noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t<Rcvr>, Rcvr>)
+            : backend_(::std::move(backend)), proxy_(::std::forward<Rcvr>(rcvr)) {}
+
+        auto start() & noexcept -> void {
+            auto stop_token =
+                proxy_.template try_query<::beman::execution::inplace_stop_token>(::beman::execution::get_stop_token);
+            if (stop_token.has_value() && stop_token->stop_requested()) {
+                proxy_.set_stopped();
+                return;
+            }
+            this->backend_->schedule(proxy_, ::std::span<::std::byte>{this->storage_});
+        }
+
+      private:
+        ::std::shared_ptr<backend_type>         backend_;
+        rcvr_proxy<::std::remove_cvref_t<Rcvr>> proxy_;
+        alignas(void*)::std::byte storage_[sizeof(void*) * 4]{};
+    };
+
+    explicit sender(::std::shared_ptr<backend_type> backend) noexcept : backend_(::std::move(backend)) {}
+
+    template <typename...>
+    static consteval auto get_completion_signatures() noexcept -> completion_signatures {
+        return {};
+    }
+
+    auto get_env() const noexcept -> env { return env{this->backend_}; }
+
+    template <::beman::execution::receiver Rcvr>
+    auto connect(Rcvr&& rcvr) & noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t<Rcvr>, Rcvr>)
+        -> operation<Rcvr> {
+        return operation<Rcvr>{this->backend_, ::std::forward<Rcvr>(rcvr)};
+    }
+
+    template <::beman::execution::receiver Rcvr>
+    auto connect(Rcvr&& rcvr) && noexcept(::std::is_nothrow_constructible_v<::std::remove_cvref_t<Rcvr>, Rcvr>)
+        -> operation<Rcvr> {
+        return operation<Rcvr>{::std::move(this->backend_), ::std::forward<Rcvr>(rcvr)};
+    }
+
+  private:
+    ::std::shared_ptr<backend_type> backend_;
+};
+
+inline auto parallel_scheduler::schedule() const noexcept -> sender { return sender{this->backend_}; }
+
+[[nodiscard]] inline auto get_parallel_scheduler() -> parallel_scheduler {
+    auto backend = ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend();
+    if (backend == nullptr) [[unlikely]] {
+        ::std::terminate();
+    }
+    return parallel_scheduler{::std::move(backend)};
+}
+
+} // namespace beman::execution
+
+// ----------------------------------------------------------------------------
+
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_PARALLEL_SCHEDULER
diff --git a/include/beman/execution/detail/stop_propagator.hpp b/include/beman/execution/detail/stop_propagator.hpp
new file mode 100644
index 00000000..3c092da3
--- /dev/null
+++ b/include/beman/execution/detail/stop_propagator.hpp
@@ -0,0 +1,77 @@
+// include/beman/execution/detail/stop_propagator.hpp                -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR
+#define INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR
+
+#include <beman/execution/detail/common.hpp>
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <functional>
+#include <optional>
+#include <type_traits>
+#include <utility>
+#endif
+#ifdef BEMAN_HAS_MODULES
+import beman.execution.detail.inplace_stop_source;
+import beman.execution.detail.stoppable_source;
+import beman.execution.detail.stoppable_token;
+import beman.execution.detail.stop_callback_for_t;
+import beman.execution.detail.unstoppable_token;
+#else
+#include <beman/execution/detail/inplace_stop_source.hpp>
+#include <beman/execution/detail/stoppable_source.hpp>
+#include <beman/execution/detail/stoppable_token.hpp>
+#include <beman/execution/detail/stop_callback_for_t.hpp>
+#include <beman/execution/detail/unstoppable_token.hpp>
+#endif
+
+// ----------------------------------------------------------------------------
+
+namespace beman::execution::detail {
+template <::beman::execution::stoppable_token StopToken>
+class stop_propagator {
+  public:
+    explicit stop_propagator(StopToken stop_token) noexcept
+        : stop_callback_(::std::move(stop_token), ::std::bind_front(&stop_propagator::propagate, this)) {}
+
+    auto get_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> {
+        return stop_source_.get_token();
+    }
+
+  private:
+    auto propagate() noexcept -> void { stop_source_.request_stop(); }
+
+  private:
+    using stop_cb_t = decltype(::std::bind_front(&stop_propagator::propagate, ::std::declval<stop_propagator*>()));
+
+    ::beman::execution::inplace_stop_source                       stop_source_;
+    ::beman::execution::stop_callback_for_t<StopToken, stop_cb_t> stop_callback_;
+};
+
+template <>
+class stop_propagator<::beman::execution::inplace_stop_token> {
+  public:
+    explicit stop_propagator(::beman::execution::inplace_stop_token stop_token) noexcept : stop_token_(stop_token) {}
+
+    auto get_token() const noexcept -> ::std::optional<::beman::execution::inplace_stop_token> { return stop_token_; }
+
+  private:
+    ::beman::execution::inplace_stop_token stop_token_;
+};
+
+template <::beman::execution::unstoppable_token StopToken>
+class stop_propagator<StopToken> {
+  public:
+    explicit stop_propagator(StopToken) noexcept {}
+
+    static auto get_token() noexcept -> ::std::optional<::beman::execution::inplace_stop_token> {
+        return ::std::nullopt;
+    }
+};
+} // namespace beman::execution::detail
+
+// ----------------------------------------------------------------------------
+
+#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_STOP_PROPAGATOR
diff --git a/include/beman/execution/execution.hpp b/include/beman/execution/execution.hpp
index 70a5f5c5..57656396 100644
--- a/include/beman/execution/execution.hpp
+++ b/include/beman/execution/execution.hpp
@@ -42,6 +42,7 @@ import beman.execution.detail.let;
 import beman.execution.detail.matching_sig;
 import beman.execution.detail.on;
 import beman.execution.detail.operation_state;
+import beman.execution.detail.parallel_scheduler;
 import beman.execution.detail.prop;
 import beman.execution.detail.read_env;
 import beman.execution.detail.receiver;
@@ -107,6 +108,7 @@ import beman.execution.detail.write_env;
 #include <beman/execution/detail/matching_sig.hpp>
 #include <beman/execution/detail/on.hpp>
 #include <beman/execution/detail/operation_state.hpp>
+#include <beman/execution/detail/parallel_scheduler.hpp>
 #include <beman/execution/detail/prop.hpp>
 #include <beman/execution/detail/read_env.hpp>
 #include <beman/execution/detail/receiver.hpp>
diff --git a/include/beman/execution26/execution.hpp b/include/beman/execution26/execution.hpp
index d414719b..5b685670 100644
--- a/include/beman/execution26/execution.hpp
+++ b/include/beman/execution26/execution.hpp
@@ -42,6 +42,7 @@ using ::beman::execution::get_domain;
 using ::beman::execution::get_domain_t;
 using ::beman::execution::get_env;
 using ::beman::execution::get_env_t;
+using ::beman::execution::get_parallel_scheduler;
 using ::beman::execution::get_scheduler;
 using ::beman::execution::get_scheduler_t;
 using ::beman::execution::get_stop_token;
@@ -62,6 +63,7 @@ using ::beman::execution::let_value;
 using ::beman::execution::let_value_t;
 using ::beman::execution::operation_state;
 using ::beman::execution::operation_state_tag;
+using ::beman::execution::parallel_scheduler;
 using ::beman::execution::read_env;
 using ::beman::execution::receiver;
 using ::beman::execution::receiver_of;
@@ -107,6 +109,13 @@ using ::beman::execution::when_all_with_variant;
 using ::beman::execution::when_all_with_variant_t;
 using ::beman::execution::with_awaitable_senders;
 
+namespace parallel_scheduler_replacement {
+using ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy;
+using ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend;
+using ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend;
+using ::beman::execution::parallel_scheduler_replacement::receiver_proxy;
+} // namespace parallel_scheduler_replacement
+
 } // namespace beman::execution26
 
 // ----------------------------------------------------------------------------
diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt
index e3cbe916..02d9458f 100644
--- a/src/beman/execution/CMakeLists.txt
+++ b/src/beman/execution/CMakeLists.txt
@@ -144,6 +144,7 @@ target_sources(
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/on_stop_request.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/operation_state.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/operation_state_task.hpp
+                ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/parallel_scheduler.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/product_type.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/poly.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/promise_env.hpp
@@ -187,6 +188,7 @@ target_sources(
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/state_rep.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/state_type.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_callback_for_t.hpp
+                ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_propagator.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_source.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_source_of.hpp
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/stop_token_of_t.hpp
@@ -350,6 +352,7 @@ if(BEMAN_USE_MODULES)
                     on.cppm
                     operation_state_task.cppm
                     operation_state.cppm
+                    parallel_scheduler.cppm
                     product_type.cppm
                     prop.cppm
                     query_with_default.cppm
@@ -391,6 +394,7 @@ if(BEMAN_USE_MODULES)
                     state_type.cppm
                     stoppable_callback_for.cppm
                     stop_callback_for_t.cppm
+                    stop_propagator.cppm
                     stop_source.cppm
                     stop_token_of_t.cppm
                     stop_when.cppm
diff --git a/src/beman/execution/execution-detail.cppm b/src/beman/execution/execution-detail.cppm
index 36f1328c..1792ed0c 100644
--- a/src/beman/execution/execution-detail.cppm
+++ b/src/beman/execution/execution-detail.cppm
@@ -59,6 +59,7 @@ export import beman.execution.detail.non_assignable;
 export import beman.execution.detail.nothrow_callable;
 export import beman.execution.detail.notify;
 export import beman.execution.detail.operation_state_task;
+export import beman.execution.detail.parallel_scheduler;
 export import beman.execution.detail.product_type;
 export import beman.execution.detail.query_with_default;
 export import beman.execution.detail.queryable;
diff --git a/src/beman/execution/execution.cppm b/src/beman/execution/execution.cppm
index 9d0c91d7..a43dcf2e 100644
--- a/src/beman/execution/execution.cppm
+++ b/src/beman/execution/execution.cppm
@@ -46,6 +46,7 @@ import beman.execution.detail.never_stop_token;
 import beman.execution.detail.nostopstate;
 import beman.execution.detail.on;
 export import beman.execution.detail.operation_state; // [exec.opstate], operation states
+import beman.execution.detail.parallel_scheduler;
 import beman.execution.detail.prop;
 import beman.execution.detail.read_env;
 import beman.execution.detail.run_loop;
@@ -242,6 +243,17 @@ export using ::beman::execution::stopped_as_error;
 // [exec.run.loop], run_loop
 export using ::beman::execution::run_loop;
 
+// [exec.parallel.scheduler], parallel scheduler
+export using ::beman::execution::parallel_scheduler;
+export using ::beman::execution::get_parallel_scheduler;
+
+namespace parallel_scheduler_replacement {
+export using ::beman::execution::parallel_scheduler_replacement::receiver_proxy;
+export using ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy;
+export using ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend;
+export using ::beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend;
+} // namespace parallel_scheduler_replacement
+
 // [exec.consumers], consumers
 export using ::beman::execution::sync_wait_t;
 export using ::beman::execution::sync_wait_with_variant_t;
diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm
new file mode 100644
index 00000000..8fcf75e4
--- /dev/null
+++ b/src/beman/execution/parallel_scheduler.cppm
@@ -0,0 +1,19 @@
+module;
+// src/beman/execution/parallel_scheduler.cppm                       -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution/detail/parallel_scheduler.hpp>
+
+export module beman.execution.detail.parallel_scheduler;
+
+namespace beman::execution {
+export using beman::execution::parallel_scheduler;
+export using beman::execution::get_parallel_scheduler;
+
+namespace parallel_scheduler_replacement {
+export using beman::execution::parallel_scheduler_replacement::receiver_proxy;
+export using beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy;
+export using beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend;
+export using beman::execution::parallel_scheduler_replacement::query_parallel_scheduler_backend;
+} // namespace parallel_scheduler_replacement
+} // namespace beman::execution
diff --git a/src/beman/execution/stop_propagator.cppm b/src/beman/execution/stop_propagator.cppm
new file mode 100644
index 00000000..b226a052
--- /dev/null
+++ b/src/beman/execution/stop_propagator.cppm
@@ -0,0 +1,11 @@
+module;
+// src/beman/execution/stop_propagator.cppm                     -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <beman/execution/detail/stop_propagator.hpp>
+
+export module beman.execution.detail.stop_propagator;
+
+namespace beman::execution::detail {
+export using beman::execution::detail::stop_propagator;
+} // namespace beman::execution::detail
diff --git a/tests/beman/execution/CMakeLists.txt b/tests/beman/execution/CMakeLists.txt
index 59e58f6a..d6b49092 100644
--- a/tests/beman/execution/CMakeLists.txt
+++ b/tests/beman/execution/CMakeLists.txt
@@ -36,6 +36,7 @@ list(
     exec-just.test
     exec-let.test
     exec-on.test
+    exec-parallel-scheduler.test
     exec-opstate-start.test
     exec-opstate.test
     exec-prop.test
diff --git a/tests/beman/execution/exec-associate.test.cpp b/tests/beman/execution/exec-associate.test.cpp
index 9d2987ed..07144f32 100644
--- a/tests/beman/execution/exec-associate.test.cpp
+++ b/tests/beman/execution/exec-associate.test.cpp
@@ -119,7 +119,6 @@ TEST(exec_associate) {
         static_assert(std::same_as<test_std::completion_signatures<test_std::set_value_t()>,
                                    test_std::completion_signatures_of_t<snd0_t, test_std::env<>>>);
 
-#ifndef _MSC_VER //-dk:TODO MSVC++ struggles with more than one of these test
         using snd1_t = decltype(test_std::associate(test_std::just(std::string{}), null_token{}));
         static_assert(std::same_as<test_std::completion_signatures<test_std::set_value_t(std::string)>,
                                    test_std::completion_signatures_of_t<snd1_t, test_std::env<>>>);
@@ -136,10 +135,8 @@ TEST(exec_associate) {
         using snd4_t = decltype(test_std::associate(test_std::just(std::ref(i)), null_token{}));
         static_assert(std::same_as<test_std::completion_signatures<test_std::set_value_t(std::reference_wrapper<int>)>,
                                    test_std::completion_signatures_of_t<snd4_t, test_std::env<>>>);
-#endif
     }
 
-#ifndef _MSC_VER //-dk:TODO MSVC++ struggles with more than one of these test
     // Identity behavior with null_token for value path + piping works.
     {
         {
@@ -241,5 +238,4 @@ TEST(exec_associate) {
         ASSERT(completes_with_value(test_std::just(1) | test_std::associate(null_token{})));
         ASSERT(!completes_with_value(test_std::just(1) | test_std::associate(expired_token{})));
     }
-#endif
 }
diff --git a/tests/beman/execution/exec-bulk.test.cpp b/tests/beman/execution/exec-bulk.test.cpp
index 0febe816..01a8b0a7 100644
--- a/tests/beman/execution/exec-bulk.test.cpp
+++ b/tests/beman/execution/exec-bulk.test.cpp
@@ -391,10 +391,10 @@ struct pstl_for_each_sender {
         template <typename... Args>
         auto set_value(Args&&... args) noexcept -> void {
             try {
-                auto iota = std::views::iota(Shape(0), shape);
-                std::for_each(policy, std::ranges::begin(iota), std::ranges::end(iota), [&](Shape i) {
+                auto indices = std::views::iota(Shape(0), shape);
+                std::for_each(policy, std::ranges::begin(indices), std::ranges::end(indices), [&](Shape i) {
                     if constexpr (IsChunked) {
-                        std::invoke(fn, i, i + 1, args...);
+                        std::invoke(fn, i, i + Shape(1), args...);
                     } else {
                         std::invoke(fn, i, args...);
                     };
@@ -430,7 +430,8 @@ struct pstl_for_each_sender {
 
     template <typename Rcvr>
     auto connect(Rcvr rcvr) && noexcept {
-        return test_std::connect(child, receiver{std::move(rcvr), std::move(policy), std::move(shape), std::move(fn)});
+        return test_std::connect(std::move(child),
+                                 receiver{std::move(rcvr), std::move(policy), std::move(shape), std::move(fn)});
     }
 
     auto get_env() const noexcept { return test_std::get_env(child); }
diff --git a/tests/beman/execution/exec-just.test.cpp b/tests/beman/execution/exec-just.test.cpp
index 8feab9f9..3ce04652 100644
--- a/tests/beman/execution/exec-just.test.cpp
+++ b/tests/beman/execution/exec-just.test.cpp
@@ -191,16 +191,17 @@ auto test_just_allocator() -> void {
     ASSERT(resource.count == 0u);
     auto copy(std::make_obj_using_allocator<std::pmr::string>(std::pmr::polymorphic_allocator<>(&resource), str));
     test::use(copy);
-    ASSERT(resource.count == 1u);
+    auto old_count = resource.count;
+    ASSERT(old_count > 0u);
 
     auto env{test_std::get_env(receiver)};
     auto alloc{test_std::get_allocator(env)};
     test::use(alloc);
 
-    ASSERT(resource.count == 1u);
+    ASSERT(resource.count == old_count);
     auto state{test_std::connect(std::move(sender), memory_receiver{&resource})};
     test::use(state);
-    ASSERT(resource.count == 2u);
+    ASSERT(resource.count > old_count);
 }
 
 auto test_completion_signatures() -> void {
@@ -220,10 +221,7 @@ TEST(exec_just) {
         test_just_constraints();
         test_just();
         test_completion_signatures();
-#ifndef _MSC_VER
-        //-dk:TODO re-enable allocator test for MSVC++
         test_just_allocator();
-#endif
     } catch (...) {
         // NOLINTNEXTLINE(cert-dcl03-c,hicpp-static-assert,misc-static-assert)
         ASSERT(nullptr == "the just tests shouldn't throw");
diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp
new file mode 100644
index 00000000..4b17ae79
--- /dev/null
+++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp
@@ -0,0 +1,292 @@
+// src/beman/execution/tests/exec-parallel-scheduler.test.cpp       -*-C++-*-
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+
+#include <numeric>
+#include <atomic>
+#include <concepts>
+#include <condition_variable>
+#include <cstddef>
+#include <exception>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <queue>
+#include <span>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <type_traits>
+#include <utility>
+#include <vector>
+#include <test/execution.hpp>
+#ifdef BEMAN_HAS_MODULES
+import beman.execution;
+import beman.execution.detail.schedule_result_t;
+#else
+#include <beman/execution/execution.hpp>
+#endif
+
+namespace {
+namespace replaceability = test_std::parallel_scheduler_replacement;
+
+struct proxy : replaceability::receiver_proxy {
+    auto set_value() noexcept -> void override {}
+    auto set_error(::std::exception_ptr) noexcept -> void override {}
+    auto set_stopped() noexcept -> void override {}
+};
+
+struct bulk_proxy : replaceability::bulk_item_receiver_proxy {
+    auto set_value() noexcept -> void override {}
+    auto set_error(::std::exception_ptr) noexcept -> void override {}
+    auto set_stopped() noexcept -> void override {}
+    auto execute(::std::size_t, ::std::size_t) noexcept -> void override {}
+};
+
+struct backend : replaceability::parallel_scheduler_backend {
+    auto schedule(replaceability::receiver_proxy&, ::std::span<::std::byte>) noexcept -> void override {}
+    auto schedule_bulk_chunked(::std::size_t,
+                               replaceability::bulk_item_receiver_proxy&,
+                               ::std::span<::std::byte>) noexcept -> void override {}
+    auto schedule_bulk_unchunked(::std::size_t,
+                                 replaceability::bulk_item_receiver_proxy&,
+                                 ::std::span<::std::byte>) noexcept -> void override {}
+};
+
+struct thread_pool_backend : replaceability::parallel_scheduler_backend {
+    struct task {
+        task() = default;
+
+        task(const task&) = delete;
+
+        task(task&&) = delete;
+
+        virtual ~task() = default;
+
+        auto operator=(const task&) -> task& = delete;
+
+        auto operator=(task&&) -> task& = delete;
+
+        virtual auto exec() noexcept -> void = 0;
+    };
+
+    struct schedule_task : task {
+        explicit schedule_task(replaceability::receiver_proxy& p) noexcept : proxy(p) {}
+
+        auto exec() noexcept -> void override { proxy.set_value(); }
+
+        replaceability::receiver_proxy& proxy;
+    };
+
+    struct bulk_shared_state {
+        std::atomic<std::size_t> counter;
+        std::exception_ptr       exception;
+    };
+
+    struct bulk_task : task {
+        bulk_task(std::shared_ptr<bulk_shared_state>        counter,
+                  replaceability::bulk_item_receiver_proxy& proxy,
+                  std::size_t                               i,
+                  std::size_t                               j) noexcept
+            : shared_state(std::move(counter)), proxy(proxy), i(i), j(j) {}
+
+        auto exec() noexcept -> void override {
+            proxy.execute(i, j);
+            if (shared_state->counter.fetch_sub(1uz, std::memory_order_acq_rel) == 1uz) {
+                if (shared_state->exception) {
+                    proxy.set_error(shared_state->exception);
+                } else {
+                    proxy.set_value();
+                }
+            }
+        }
+
+        std::shared_ptr<bulk_shared_state>        shared_state;
+        replaceability::bulk_item_receiver_proxy& proxy;
+        std::size_t                               i;
+        std::size_t                               j;
+    };
+
+    thread_pool_backend() {
+        for (auto& worker : workers) {
+            worker = std::thread{&thread_pool_backend::run, this};
+        }
+    }
+
+    thread_pool_backend(const thread_pool_backend&) = delete;
+
+    ~thread_pool_backend() override {
+        std::unique_lock guard{mtx};
+        stopped = true;
+        guard.unlock();
+        cv.notify_all();
+        for (auto& worker : workers) {
+            worker.join();
+        }
+    }
+
+    auto operator=(const thread_pool_backend&) = delete;
+
+    auto schedule(replaceability::receiver_proxy& proxy, ::std::span<::std::byte>) noexcept -> void override {
+        try {
+            auto             t = std::make_unique<schedule_task>(proxy);
+            std::unique_lock guard{mtx};
+            tasks.push(std::move(t));
+            guard.unlock();
+            cv.notify_one();
+        } catch (...) {
+            proxy.set_error(std::current_exception());
+        }
+    }
+
+    auto schedule_bulk(std::size_t                               shape,
+                       std::size_t                               chunk_length,
+                       replaceability::bulk_item_receiver_proxy& proxy,
+                       std::span<::std::byte>) noexcept -> void {
+        const std::size_t                  chunk_count = (shape + chunk_length - 1uz) / chunk_length;
+        std::shared_ptr<bulk_shared_state> shared_state;
+        try {
+            shared_state = std::make_shared<bulk_shared_state>(chunk_count);
+        } catch (...) {
+            proxy.set_error(std::current_exception());
+            return;
+        }
+        std::unique_lock guard{mtx};
+        for (std::size_t i = 0; i < chunk_count; ++i) {
+            try {
+                const std::size_t begin = i * chunk_length;
+                const std::size_t end   = std::min(begin + chunk_length, shape);
+                tasks.push(std::make_unique<bulk_task>(shared_state, proxy, begin, end));
+            } catch (...) {
+                const std::size_t n = chunk_count - i; // the count of `bulk_task` which are not enqueued successfully
+
+                guard.unlock();
+                if (n == 1uz) {
+                    cv.notify_one();
+                } else if (n > 1uz) {
+                    cv.notify_all();
+                }
+
+                // happens-before `proxy.set_value()/proxy.set_error(...)` in `bulk_task::exec`
+                shared_state->exception = std::current_exception();
+
+                if (shared_state->counter.fetch_sub(n, std::memory_order_acq_rel) == n) {
+                    proxy.set_error(shared_state->exception);
+                }
+                return;
+            }
+        }
+        guard.unlock();
+        cv.notify_all();
+    }
+
+    auto schedule_bulk_chunked(::std::size_t                             shape,
+                               replaceability::bulk_item_receiver_proxy& proxy,
+                               ::std::span<::std::byte>                  storage) noexcept -> void override {
+        const std::size_t chunk_length = (shape + num_threads - 1uz) / num_threads;
+        schedule_bulk(shape, chunk_length, proxy, storage);
+    }
+
+    auto schedule_bulk_unchunked(::std::size_t                             shape,
+                                 replaceability::bulk_item_receiver_proxy& proxy,
+                                 ::std::span<::std::byte>                  storage) noexcept -> void override {
+        schedule_bulk(shape, 1uz, proxy, storage);
+    }
+
+  private:
+    auto run() noexcept -> void {
+        while (true) {
+            std::unique_lock guard{mtx};
+            cv.wait(guard, [this]() noexcept { return !tasks.empty() || stopped; });
+            if (stopped) {
+                return;
+            }
+            auto task = std::move(tasks.front());
+            tasks.pop();
+            guard.unlock();
+            task->exec();
+        }
+    }
+
+    static constexpr std::size_t      num_threads = 4uz;
+    bool                              stopped     = false;
+    std::mutex                        mtx;
+    std::condition_variable           cv;
+    std::queue<std::unique_ptr<task>> tasks;
+    std::thread                       workers[num_threads];
+};
+
+auto test_parallel_scheduler_synopsis() -> void {
+    static_assert(!::std::default_initializable<test_std::parallel_scheduler>);
+    static_assert(::std::copy_constructible<test_std::parallel_scheduler>);
+    static_assert(::std::move_constructible<test_std::parallel_scheduler>);
+    static_assert(test_std::scheduler<test_std::parallel_scheduler>);
+
+    static_assert(::std::same_as<decltype(test_std::get_parallel_scheduler()), test_std::parallel_scheduler>);
+    static_assert(::std::same_as<test_std::schedule_result_t<test_std::parallel_scheduler>,
+                                 test_std::parallel_scheduler::sender>);
+    static_assert(test_std::sender<test_std::parallel_scheduler::sender>);
+    static_assert(::std::same_as<decltype(test_std::get_completion_signatures<test_std::parallel_scheduler::sender>()),
+                                 test_std::completion_signatures<test_std::set_value_t(),
+                                                                 test_std::set_error_t(::std::exception_ptr),
+                                                                 test_std::set_stopped_t()>>);
+
+    static_assert(
+        noexcept(test_std::get_forward_progress_guarantee(::std::declval<const test_std::parallel_scheduler&>())));
+    static_assert(::std::same_as<decltype(test_std::get_forward_progress_guarantee(
+                                     ::std::declval<const test_std::parallel_scheduler&>())),
+                                 test_std::forward_progress_guarantee>);
+}
+
+auto test_replaceability_synopsis() -> void {
+    static_assert(::std::is_abstract_v<replaceability::receiver_proxy>);
+    static_assert(::std::is_abstract_v<replaceability::bulk_item_receiver_proxy>);
+    static_assert(::std::is_abstract_v<replaceability::parallel_scheduler_backend>);
+    static_assert(::std::derived_from<bulk_proxy, replaceability::receiver_proxy>);
+    static_assert(::std::derived_from<backend, replaceability::parallel_scheduler_backend>);
+    static_assert(::std::same_as<decltype(::std::declval<proxy&>().template try_query<int>(0)), ::std::optional<int>>);
+    static_assert(::std::same_as<decltype(replaceability::query_parallel_scheduler_backend()),
+                                 ::std::shared_ptr<replaceability::parallel_scheduler_backend>>);
+}
+
+auto test_parallel_scheduler_schedule() -> void {
+    auto sch = test_std::get_parallel_scheduler();
+    {
+        int i = 0;
+        test_std::sync_wait(test_std::schedule(sch) | test_std::then([&i]() noexcept { i = 114514; }));
+        ASSERT(i == 114514);
+    }
+    {
+        for (auto size : {1uz, 4uz, 8uz, 16uz, 32uz}) {
+            std::vector<int> vec(size);
+            std::iota(vec.begin(), vec.end(), 0);
+
+            test_std::sync_wait(
+                test_std::schedule(sch) |
+                test_std::bulk(test_std::par, vec.size(), [&vec](std::size_t i) noexcept { vec[i] = 2 * vec[i]; }));
+            for (std::size_t i = 0; i < vec.size(); ++i) {
+                ASSERT(vec[i] == 2 * static_cast<int>(i));
+            }
+
+            test_std::sync_wait(
+                test_std::schedule(sch) |
+                test_std::bulk(test_std::seq, vec.size(), [&vec](std::size_t i) noexcept { ++vec[i]; }));
+            for (std::size_t i = 0; i < vec.size(); ++i) {
+                ASSERT(vec[i] == 2 * static_cast<int>(i) + 1);
+            }
+        }
+    }
+}
+} // namespace
+
+namespace beman::execution::parallel_scheduler_replacement {
+auto query_parallel_scheduler_backend() -> std::shared_ptr<parallel_scheduler_backend> {
+    static auto backend = std::make_shared<::thread_pool_backend>();
+    return backend;
+}
+} // namespace beman::execution::parallel_scheduler_replacement
+
+TEST(exec_parallel_scheduler) {
+    test_parallel_scheduler_synopsis();
+    test_replaceability_synopsis();
+    test_parallel_scheduler_schedule();
+}
diff --git a/tests/beman/execution/exec-then.test.cpp b/tests/beman/execution/exec-then.test.cpp
index ec0abef7..101fde6a 100644
--- a/tests/beman/execution/exec-then.test.cpp
+++ b/tests/beman/execution/exec-then.test.cpp
@@ -58,6 +58,8 @@ struct sender {
     }
 };
 
+constexpr auto consume_all = [](auto&&...) {};
+
 template <bool Expect>
 auto test_has(auto cpo, auto in_sender, auto fun) -> void {
     static_assert(test_std::receiver<receiver>);
@@ -70,14 +72,9 @@ auto test_has(auto cpo, auto in_sender, auto fun) -> void {
         static_assert(requires {
             { in_sender | cpo(fun) } -> test_std::sender;
         });
-#ifndef _MSC_VER
-        //-dk:TODO re-enable this test
         static_assert(requires {
-            {
-                in_sender | cpo(fun) | cpo([](auto&&...) {})
-            } -> test_std::sender;
+            { in_sender | cpo(fun) | cpo(consume_all) } -> test_std::sender;
         });
-#endif
         auto sender{cpo(in_sender, fun)};
         auto op{test_std::connect(::std::move(sender), receiver{})};
         test_std::start(op);