diff --git a/libs/core/executors/CMakeLists.txt b/libs/core/executors/CMakeLists.txt
index 8deb1494338..9149359ac1c 100644
--- a/libs/core/executors/CMakeLists.txt
+++ b/libs/core/executors/CMakeLists.txt
@@ -26,6 +26,8 @@ set(executors_headers
     hpx/executors/execution_policy_parameters.hpp
     hpx/executors/execution_policy_scheduling_property.hpp
     hpx/executors/execution_policy.hpp
+    hpx/executors/executor_scheduler.hpp
+    hpx/executors/executor_scheduler_bulk.hpp
     hpx/executors/explicit_scheduler_executor.hpp
     hpx/executors/fork_join_executor.hpp
     hpx/executors/limiting_executor.hpp
diff --git a/libs/core/executors/include/hpx/executors/executor_scheduler.hpp b/libs/core/executors/include/hpx/executors/executor_scheduler.hpp
new file mode 100644
index 00000000000..6949b4d2777
--- /dev/null
+++ b/libs/core/executors/include/hpx/executors/executor_scheduler.hpp
@@ -0,0 +1,148 @@
+//  Copyright (c) 2026 The STE||AR-Group
+//
+//  SPDX-License-Identifier: BSL-1.0
+//  Distributed under the Boost Software License, Version 1.0. (See accompanying
+//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+/// \file parallel/executors/executor_scheduler.hpp
+
+#pragma once
+
+#include <hpx/config.hpp>
+#include <hpx/modules/errors.hpp>
+#include <hpx/modules/execution.hpp>
+#include <hpx/modules/execution_base.hpp>
+
+#include <exception>
+#include <type_traits>
+#include <utility>
+
+namespace hpx::execution::experimental {
+
+    // Forward declarations
+    template <typename Executor>
+    struct executor_scheduler;
+
+    template <typename Executor>
+    struct executor_sender;
+
+    ///////////////////////////////////////////////////////////////////////////
+    template <typename Executor, typename Receiver>
+    struct executor_operation_state
+    {
+        HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> exec_;
+        HPX_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver_;
+
+        template <typename Exec, typename Recv>
+        executor_operation_state(Exec&& exec, Recv&& recv)
+          : exec_(HPX_FORWARD(Exec, exec))
+          , receiver_(HPX_FORWARD(Recv, recv))
+        {
+        }
+
+        executor_operation_state(executor_operation_state&&) = delete;
+        executor_operation_state(executor_operation_state const&) = delete;
+        executor_operation_state& operator=(
+            executor_operation_state&&) = delete;
+        executor_operation_state& operator=(
+            executor_operation_state const&) = delete;
+
+        ~executor_operation_state() = default;
+
+        friend void tag_invoke(start_t, executor_operation_state& os) noexcept
+        {
+            hpx::detail::try_catch_exception_ptr(
+                [&]() {
+                    hpx::parallel::execution::post(os.exec_,
+                        [receiver = HPX_MOVE(os.receiver_)]() mutable {
+                            hpx::execution::experimental::set_value(
+                                HPX_MOVE(receiver));
+                        });
+                },
+                [&](std::exception_ptr ep) {
+                    hpx::execution::experimental::set_error(
+                        HPX_MOVE(os.receiver_), HPX_MOVE(ep));
+                });
+        }
+    };
+
+    ///////////////////////////////////////////////////////////////////////////
+    template <typename Executor>
+    struct executor_sender
+    {
+        HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> exec_;
+
+        using completion_signatures =
+            hpx::execution::experimental::completion_signatures<
+                hpx::execution::experimental::set_value_t(),
+                hpx::execution::experimental::set_error_t(std::exception_ptr)>;
+
+        template <typename Env>
+        friend auto tag_invoke(
+            hpx::execution::experimental::get_completion_signatures_t,
+            executor_sender const&, Env) noexcept -> completion_signatures;
+
+        friend constexpr auto tag_invoke(
+            hpx::execution::experimental::get_completion_scheduler_t<
+                hpx::execution::experimental::set_value_t>,
+            executor_sender const& s) noexcept
+        {
+            return executor_scheduler<Executor>{s.exec_};
+        }
+
+        template <typename Receiver>
+        friend executor_operation_state<Executor, Receiver> tag_invoke(
+            connect_t, executor_sender&& s, Receiver&& receiver)
+        {
+            return {HPX_MOVE(s.exec_), HPX_FORWARD(Receiver, receiver)};
+        }
+
+        template <typename Receiver>
+        friend executor_operation_state<Executor, Receiver> tag_invoke(
+            connect_t, executor_sender const& s, Receiver&& receiver)
+        {
+            return {s.exec_, HPX_FORWARD(Receiver, receiver)};
+        }
+    };
+
+    ///////////////////////////////////////////////////////////////////////////
+    template <typename Executor>
+    struct executor_scheduler
+    {
+        using executor_type = std::decay_t<Executor>;
+
+        HPX_NO_UNIQUE_ADDRESS executor_type exec_;
+
+        constexpr executor_scheduler() = default;
+
+        template <typename Exec,
+            typename = std::enable_if_t<
+                !std::is_same_v<std::decay_t<Exec>, executor_scheduler>>>
+        explicit executor_scheduler(Exec&& exec)
+          : exec_(HPX_FORWARD(Exec, exec))
+        {
+        }
+
+        constexpr bool operator==(executor_scheduler const& rhs) const noexcept
+        {
+            return exec_ == rhs.exec_;
+        }
+
+        constexpr bool operator!=(executor_scheduler const& rhs) const noexcept
+        {
+            return !(*this == rhs);
+        }
+
+        friend executor_sender<Executor> tag_invoke(
+            schedule_t, executor_scheduler&& sched)
+        {
+            return {HPX_MOVE(sched.exec_)};
+        }
+
+        friend executor_sender<Executor> tag_invoke(
+            schedule_t, executor_scheduler const& sched)
+        {
+            return {sched.exec_};
+        }
+    };
+}    // namespace hpx::execution::experimental
diff --git a/libs/core/executors/include/hpx/executors/executor_scheduler_bulk.hpp b/libs/core/executors/include/hpx/executors/executor_scheduler_bulk.hpp
new file mode 100644
index 00000000000..52ce7a62e4e
--- /dev/null
+++ b/libs/core/executors/include/hpx/executors/executor_scheduler_bulk.hpp
@@ -0,0 +1,222 @@
+//  Copyright (c) 2026 The STE||AR-Group
+//
+//  SPDX-License-Identifier: BSL-1.0
+//  Distributed under the Boost Software License, Version 1.0. (See accompanying
+//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+#pragma once
+
+#include <hpx/config.hpp>
+
+#if defined(HPX_HAVE_STDEXEC)
+#include <hpx/execution_base/stdexec_forward.hpp>
+#endif
+
+#include <hpx/execution/algorithms/bulk.hpp>
+#include <hpx/execution_base/completion_scheduler.hpp>
+#include <hpx/execution_base/completion_signatures.hpp>
+#include <hpx/execution_base/execution.hpp>
+#include <hpx/execution_base/receiver.hpp>
+#include <hpx/execution_base/sender.hpp>
+#include <hpx/executors/executor_scheduler.hpp>
+#include <hpx/modules/errors.hpp>
+
+#include <exception>
+#include <type_traits>
+#include <utility>
+
+namespace hpx::execution::experimental {
+
+    namespace detail {
+        template <typename Executor, typename Receiver, typename Shape,
+            typename F>
+        struct executor_bulk_receiver
+        {
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> exec_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Shape> shape_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<F> f_;
+
+            template <typename Error>
+            friend void tag_invoke(
+                set_error_t, executor_bulk_receiver&& r, Error&& error) noexcept
+            {
+                hpx::execution::experimental::set_error(
+                    HPX_MOVE(r.receiver_), HPX_FORWARD(Error, error));
+            }
+
+            friend void tag_invoke(
+                set_stopped_t, executor_bulk_receiver&& r) noexcept
+            {
+                hpx::execution::experimental::set_stopped(
+                    HPX_MOVE(r.receiver_));
+            }
+
+            template <typename... Ts>
+            friend void tag_invoke(
+                set_value_t, executor_bulk_receiver&& r, Ts&&... ts) noexcept
+            {
+                hpx::detail::try_catch_exception_ptr(
+                    [&]() {
+                        hpx::parallel::execution::bulk_sync_execute(
+                            r.exec_, r.f_, r.shape_, ts...);
+
+                        hpx::execution::experimental::set_value(
+                            HPX_MOVE(r.receiver_), HPX_FORWARD(Ts, ts)...);
+                    },
+                    [&](std::exception_ptr ep) {
+                        hpx::execution::experimental::set_error(
+                            HPX_MOVE(r.receiver_), HPX_MOVE(ep));
+                    });
+            }
+        };
+
+        template <typename Executor, typename Sender, typename Shape,
+            typename F>
+        struct executor_bulk_sender
+        {
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> exec_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Sender> sender_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<Shape> shape_;
+            HPX_NO_UNIQUE_ADDRESS std::decay_t<F> f_;
+
+#if defined(HPX_HAVE_STDEXEC)
+            using sender_concept = hpx::execution::experimental::sender_t;
+
+            template <typename Env>
+            friend auto tag_invoke(
+                hpx::execution::experimental::get_completion_signatures_t,
+                executor_bulk_sender const&, Env const&) -> hpx::execution::
+                experimental::transform_completion_signatures_of<Sender, Env,
+                    hpx::execution::experimental::completion_signatures<
+                        hpx::execution::experimental::set_error_t(
+                            std::exception_ptr)>>;
+
+            struct env
+            {
+                std::decay_t<Sender> const& pred_snd;
+                std::decay_t<Executor> const& exec;
+
+                template <typename CPO>
+                    requires(
+                        meta::value<meta::one_of<CPO,
+                            hpx::execution::experimental::set_error_t,
+                            hpx::execution::experimental::set_stopped_t>> &&
+                        hpx::execution::experimental::detail::
+                            has_completion_scheduler_v<CPO,
+                                std::decay_t<Sender>>)
+                friend constexpr auto tag_invoke(
+                    hpx::execution::experimental::get_completion_scheduler_t<
+                        CPO>
+                        tag,
+                    env const& e) noexcept
+                {
+                    return tag(
+                        hpx::execution::experimental::get_env(e.pred_snd));
+                }
+
+                friend constexpr auto tag_invoke(
+                    hpx::execution::experimental::get_completion_scheduler_t<
+                        hpx::execution::experimental::set_value_t>,
+                    env const& e) noexcept
+                {
+                    return hpx::execution::experimental::executor_scheduler<
+                        Executor>{e.exec};
+                }
+            };
+
+            friend constexpr auto tag_invoke(
+                hpx::execution::experimental::get_env_t,
+                executor_bulk_sender const& s) noexcept
+            {
+                return env{s.sender_, s.exec_};
+            }
+#else
+            template <typename Env>
+            struct generate_completion_signatures
+            {
+                template <template <typename...> typename Tuple,
+                    template <typename...> typename Variant>
+                using value_types =
+                    value_types_of_t<Sender, Env, Tuple, Variant>;
+
+                template <template <typename...> typename Variant>
+                using error_types = hpx::util::detail::unique_concat_t<
+                    error_types_of_t<Sender, Env, Variant>,
+                    Variant<std::exception_ptr>>;
+
+                static constexpr bool sends_stopped =
+                    sends_stopped_of_v<Sender, Env>;
+            };
+
+            template <typename Env>
+            friend auto tag_invoke(
+                hpx::execution::experimental::get_completion_signatures_t,
+                executor_bulk_sender const&, Env)
+                -> generate_completion_signatures<Env>;
+
+            template <typename CPO>
+            friend constexpr auto tag_invoke(
+                hpx::execution::experimental::get_completion_scheduler_t<CPO>
+                    tag,
+                executor_bulk_sender const& s)
+            {
+                return tag(s.sender_);
+            }
+
+            friend constexpr auto tag_invoke(
+                hpx::execution::experimental::get_completion_scheduler_t<
+                    hpx::execution::experimental::set_value_t>,
+                executor_bulk_sender const& s)
+            {
+                return hpx::execution::experimental::executor_scheduler<
+                    Executor>{s.exec_};
+            }
+#endif
+
+            template <typename Receiver>
+            friend auto tag_invoke(connect_t,
+                executor_bulk_sender<Executor, Sender, Shape, F>&& s,
+                Receiver&& receiver)
+            {
+                return hpx::execution::experimental::connect(
+                    HPX_MOVE(s.sender_),
+                    executor_bulk_receiver<Executor, std::decay_t<Receiver>,
+                        Shape, F>{HPX_MOVE(s.exec_),
+                        HPX_FORWARD(Receiver, receiver), HPX_MOVE(s.shape_),
+                        HPX_MOVE(s.f_)});
+            }
+
+            template <typename Receiver>
+            friend auto tag_invoke(connect_t,
+                executor_bulk_sender<Executor, Sender, Shape, F>& s,
+                Receiver&& receiver)
+            {
+                return hpx::execution::experimental::connect(s.sender_,
+                    executor_bulk_receiver<Executor, std::decay_t<Receiver>,
+                        Shape, F>{s.exec_, HPX_FORWARD(Receiver, receiver),
+                        s.shape_, s.f_});
+            }
+
+            template <typename Receiver>
+            friend auto tag_invoke(connect_t,
+                executor_bulk_sender<Executor, Sender, Shape, F> const& s,
+                Receiver&& receiver)
+            {
+                return hpx::execution::experimental::connect(s.sender_,
+                    executor_bulk_receiver<Executor, std::decay_t<Receiver>,
+                        Shape, F>{s.exec_, HPX_FORWARD(Receiver, receiver),
+                        s.shape_, s.f_});
+            }
+        };
+    }    // namespace detail
+
+    template <typename Executor, typename Sender, typename Shape, typename F>
+    auto tag_invoke(bulk_t, executor_scheduler<Executor> const& sched,
+        Sender&& sender, Shape const& shape, F&& f)
+    {
+        return detail::executor_bulk_sender<Executor, std::decay_t<Sender>,
+            Shape, std::decay_t<F>>{
+            sched.exec_, HPX_FORWARD(Sender, sender), shape, HPX_FORWARD(F, f)};
+    }
+}    // namespace hpx::execution::experimental
diff --git a/libs/core/executors/include/hpx/executors/parallel_executor.hpp b/libs/core/executors/include/hpx/executors/parallel_executor.hpp
index be279bc9865..c2a972a32b2 100644
--- a/libs/core/executors/include/hpx/executors/parallel_executor.hpp
+++ b/libs/core/executors/include/hpx/executors/parallel_executor.hpp
@@ -10,9 +10,12 @@
 
 #include <hpx/config.hpp>
 #include <hpx/assert.hpp>
+#include <hpx/execution/queries/get_scheduler.hpp>
 #include <hpx/executors/detail/hierarchical_spawning.hpp>
 #include <hpx/executors/detail/index_queue_spawning.hpp>
 #include <hpx/executors/execution_policy_mappings.hpp>
+#include <hpx/executors/executor_scheduler.hpp>
+#include <hpx/executors/executor_scheduler_bulk.hpp>
 #include <hpx/modules/allocator_support.hpp>
 #include <hpx/modules/async_base.hpp>
 #include <hpx/modules/concepts.hpp>
@@ -474,9 +477,7 @@ namespace hpx::execution {
         parallel_policy_executor& operator=(
             parallel_policy_executor&&) = default;
 
-#if defined(__NVCC__) || defined(__CUDACC__)
-        constexpr ~parallel_policy_executor() {}
-#endif
+        ~parallel_policy_executor() = default;
 
     private:
         // property implementations
@@ -687,6 +688,15 @@ namespace hpx::execution {
 #endif
             }
         }
+
+        friend hpx::execution::experimental::executor_scheduler<
+            parallel_policy_executor>
+        tag_invoke(hpx::execution::experimental::get_scheduler_t,
+            parallel_policy_executor const& exec)
+        {
+            return hpx::execution::experimental::executor_scheduler<
+                parallel_policy_executor>(exec);
+        }
         /// \endcond
 
     public:
@@ -1015,6 +1025,15 @@ namespace hpx::execution {
 #endif
             }
         }
