diff --git a/libs/pika/execution/CMakeLists.txt b/libs/pika/execution/CMakeLists.txt
index abe425a3ab..f5dfe31dc7 100644
--- a/libs/pika/execution/CMakeLists.txt
+++ b/libs/pika/execution/CMakeLists.txt
@@ -17,6 +17,7 @@ set(execution_headers
     pika/execution/algorithms/just.hpp
     pika/execution/algorithms/let_error.hpp
     pika/execution/algorithms/let_value.hpp
+    pika/execution/algorithms/require_started.hpp
     pika/execution/algorithms/schedule_from.hpp
     pika/execution/algorithms/split.hpp
     pika/execution/algorithms/split_tuple.hpp
diff --git a/libs/pika/execution/include/pika/execution/algorithms/require_started.hpp b/libs/pika/execution/include/pika/execution/algorithms/require_started.hpp
new file mode 100644
index 0000000000..86545d2041
--- /dev/null
+++ b/libs/pika/execution/include/pika/execution/algorithms/require_started.hpp
@@ -0,0 +1,448 @@
+//  Copyright (c) 2023 ETH Zurich
+//
+//  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 <pika/config.hpp>
+
+#include <pika/assert.hpp>
+#include <pika/concepts/concepts.hpp>
+#include <pika/errors/error_code.hpp>
+#include <pika/execution/algorithms/detail/partial_algorithm.hpp>
+#include <pika/execution_base/operation_state.hpp>
+#include <pika/execution_base/receiver.hpp>
+#include <pika/execution_base/sender.hpp>
+#include <pika/functional/bind_front.hpp>
+#include <pika/functional/detail/tag_fallback_invoke.hpp>
+#include <pika/type_support/detail/with_result_of.hpp>
+#include <pika/type_support/pack.hpp>
+
+#include <fmt/core.h>
+#include <fmt/ostream.h>
+#include <fmt/printf.h>
+
+#include <exception>
+#include <iostream>
+#include <optional>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+
+namespace pika {
+#if !defined(PIKA_HAVE_STDEXEC)
+    namespace execution::experimental {
+        // We only make the choice of mode available when not using stdexec. stdexec's sender
+        // concepts require nothrow destructibility, which is not satisfied by throw_on_unstarted.
+        // With stdexec enabled, an unstarted sender will always terminate.
+        enum class require_started_mode
+        {
+            terminate_on_unstarted,
+            throw_on_unstarted
+        };
+    }    // namespace execution::experimental
+#endif
+
+    namespace require_started_detail {
+#if defined(PIKA_HAVE_STDEXEC)
+# define PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(f, message)                           \
+  fmt::print(std::cerr, "{}: {}\n", f, message);                                                   \
+  std::terminate();
+#else
+        using pika::execution::experimental::require_started_mode;
+
+# define PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(mode, f, message)                     \
+  {                                                                                                \
+   switch (mode)                                                                                   \
+   {                                                                                               \
+   case require_started_mode::terminate_on_unstarted:                                              \
+    fmt::print(std::cerr, "{}: {}\n", f, message);                                                 \
+    std::terminate();                                                                              \
+    break;                                                                                         \
+                                                                                                   \
+   case require_started_mode::throw_on_unstarted:                                                  \
+    PIKA_THROW_EXCEPTION(pika::error::invalid_status, f, fmt::runtime(message));                   \
+    break;                                                                                         \
+   }                                                                                               \
+  }
+#endif
+
+        template <typename OpState>
+        struct require_started_receiver_impl
+        {
+            struct require_started_receiver_type;
+        };
+
+        template <typename OpState>
+        using require_started_receiver =
+            typename require_started_receiver_impl<OpState>::require_started_receiver_type;
+
+        template <typename OpState>
+        struct require_started_receiver_impl<OpState>::require_started_receiver_type
+        {
+            using is_receiver = void;
+
+            OpState* op_state = nullptr;
+
+            template <typename Error>
+            friend void tag_invoke(pika::execution::experimental::set_error_t,
+                require_started_receiver_type r, Error&& error) noexcept
+            {
+                PIKA_ASSERT(r.op_state != nullptr);
+                pika::execution::experimental::set_error(
+                    PIKA_MOVE(r.op_state->receiver), PIKA_FORWARD(Error, error));
+            }
+
+            friend void tag_invoke(pika::execution::experimental::set_stopped_t,
+                require_started_receiver_type r) noexcept
+            {
+                PIKA_ASSERT(r.op_state != nullptr);
+                pika::execution::experimental::set_stopped(PIKA_MOVE(r.op_state->receiver));
+            };
+
+            template <typename... Ts>
+            friend void tag_invoke(pika::execution::experimental::set_value_t,
+                require_started_receiver_type r, Ts&&... ts) noexcept
+            {
+                PIKA_ASSERT(r.op_state != nullptr);
+                pika::execution::experimental::set_value(
+                    PIKA_MOVE(r.op_state->receiver), PIKA_FORWARD(Ts, ts)...);
+            }
+
+            friend constexpr pika::execution::experimental::empty_env tag_invoke(
+                pika::execution::experimental::get_env_t,
+                require_started_receiver_type const&) noexcept
+            {
+                return {};
+            }
+        };
+
+        template <typename Sender, typename Receiver>
+        struct require_started_op_state_impl
+        {
+            struct require_started_op_state_type;
+        };
+
+        template <typename Sender, typename Receiver>
+        using require_started_op_state =
+            typename require_started_op_state_impl<Sender, Receiver>::require_started_op_state_type;
+
+        template <typename Sender, typename Receiver>
+        struct require_started_op_state_impl<Sender, Receiver>::require_started_op_state_type
+        {
+            using operation_state_type = pika::execution::experimental::connect_result_t<Sender,
+                require_started_receiver<require_started_op_state_type>>;
+
+            PIKA_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver;
+            std::optional<operation_state_type> op_state{std::nullopt};
+#if !defined(PIKA_HAVE_STDEXEC)
+            require_started_mode mode{require_started_mode::terminate_on_unstarted};
+#endif
+            bool started{false};
+
+            template <typename Receiver_>
+            require_started_op_state_type(std::decay_t<Sender> sender, Receiver_&& receiver
+#if !defined(PIKA_HAVE_STDEXEC)
+                ,
+                require_started_mode mode
+#endif
+                )
+              : receiver(PIKA_FORWARD(Receiver_, receiver))
+              , op_state(pika::detail::with_result_of([&]() {
+                  return pika::execution::experimental::connect(PIKA_MOVE(sender),
+                      require_started_receiver<require_started_op_state_type>{this});
+              }))
+#if !defined(PIKA_HAVE_STDEXEC)
+              , mode(mode)
+#endif
+            {
+            }
+
+            ~require_started_op_state_type()
+#if !defined(PIKA_HAVE_STDEXEC)
+                noexcept(false)
+#endif
+            {
+                if (!started)
+                {
+                    op_state.reset();
+
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::~require_started_operation_state",
+                        "The operation state of a require_started sender was never started");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(mode,
+                        "pika::execution::experimental::~require_started_operation_state",
+                        "The operation state of a require_started sender was never started");
+#endif
+                }
+            }
+            require_started_op_state_type(require_started_op_state_type&) = delete;
+            require_started_op_state_type& operator=(require_started_op_state_type&) = delete;
+            require_started_op_state_type(require_started_op_state_type const&) = delete;
+            require_started_op_state_type& operator=(require_started_op_state_type const&) = delete;
+
+            friend void tag_invoke(
+                pika::execution::experimental::start_t, require_started_op_state_type& os) noexcept
+            {
+                PIKA_ASSERT(os.op_state.has_value());
+
+                os.started = true;
+                pika::execution::experimental::start(os.op_state.value());
+            }
+        };
+
+        template <typename Sender>
+        struct require_started_sender_impl
+        {
+            struct require_started_sender_type;
+        };
+
+        template <typename Sender>
+        using require_started_sender =
+            typename require_started_sender_impl<Sender>::require_started_sender_type;
+
+        template <typename Sender>
+        struct require_started_sender_impl<Sender>::require_started_sender_type
+        {
+            using is_sender = void;
+
+            std::optional<std::decay_t<Sender>> sender{std::nullopt};
+#if !defined(PIKA_HAVE_STDEXEC)
+            require_started_mode mode{require_started_mode::terminate_on_unstarted};
+#endif
+            mutable bool connected{false};
+
+#if defined(PIKA_HAVE_STDEXEC)
+            using completion_signatures =
+                pika::execution::experimental::make_completion_signatures<std::decay_t<Sender>,
+                    pika::execution::experimental::empty_env>;
+#else
+            template <template <typename...> class Tuple, template <typename...> class Variant>
+            using value_types = typename pika::execution::experimental::sender_traits<
+                Sender>::template value_types<Tuple, Variant>;
+
+            template <template <typename...> class Variant>
+            using error_types = typename pika::execution::experimental::sender_traits<
+                Sender>::template error_types<Variant>;
+
+            static constexpr bool sends_done = false;
+#endif
+
+            template <typename Sender_,
+                typename Enable = std::enable_if_t<
+                    !std::is_same_v<std::decay_t<Sender_>, require_started_sender_type>>>
+            explicit require_started_sender_type(Sender_&& sender
+#if !defined(PIKA_HAVE_STDEXEC)
+                ,
+                require_started_mode mode = require_started_mode::terminate_on_unstarted
+#endif
+                )
+              : sender(PIKA_FORWARD(Sender_, sender))
+#if !defined(PIKA_HAVE_STDEXEC)
+              , mode(mode)
+#endif
+            {
+            }
+
+            require_started_sender_type(require_started_sender_type&& other) noexcept
+              : sender(std::exchange(other.sender, std::nullopt))
+#if !defined(PIKA_HAVE_STDEXEC)
+              , mode(other.mode)
+#endif
+              , connected(other.connected)
+            {
+            }
+
+            require_started_sender_type& operator=(require_started_sender_type&& other)
+#if defined(PIKA_HAVE_STDEXEC)
+                noexcept
+#else
+                noexcept(false)
+#endif
+            {
+                if (sender.has_value() && !connected)
+                {
+                    sender.reset();
+
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::require_started_sender::operator=(require_"
+                        "started_sender&&)",
+                        "Assigning to a require_started sender that was never started, the target "
+                        "would be discarded");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(mode,
+                        "pika::execution::experimental::require_started_sender::operator=(require_"
+                        "started_sender&&)",
+                        "Assigning to a require_started sender that was never started, the target "
+                        "would be discarded");
+#endif
+                }
+
+                sender = std::exchange(other.sender, std::nullopt);
+#if !defined(PIKA_HAVE_STDEXEC)
+                mode = other.mode;
+#endif
+                connected = other.connected;
+
+                return *this;
+            }
+
+            require_started_sender_type(require_started_sender_type const& other)
+              : sender(other.sender)
+#if !defined(PIKA_HAVE_STDEXEC)
+              , mode(other.mode)
+#endif
+              , connected(false)
+            {
+            }
+
+            require_started_sender_type& operator=(require_started_sender_type const& other)
+            {
+                if (sender.has_value() && !connected)
+                {
+                    sender.reset();
+
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::require_started_sender::operator=(require_"
+                        "started_sender const&)",
+                        "Assigning to a require_started sender that was never started, the target "
+                        "would be discarded");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(mode,
+                        "pika::execution::experimental::require_started_sender::operator=(require_"
+                        "started_sender const&)",
+                        "Assigning to a require_started sender that was never started, the target "
+                        "would be discarded");
+#endif
+                }
+
+                sender = other.sender;
+#if !defined(PIKA_HAVE_STDEXEC)
+                mode = other.mode;
+#endif
+                connected = false;
+
+                return *this;
+            }
+
+            ~require_started_sender_type()
+#if !defined(PIKA_HAVE_STDEXEC)
+                noexcept(false)
+#endif
+            {
+                if (sender.has_value() && !connected)
+                {
+                    sender.reset();
+
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::~require_started_sender",
+                        "A require_started sender was never started");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(mode,
+                        "pika::execution::experimental::~require_started_sender",
+                        "A require_started sender was never started");
+#endif
+                }
+            }
+
+            template <typename Receiver>
+            friend require_started_op_state<Sender, Receiver>
+            tag_invoke(pika::execution::experimental::connect_t, require_started_sender_type&& s,
+                Receiver&& receiver)
+            {
+                if (!s.sender.has_value())
+                {
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::connect(require_started_sender&&)",
+                        "Trying to connect an empty require_started sender");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(s.mode,
+                        "pika::execution::experimental::connect(require_started_sender&&)",
+                        "Trying to connect an empty require_started sender");
+#endif
+                }
+
+                s.connected = true;
+                return
+                {
+                    std::exchange(s.sender, std::nullopt).value(), PIKA_FORWARD(Receiver, receiver)
+#if !defined(PIKA_HAVE_STDEXEC)
+                                                                       ,
+                        s.mode
+#endif
+                };
+            }
+
+            template <typename Receiver>
+            friend require_started_op_state<Sender, Receiver>
+            tag_invoke(pika::execution::experimental::connect_t,
+                require_started_sender_type const& s, Receiver&& receiver)
+            {
+                if (!s.sender.has_value())
+                {
+#if defined(PIKA_HAVE_STDEXEC)
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(
+                        "pika::execution::experimental::connect(require_started_sender const&)",
+                        "Trying to connect an empty require_started sender");
+#else
+                    PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER(s.mode,
+                        "pika::execution::experimental::connect(require_started_sender const&)",
+                        "Trying to connect an empty require_started sender");
+#endif
+                }
+
+                s.connected = true;
+                return
+                {
+                    s.sender.value(), PIKA_FORWARD(Receiver, receiver)
+#if !defined(PIKA_HAVE_STDEXEC)
+                                          ,
+                        s.mode
+#endif
+                };
+            }
+
+            void discard() noexcept { connected = true; }
+#if !defined(PIKA_HAVE_STDEXEC)
+            void set_mode(require_started_mode mode) noexcept { this->mode = mode; }
+#endif
+        };
+
+#undef PIKA_DETAIL_HANDLE_UNSTARTED_REQUIRE_STARTED_SENDER
+    }    // namespace require_started_detail
+
+    namespace execution::experimental {
+        inline constexpr struct require_started_t final
+        {
+            template <typename Sender, PIKA_CONCEPT_REQUIRES_(is_sender_v<Sender>)>
+            constexpr PIKA_FORCEINLINE auto operator()(Sender&& sender
+#if !defined(PIKA_HAVE_STDEXEC)
+                ,
+                require_started_mode mode = require_started_mode::terminate_on_unstarted
+#endif
+            ) const
+            {
+                return require_started_detail::require_started_sender<Sender>
+                {
+                    PIKA_FORWARD(Sender, sender)
+#if !defined(PIKA_HAVE_STDEXEC)
+                    , mode
+#endif
+                };
+            }
+
+            constexpr PIKA_FORCEINLINE auto operator()() const
+            {
+                return detail::partial_algorithm<require_started_t>{};
+            }
+        } require_started{};
+    }    // namespace execution::experimental
+}    // namespace pika
diff --git a/libs/pika/execution/tests/unit/CMakeLists.txt b/libs/pika/execution/tests/unit/CMakeLists.txt
index ad9444d259..1bd727a7ad 100644
--- a/libs/pika/execution/tests/unit/CMakeLists.txt
+++ b/libs/pika/execution/tests/unit/CMakeLists.txt
@@ -13,6 +13,8 @@ set(tests
     algorithm_just
     algorithm_let_error
     algorithm_let_value
+    algorithm_require_started
+    algorithm_require_started_terminate
     algorithm_split
     algorithm_split_tuple
     algorithm_start_detached
@@ -33,7 +35,6 @@ foreach(test ${tests})
 
   set(folder_name "Tests/Unit/Modules/Execution")
 
-  # add example executable
   pika_add_executable(
     ${test}_test INTERNAL_FLAGS
     SOURCES ${sources} ${${test}_FLAGS}
@@ -47,3 +48,10 @@ foreach(test ${tests})
     "modules.execution" ${test} ${${test}_PARAMETERS} THREADS 4 VALGRIND
   )
 endforeach()
+
+set_tests_properties(
+  tests.unit.modules.execution.algorithm_require_started_terminate
+  PROPERTIES
+    PASS_REGULAR_EXPRESSION
+    ".*~require_started_sender: A require_started sender was never started\nstd::terminate called*"
+)
diff --git a/libs/pika/execution/tests/unit/algorithm_require_started.cpp b/libs/pika/execution/tests/unit/algorithm_require_started.cpp
new file mode 100644
index 0000000000..3c8ec54126
--- /dev/null
+++ b/libs/pika/execution/tests/unit/algorithm_require_started.cpp
@@ -0,0 +1,352 @@
+//  Copyright (c) 2023 ETH Zurich
+//
+//  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 <pika/execution.hpp>
+#include <pika/execution_base/tests/algorithm_test_utils.hpp>
+#include <pika/testing.hpp>
+
+#include <atomic>
+#include <exception>
+#include <memory>
+#include <stdexcept>
+#include <utility>
+
+namespace ex = pika::execution::experimental;
+namespace tt = pika::this_thread::experimental;
+
+void test_basics()
+{
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> started{false};
+        auto s1 = ex::then(void_sender{}, [&] { started = true; });
+        auto s2 = ex::require_started(std::move(s1));
+        PIKA_TEST(!started);
+        auto f = [] {};
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        PIKA_TEST(!started);
+        ex::start(os);
+        PIKA_TEST(started);
+        PIKA_TEST(set_value_called);
+    }
+
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> started{false};
+        auto s1 = ex::then(ex::just(0), [&](int x) {
+            started = true;
+            return x;
+        });
+        auto s2 = ex::require_started(std::move(s1));
+        PIKA_TEST(!started);
+        auto f = [](int x) { PIKA_TEST_EQ(x, 0); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        PIKA_TEST(!started);
+        ex::start(os);
+        PIKA_TEST(started);
+        PIKA_TEST(set_value_called);
+    }
+
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> started{false};
+        auto s1 = ex::then(ex::just(custom_type_non_default_constructible{42}),
+            [&](custom_type_non_default_constructible x) {
+                started = true;
+                return x;
+            });
+        auto s2 = ex::require_started(std::move(s1));
+        PIKA_TEST(!started);
+        auto f = [](auto x) { PIKA_TEST_EQ(x.x, 42); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        PIKA_TEST(!started);
+        ex::start(os);
+        PIKA_TEST(started);
+        PIKA_TEST(set_value_called);
+    }
+
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> started{false};
+        auto s1 = ex::then(ex::just(custom_type_non_default_constructible_non_copyable{42}),
+            [&](custom_type_non_default_constructible_non_copyable&& x) {
+                started = true;
+                return std::move(x);
+            });
+        auto s2 = ex::require_started(std::move(s1));
+        PIKA_TEST(!started);
+        auto f = [](auto&& x) { PIKA_TEST_EQ(x.x, 42); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        PIKA_TEST(!started);
+        ex::start(os);
+        PIKA_TEST(started);
+        PIKA_TEST(set_value_called);
+    }
+
+    {
+        std::atomic<bool> set_value_called{false};
+        int x = 42;
+        auto s1 = const_reference_sender<int>{x};
+        auto s2 = ex::require_started(std::move(s1));
+        auto f = [](auto&& x) { PIKA_TEST_EQ(x, 42); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+    }
+}
+
+void test_pipe_operator()
+{
+    {
+        std::atomic<bool> set_value_called{false};
+        auto s = void_sender{} | ex::require_started();
+        auto f = [] {};
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+    }
+}
+
+void test_failure()
+{
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = error_sender{} | ex::require_started();
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = const_reference_error_sender{} | ex::require_started();
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s =
+            error_sender{} | ex::require_started() | ex::require_started() | ex::require_started();
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+}
+
+void test_adl()
+{
+    auto s = ex::require_started(my_namespace::my_sender{});
+    test_adl_isolation(s);
+    tt::sync_wait(std::move(s));
+}
+
+enum class expect_exception
+{
+    no,
+    yes
+};
+
+// Check that the sender throws if it's not connected or the operation state isn't started
+template <typename F>
+void check_exception(expect_exception e, F test)
+{
+    try
+    {
+        test();
+        PIKA_TEST(e == expect_exception::no);
+    }
+    catch (...)
+    {
+        PIKA_TEST(e == expect_exception::yes);
+    }
+}
+
+void test_no_exception()
+{
+    // Expect no exception to be thrown
+    check_exception(expect_exception::no, [] {
+        auto s = void_sender{};
+        auto rs = ex::require_started(s);
+        tt::sync_wait(std::move(rs));
+    });
+
+    check_exception(expect_exception::no, [] {
+        auto s = void_sender{};
+        auto rs1 = ex::require_started(s);
+        auto rs2 = std::move(rs1);
+        tt::sync_wait(std::move(rs2));
+    });
+
+    check_exception(expect_exception::no, [] {
+        auto s = void_sender{};
+        auto rs1 = ex::require_started(s);
+        auto rs2 = std::move(rs1);
+        rs1 = std::move(rs2);
+        tt::sync_wait(std::move(rs1));
+    });
+
+    check_exception(expect_exception::no, [] {
+        auto s = void_sender{};
+        auto rs1 = ex::require_started(s);
+        auto rs2 = std::move(rs1);
+        // NOLINTNEXTLINE(bugprone-use-after-move)
+        auto rs3 = std::move(rs1);
+        tt::sync_wait(std::move(rs2));
+    });
+
+    check_exception(expect_exception::no, [] {
+        auto s = void_sender{};
+        auto rs1 = ex::require_started(s);
+        auto rs2 = std::move(rs1);
+        // NOLINTNEXTLINE(bugprone-use-after-move)
+        auto rs3 = rs1;
+        tt::sync_wait(std::move(rs2));
+    });
+}
+
+enum class exception_test_mode
+{
+    no_discard,
+    discard
+};
+
+template <typename S>
+void discard_if_required(S& s, exception_test_mode mode)
+{
+    if (mode == exception_test_mode::discard) { s.discard(); }
+}
+
+void test_exception(exception_test_mode mode)
+{
+#if !defined(PIKA_HAVE_STDEXEC)
+    // When the mode is no_discard we expect exceptions, otherwise we don't expect exceptions
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs = ex::require_started(
+                void_sender{}, ex::require_started_mode::throw_on_unstarted);    // never started
+            discard_if_required(rs, mode);
+        });
+
+    // Move constructor
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 = ex::require_started(
+                void_sender{}, ex::require_started_mode::throw_on_unstarted);    // empty
+            auto rs2 = std::move(rs1);                                           // never started
+            discard_if_required(rs2, mode);
+        });
+
+    // Copy constructor
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            auto rs2 = rs1;        // never started
+            tt::sync_wait(rs1);    // started
+            discard_if_required(rs2, mode);
+        });
+
+    // Move assignment
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            auto rs2 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            tt::sync_wait(std::move(rs2));    // started
+            rs2 = std::move(rs1);             // never started
+            discard_if_required(rs2, mode);
+        });
+
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            tt::sync_wait(std::move(rs1));    // started
+            auto rs2 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            discard_if_required(rs2, mode);
+            rs2 = std::move(rs1);    // never started
+        });
+
+    // Copy assignment
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            auto rs2 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            tt::sync_wait(std::move(rs2));    // started
+            rs2 = rs1;                        // never started
+            tt::sync_wait(std::move(rs1));    // started
+            discard_if_required(rs2, mode);
+        });
+
+    check_exception(
+        mode == exception_test_mode::discard ? expect_exception::no : expect_exception::yes, [=] {
+            auto rs1 =
+                ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+            tt::sync_wait(std::move(rs1));    // started
+            auto rs2 = ex::require_started(
+                void_sender{}, ex::require_started_mode::throw_on_unstarted);    // never started
+            discard_if_required(rs2, mode);
+            rs2 = rs1;
+        });
+#else
+    PIKA_UNUSED(mode);
+#endif
+}
+
+void test_unstarted()
+{
+#if !defined(PIKA_HAVE_STDEXEC)
+    // Connected, but not started
+    check_exception(expect_exception::yes, [] {
+        auto rs = ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+        std::atomic<bool> set_value_called{false};
+        auto f = [] {};
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(rs), std::move(r));
+    });
+
+    check_exception(expect_exception::yes, [] {
+        auto rs = ex::require_started(void_sender{}, ex::require_started_mode::throw_on_unstarted);
+        std::atomic<bool> set_value_called{false};
+        auto f = [] {};
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(rs, std::move(r));
+        tt::sync_wait(std::move(rs));
+    });
+#endif
+}
+
+int main()
+{
+    test_basics();
+    test_pipe_operator();
+    test_failure();
+    test_adl();
+    test_no_exception();
+    test_exception(exception_test_mode::no_discard);
+    test_exception(exception_test_mode::discard);
+    test_unstarted();
+
+    return 0;
+}
diff --git a/libs/pika/execution/tests/unit/algorithm_require_started_terminate.cpp b/libs/pika/execution/tests/unit/algorithm_require_started_terminate.cpp
new file mode 100644
index 0000000000..c97692d997
--- /dev/null
+++ b/libs/pika/execution/tests/unit/algorithm_require_started_terminate.cpp
@@ -0,0 +1,30 @@
+//  Copyright (c) 2023 ETH Zurich
+//
+//  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 <pika/execution.hpp>
+#include <pika/testing.hpp>
+
+#include <cstdlib>
+#include <exception>
+#include <iostream>
+
+namespace ex = pika::execution::experimental;
+
+int main()
+{
+    std::set_terminate([] {
+        std::cout << "std::terminate called\n";
+        std::exit(pika::detail::report_errors());
+    });
+
+    {
+        PIKA_TEST(true);
+        auto rs = ex::require_started(ex::just());
+    }
+
+    // The test should terminate above because rs is never connected or started
+    PIKA_TEST(false);
+}