+
+        friend hpx::execution::experimental::executor_scheduler<
+            parallel_policy_executor>
+        tag_invoke(hpx::execution::experimental::get_scheduler_t,
+            parallel_policy_executor const& exec)
+        {
+            return hpx::execution::experimental::executor_scheduler<
+                parallel_policy_executor>(exec);
+        }
         /// \endcond
 
     public:
diff --git a/libs/core/executors/include/hpx/executors/sequenced_executor.hpp b/libs/core/executors/include/hpx/executors/sequenced_executor.hpp
index 6d61e286ba3..3a41e5fb24d 100644
--- a/libs/core/executors/include/hpx/executors/sequenced_executor.hpp
+++ b/libs/core/executors/include/hpx/executors/sequenced_executor.hpp
@@ -10,6 +10,8 @@
 
 #include <hpx/config.hpp>
 #include <hpx/executors/execution_policy_mappings.hpp>
+#include <hpx/executors/executor_scheduler.hpp>
+#include <hpx/executors/executor_scheduler_bulk.hpp>
 #include <hpx/executors/parallel_executor.hpp>
 #include <hpx/modules/errors.hpp>
 #include <hpx/modules/execution.hpp>
@@ -236,6 +238,15 @@ namespace hpx::execution {
 #endif
         }
 
+        friend hpx::execution::experimental::executor_scheduler<
+            sequenced_executor>
+        tag_invoke(hpx::execution::experimental::get_scheduler_t,
+            sequenced_executor const& exec)
+        {
+            return hpx::execution::experimental::executor_scheduler<
+                sequenced_executor>(exec);
+        }
+
     private:
         friend class hpx::serialization::access;
 
diff --git a/libs/core/executors/tests/unit/CMakeLists.txt b/libs/core/executors/tests/unit/CMakeLists.txt
index 7248e7401a8..f0bb1798d14 100644
--- a/libs/core/executors/tests/unit/CMakeLists.txt
+++ b/libs/core/executors/tests/unit/CMakeLists.txt
@@ -9,6 +9,7 @@ set(tests
     annotation_property
     created_executor
     execution_policy_mappings
+    executor_algorithm_bulk
     explicit_scheduler_executor
     fork_join_executor
     fork_join_executor_from
diff --git a/libs/core/executors/tests/unit/executor_algorithm_bulk.cpp b/libs/core/executors/tests/unit/executor_algorithm_bulk.cpp
new file mode 100644
index 00000000000..0aa1b4ff526
--- /dev/null
+++ b/libs/core/executors/tests/unit/executor_algorithm_bulk.cpp
@@ -0,0 +1,94 @@
+//  Copyright (c) 2026 The STE||AR-Group
+//
+//  SPDX-License-Identifier: BSL-1.0
+//  Distributed under the Boost Software License, Version 1.0. (See accompanying
+//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+#include <hpx/execution.hpp>
+#include <hpx/init.hpp>
+#include <hpx/modules/testing.hpp>
+
+#include <atomic>
+#include <cstdint>
+#include <iostream>
+#include <string>
+#include <vector>
+
+namespace ex = hpx::execution::experimental;
+
+void test_sequential_bulk()
+{
+    hpx::execution::sequenced_executor exec;
+    auto sched =
+        ex::executor_scheduler<hpx::execution::sequenced_executor>{exec};
+
+    std::atomic<int> call_count{0};
+
+    auto snd = ex::schedule(sched) | ex::bulk(10, [&](int i) {
+        (void) i;
+        ++call_count;
+        // Should run sequentially, so it's safe
+    });
+
+    hpx::this_thread::experimental::sync_wait(snd);
+
+    HPX_TEST_EQ(call_count.load(), 10);
+}
+
+void test_parallel_bulk()
+{
+    hpx::execution::parallel_executor exec;
+    auto sched =
+        ex::executor_scheduler<hpx::execution::parallel_executor>{exec};
+
+    std::atomic<int> call_count{0};
+
+    auto snd = ex::schedule(sched) | ex::bulk(1000, [&](int i) {
+        (void) i;
+        ++call_count;
+    });
+
+    hpx::this_thread::experimental::sync_wait(snd);
+
+    HPX_TEST_EQ(call_count.load(), 1000);
+}
+
+void test_parallel_bulk_with_value()
+{
+    hpx::execution::parallel_executor exec;
+    auto sched =
+        ex::executor_scheduler<hpx::execution::parallel_executor>{exec};
+
+    std::atomic<int> call_count{0};
+
+    auto snd = ex::schedule(sched) | ex::then([]() { return 42; }) |
+        ex::bulk(100, [&](int i, int val) {
+            (void) i;
+            HPX_TEST_EQ(val, 42);
+            ++call_count;
+        });
+
+    hpx::this_thread::experimental::sync_wait(snd);
+
+    HPX_TEST_EQ(call_count.load(), 100);
+}
+
+int hpx_main()
+{
+    test_sequential_bulk();
+    test_parallel_bulk();
+    test_parallel_bulk_with_value();
+    return hpx::local::finalize();
+}
+
+int main(int argc, char* argv[])
+{
+    std::vector<std::string> const cfg = {"hpx.os_threads=all"};
+    hpx::local::init_params init_args;
+    init_args.cfg = cfg;
+
+    HPX_TEST_EQ_MSG(hpx::local::init(hpx_main, argc, argv, init_args), 0,
+        "HPX main exited with non-zero status");
+
+    return hpx::util::report_errors();
+}