diff --git a/example/awaitable-sender/CMakeLists.txt b/example/awaitable-sender/CMakeLists.txt
index d4561416f..40f0ff8fe 100644
--- a/example/awaitable-sender/CMakeLists.txt
+++ b/example/awaitable-sender/CMakeLists.txt
@@ -17,20 +17,27 @@ FetchContent_Declare(
 )
 FetchContent_MakeAvailable(execution)
 
-file(GLOB_RECURSE PFILES CONFIGURE_DEPENDS *.cpp *.hpp
-    CMakeLists.txt
-    Jamfile)
-
-source_group(TREE ${CMAKE_CURRENT_SOURCE_DIR} PREFIX "" FILES ${PFILES})
-
-add_executable(capy_example_awaitable_sender ${PFILES})
-
-set_property(TARGET capy_example_awaitable_sender
-    PROPERTY FOLDER "examples")
-
-target_compile_features(capy_example_awaitable_sender
-    PRIVATE cxx_std_23)
-
-target_link_libraries(capy_example_awaitable_sender
-    Boost::capy
-    beman::execution_headers)
+add_executable(capy_example_awaitable_sender
+    awaitable_sender.cpp
+    awaitable_sender.hpp
+    awaitable_sender_base.hpp
+    awaitable_sender_detail.hpp
+    read_op.hpp)
+
+add_executable(capy_example_awaitable_sender_test
+    tests.cpp
+    awaitable_sender.hpp
+    awaitable_sender_base.hpp
+    awaitable_sender_detail.hpp
+    read_op.hpp)
+
+foreach(tgt capy_example_awaitable_sender capy_example_awaitable_sender_test)
+    set_property(TARGET ${tgt} PROPERTY FOLDER "examples")
+    target_compile_features(${tgt} PRIVATE cxx_std_23)
+    target_link_libraries(${tgt}
+        Boost::capy
+        beman::execution_headers)
+endforeach()
+
+add_test(NAME capy_example_awaitable_sender_test
+    COMMAND capy_example_awaitable_sender_test)
diff --git a/example/awaitable-sender/awaitable_sender.cpp b/example/awaitable-sender/awaitable_sender.cpp
index 1ddef26a7..7a92cf6a3 100644
--- a/example/awaitable-sender/awaitable_sender.cpp
+++ b/example/awaitable-sender/awaitable_sender.cpp
@@ -9,6 +9,8 @@
 //
 
 #include "awaitable_sender.hpp"
+#include "awaitable_sender_base.hpp"
+#include "read_op.hpp"
 
 #include <boost/capy.hpp>
 
@@ -180,6 +182,28 @@ int main()
     done4.wait();
     std::cout << "  split_ec error test done\n";
 
+    // A native awaitable-sender: read_op derives
+    // awaitable_sender_base, no as_sender() wrapping.
+    std::cout << "\n--- native awaitable-sender test ---\n";
+    std::latch done5(1);
+
+    auto rop = capy::read_op::result({}, 42);
+    auto op5 = ex::connect(
+        ex::then(
+            std::move(rop),
+            [](std::size_t n)
+            {
+                std::cout
+                    << "  read " << n << " bytes\n";
+            }),
+        demo_receiver{
+            {pool_ex, std::stop_token{}},
+            &done5});
+
+    ex::start(op5);
+    done5.wait();
+    std::cout << "  native awaitable-sender test done\n";
+
     // All demos have drained; safe to join the waker thread now.
     waker_thread.join();
 }
diff --git a/example/awaitable-sender/awaitable_sender.hpp b/example/awaitable-sender/awaitable_sender.hpp
index f942605be..0ae61617c 100644
--- a/example/awaitable-sender/awaitable_sender.hpp
+++ b/example/awaitable-sender/awaitable_sender.hpp
@@ -11,329 +11,100 @@
 #ifndef BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_HPP
 #define BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_HPP
 
+#include "awaitable_sender_base.hpp"
+#include "awaitable_sender_detail.hpp"
+
 #include <boost/capy/concept/io_awaitable.hpp>
-#include <boost/capy/detail/await_suspend_helper.hpp>
-#include <boost/capy/ex/executor_ref.hpp>
-#include <boost/capy/ex/io_env.hpp>
-#include <boost/capy/io_result.hpp>
 
 #include <beman/execution/execution.hpp>
 
-#include <concepts>
-#include <coroutine>
-#include <exception>
-#include <stop_token>
-#include <tuple>
 #include <type_traits>
 #include <utility>
 
 namespace boost::capy {
 
-// -------------------------------------------------------
-// CPO: query a receiver environment for a Capy executor
-// -------------------------------------------------------
-
-struct get_io_executor_t
-{
-    template<class Env>
-    auto operator()(Env const& env) const noexcept
-        -> decltype(env.query(std::declval<get_io_executor_t const&>()))
-    {
-        return env.query(*this);
-    }
-};
-
-inline constexpr get_io_executor_t get_io_executor{};
-
-// -------------------------------------------------------
-// Environment that carries a Capy executor + stop token
-// -------------------------------------------------------
-
-struct io_sender_env
-{
-    executor_ref io_executor;
-    std::stop_token stop_token;
-
-    auto query(
-        get_io_executor_t const&) const noexcept
-        -> executor_ref
-    {
-        return io_executor;
-    }
-
-    auto query(
-        beman::execution::get_stop_token_t const&) const noexcept
-        -> std::stop_token
-    {
-        return stop_token;
-    }
-};
-
-namespace detail {
-
-template<class T, class = void>
-struct has_tuple_protocol : std::false_type {};
-
-template<class T>
-struct has_tuple_protocol<T,
-    std::void_t<
-        typename std::tuple_size<T>::type,
-        typename std::tuple_element<0, T>::type>>
-    : std::true_type {};
-
-template<class T, bool = has_tuple_protocol<T>::value>
-struct is_ec_outcome : std::is_same<T, std::error_code> {};
-
-template<class T>
-struct is_ec_outcome<T, true>
-    : std::bool_constant<
-        std::tuple_size_v<T> == 1 &&
-        std::is_same_v<
-            std::tuple_element_t<0, T>,
-            std::error_code>>
-{};
-
-template<class T>
-constexpr bool is_ec_outcome_v =
-    std::is_same_v<T, std::error_code> ||
-    is_ec_outcome<T>::value;
-
-template<class T, bool = has_tuple_protocol<T>::value>
-struct is_compound_ec_result : std::false_type {};
-
-template<class T>
-struct is_compound_ec_result<T, true>
-    : std::bool_constant<
-        std::tuple_size_v<T> >= 2 &&
-        std::is_same_v<
-            std::tuple_element_t<0, T>,
-            std::error_code>>
-{};
-
-template<class T>
-constexpr bool is_compound_ec_result_v =
-    is_compound_ec_result<T>::value;
-
-// -------------------------------------------------------
-// frame_cb: synthetic coroutine frame for callback handles
-//
-// The first two members match the coroutine frame layout
-// used by MSVC, GCC, and Clang. from_address produces a
-// coroutine_handle whose .resume() calls our function
-// pointer and whose .destroy() is a no-op.
-// -------------------------------------------------------
-
-struct frame_cb
-{
-    void (*resume)(frame_cb*);
-    void (*destroy)(frame_cb*);
-    void* data;
-};
-
-} // namespace detail
-
 // -------------------------------------------------------
 // Sender that wraps an IoAwaitable
 // -------------------------------------------------------
 
+/** A sender that wraps an IoAwaitable.
+
+    Adapts an IoAwaitable to the `std::execution` sender
+    concept, enabling composition with other sender operations
+    and adapters.
+*/
 template<class IoAw>
 struct awaitable_sender
 {
-    using sender_concept = beman::execution::sender_t;
+    using sender_concept = ex::sender_t;
+
+    IoAw aw_;
 
-    using result_type = decltype(
-        std::declval<std::decay_t<IoAw>&>().await_resume());
+    /** Return the completion signatures deduced from `IoAw`.
 
-    static auto make_sigs()
+        C++26 static-template form (see @ref awaitable_sender_base
+        for the mechanism notes).
+    */
+    template<class Sndr, class... Env>
+    static consteval auto get_completion_signatures() noexcept
     {
-        if constexpr (std::is_void_v<result_type>)
-            return beman::execution::completion_signatures<
-                beman::execution::set_value_t(),
-                beman::execution::set_error_t(std::exception_ptr),
-                beman::execution::set_stopped_t()>{};
-        else if constexpr (
-            detail::is_ec_outcome_v<result_type>)
-            return beman::execution::completion_signatures<
-                beman::execution::set_value_t(),
-                beman::execution::set_error_t(std::error_code),
-                beman::execution::set_error_t(std::exception_ptr),
-                beman::execution::set_stopped_t()>{};
-        else
-            return beman::execution::completion_signatures<
-                beman::execution::set_value_t(result_type),
-                beman::execution::set_error_t(std::exception_ptr),
-                beman::execution::set_stopped_t()>{};
+        return decltype(detail::make_sigs<IoAw>()){};
     }
 
-    using completion_signatures = decltype(make_sigs());
-
-    IoAw aw_;
-
-    template<class Receiver>
-    struct op_state
+    /// beman-compat form: DROP AT GRADUATION (pre-P3164 protocol).
+    template<class Env>
+    constexpr auto get_completion_signatures(
+        Env const&) const noexcept
     {
-        using operation_state_concept =
-            beman::execution::operation_state_t;
-
-        IoAw aw_;
-        Receiver rcvr_;
-        io_env env_;
-        detail::frame_cb cb_;
-
-        op_state(IoAw aw, Receiver rcvr)
-            : aw_(std::move(aw))
-            , rcvr_(std::move(rcvr))
-            , cb_{}
-        {
-        }
-
-        op_state(op_state const&) = delete;
-        op_state(op_state&&) = delete;
-        op_state& operator=(op_state const&) = delete;
-        op_state& operator=(op_state&&) = delete;
-
-        static void
-        on_resume(detail::frame_cb* p) noexcept
-        {
-            auto* self = static_cast<op_state*>(p->data);
-            self->complete();
-        }
-
-        static void
-        on_destroy(detail::frame_cb*) noexcept
-        {
-        }
-
-        void complete() noexcept
-        {
-            try
-            {
-                if constexpr (std::is_void_v<result_type>)
-                {
-                    aw_.await_resume();
-                    if(env_.stop_token.stop_requested())
-                        beman::execution::set_stopped(
-                            std::move(rcvr_));
-                    else
-                        beman::execution::set_value(
-                            std::move(rcvr_));
-                }
-                else if constexpr (
-                    detail::is_ec_outcome_v<result_type>)
-                {
-                    auto result = aw_.await_resume();
-                    if(env_.stop_token.stop_requested())
-                    {
-                        beman::execution::set_stopped(
-                            std::move(rcvr_));
-                    }
-                    else
-                    {
-                        std::error_code ec;
-                        if constexpr (std::is_same_v<
-                            result_type, std::error_code>)
-                            ec = result;
-                        else
-                            ec = get<0>(result);
-                        if(!ec)
-                            beman::execution::set_value(
-                                std::move(rcvr_));
-                        else
-                            beman::execution::set_error(
-                                std::move(rcvr_), ec);
-                    }
-                }
-                else
-                {
-                    auto result = aw_.await_resume();
-                    if(env_.stop_token.stop_requested())
-                        beman::execution::set_stopped(
-                            std::move(rcvr_));
-                    else
-                        beman::execution::set_value(
-                            std::move(rcvr_),
-                            std::move(result));
-                }
-            }
-            catch(...)
-            {
-                beman::execution::set_error(
-                    std::move(rcvr_),
-                    std::current_exception());
-            }
-        }
-
-        void start() noexcept
-        {
-            auto renv = beman::execution::get_env(rcvr_);
-            auto ex = get_io_executor(renv);
-
-            std::stop_token st;
-            if constexpr (requires {
-                { renv.query(beman::execution::get_stop_token_t{}) }
-                    -> std::convertible_to<std::stop_token>; })
-            {
-                st = renv.query(
-                    beman::execution::get_stop_token_t{});
-            }
-
-            env_ = io_env{ex, st, nullptr};
-
-            if(aw_.await_ready())
-            {
-                complete();
-                return;
-            }
-
-            cb_.resume = &on_resume;
-            cb_.destroy = &on_destroy;
-            cb_.data = this;
-
-            auto h = std::coroutine_handle<>::from_address(
-                static_cast<void*>(&cb_));
-
-            // Not a real coroutine caller, so symmetric transfer
-            // must be driven by hand: any non-noop handle (our own
-            // frame on immediate completion, or a wrapped task's
-            // handle that still needs to run) has to be resumed
-            // explicitly or nothing ever completes.
-            auto resumed = detail::call_await_suspend(&aw_, h, &env_);
-            if(resumed != std::noop_coroutine())
-                resumed.resume();
-        }
-    };
+        return decltype(detail::make_sigs<IoAw>()){};
+    }
 
+    /// Connect this sender with a receiver to form an operation state.
     template<class Receiver>
     auto connect(Receiver rcvr) &&
-        -> op_state<Receiver>
+        -> detail::awaitable_op_state<IoAw, Receiver>
     {
-        return op_state<Receiver>(
+        return detail::awaitable_op_state<IoAw, Receiver>(
             std::move(aw_), std::move(rcvr));
     }
 
+    /// Connect a copy of the sender to a receiver.
     template<class Receiver>
     auto connect(Receiver rcvr) const&
-        -> op_state<Receiver>
+        -> detail::awaitable_op_state<IoAw, Receiver>
     {
-        return op_state<Receiver>(aw_, std::move(rcvr));
+        return detail::awaitable_op_state<IoAw, Receiver>(
+            aw_, std::move(rcvr));
     }
 };
 
-/** Create a beman::execution sender from an IoAwaitable.
+/** Create a `std::execution` sender from an IoAwaitable.
 
     The bridge routes the awaitable's result through sender
     channels based on its type:
 
     - `void` - calls `set_value()`.
-    - `error_code` (or a single-element tuple-like whose
-      element 0 is `error_code`) - calls `set_value()`
-      when the code is zero, `set_error(ec)` otherwise.
-    - Any other single value `T` - calls `set_value(T)`.
-    - Compound results whose element 0 is `error_code`
-      with additional elements are rejected at compile
-      time. Wrap the operation in a `task<error_code>`
-      that inspects the compound result and returns the
-      error code.
+    - `error_code` or an empty `io_result` - calls
+      `set_value()` when the code is zero, `set_error(ec)`
+      otherwise.
+    - `io_result<Ts...>` with payload elements - calls
+      `set_value(ts...)` when `ec` is zero, `set_error(ec)`
+      otherwise. Any partial payload accompanying a truthy
+      `ec` is dropped, since sender completion channels are
+      exclusive.
+    - Any other single value `T` - calls `set_value(T)`,
+      including generic tuple-likes that happen to lead with
+      an `error_code`: only `io_result` declares the
+      element-0-is-outcome intent, so only it is split.
+
+    For the `error_code`-carrying result types the channel is
+    chosen by the operation's own disposition: an `ec` that
+    compares equal to `errc::operation_canceled` completes with
+    `set_stopped()`, and a successful result is delivered even
+    if a stop request arrived while the operation was finishing.
+    `void` and plain-value results carry no disposition, so for
+    those the environment's stop token decides between
+    `set_stopped()` and the completion above.
 
     @par Example
     @code
@@ -347,18 +118,30 @@ struct awaitable_sender
 template<class IoAw>
 auto as_sender(IoAw&& aw)
 {
-    using R = decltype(
-        std::declval<std::decay_t<IoAw>&>().await_resume());
-    static_assert(
-        !detail::is_compound_ec_result_v<std::decay_t<R>>,
-        "as_sender does not accept awaitables whose result "
-        "destructures into (error_code, ...). Wrap the "
-        "operation in a task<error_code> that inspects the "
-        "compound result and returns the error code.");
     return awaitable_sender<std::decay_t<IoAw>>{
         std::forward<IoAw>(aw)};
 }
 
+/** Return the awaitable as a sender.
+
+    Boundary normalizer for generic code handed an arbitrary
+    @ref IoAwaitable: an op that already models
+    @ref AwaitableSender passes through unchanged, anything else
+    is lifted with @ref as_sender. Either way the result is a
+    sender by value.
+
+    @param a The IoAwaitable to normalize.
+    @return `a` itself, or `as_sender(a)`.
+*/
+template<IoAwaitable A>
+auto ensure_sender(A&& a)
+{
+    if constexpr (AwaitableSender<std::remove_cvref_t<A>>)
+        return std::forward<A>(a);
+    else
+        return as_sender(std::forward<A>(a));
+}
+
 // -------------------------------------------------------
 // split_ec: sender adapter that routes error_code to
 // set_value() or set_error(ec) at runtime.
@@ -369,56 +152,71 @@ namespace detail {
 template<class Sender>
 struct split_ec_sender
 {
-    using sender_concept = beman::execution::sender_t;
+    using sender_concept = ex::sender_t;
 
-    using completion_signatures =
-        beman::execution::completion_signatures<
-            beman::execution::set_value_t(),
-            beman::execution::set_error_t(std::error_code),
-            beman::execution::set_error_t(std::exception_ptr),
-            beman::execution::set_stopped_t()>;
+    using sigs_type =
+        ex::completion_signatures<
+            ex::set_value_t(),
+            ex::set_error_t(std::error_code),
+            ex::set_error_t(std::exception_ptr),
+            ex::set_stopped_t()>;
 
     Sender sndr_;
 
+    // C++26 static-template form plus the beman-compat instance
+    // form (DROP the latter at graduation; pre-P3164 protocol).
+    template<class Sndr, class... Env>
+    static consteval auto get_completion_signatures() noexcept
+    {
+        return sigs_type{};
+    }
+
+    template<class Env>
+    constexpr auto get_completion_signatures(
+        Env const&) const noexcept
+    {
+        return sigs_type{};
+    }
+
     template<class Receiver>
     struct ec_receiver
     {
-        using receiver_concept = beman::execution::receiver_t;
+        using receiver_concept = ex::receiver_t;
 
         Receiver rcvr_;
 
         auto get_env() const noexcept
         {
-            return beman::execution::get_env(rcvr_);
+            return ex::get_env(rcvr_);
         }
 
         void set_value(std::error_code ec) && noexcept
         {
             if (!ec)
-                beman::execution::set_value(
+                ex::set_value(
                     std::move(rcvr_));
             else
-                beman::execution::set_error(
+                ex::set_error(
                     std::move(rcvr_), ec);
         }
 
         void set_value() && noexcept
         {
-            beman::execution::set_value(
+            ex::set_value(
                 std::move(rcvr_));
         }
 
         template<class E>
         void set_error(E&& e) && noexcept
         {
-            beman::execution::set_error(
+            ex::set_error(
                 std::move(rcvr_),
                 std::forward<E>(e));
         }
 
         void set_stopped() && noexcept
         {
-            beman::execution::set_stopped(
+            ex::set_stopped(
                 std::move(rcvr_));
         }
     };
@@ -427,17 +225,17 @@ struct split_ec_sender
     struct op_state
     {
         using operation_state_concept =
-            beman::execution::operation_state_t;
+            ex::operation_state_t;
 
         using inner_op_t = decltype(
-            beman::execution::connect(
+            ex::connect(
                 std::declval<Sender>(),
                 std::declval<ec_receiver<Receiver>>()));
 
         inner_op_t op_;
 
         op_state(Sender sndr, Receiver rcvr)
-            : op_(beman::execution::connect(
+            : op_(ex::connect(
                 std::move(sndr),
                 ec_receiver<Receiver>{std::move(rcvr)}))
         {
@@ -450,7 +248,7 @@ struct split_ec_sender
 
         void start() noexcept
         {
-            beman::execution::start(op_);
+            ex::start(op_);
         }
     };
 
diff --git a/example/awaitable-sender/awaitable_sender_base.hpp b/example/awaitable-sender/awaitable_sender_base.hpp
new file mode 100644
index 000000000..e0491e6bc
--- /dev/null
+++ b/example/awaitable-sender/awaitable_sender_base.hpp
@@ -0,0 +1,171 @@
+//
+// Copyright (c) 2026 Steve Gerbino
+//
+// 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)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#ifndef BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_BASE_HPP
+#define BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_BASE_HPP
+
+#include "awaitable_sender_detail.hpp"
+
+#include <type_traits>
+#include <utility>
+
+namespace boost::capy {
+
+/** Concept for awaitables that are also senders.
+
+    Refines @ref IoAwaitable with the `std::execution` sender
+    requirements: the same op can be driven by `co_await` in a capy
+    coroutine and by `connect`/`start` in a sender pipeline. Both
+    halves are structural — deriving @ref awaitable_sender_base is
+    the convenient way to satisfy the sender half, not the only way.
+
+    @par Semantic Requirements
+    The two protocols drive the identical awaitable members, so
+    they agree only when the op reports its full disposition
+    in-band through `await_resume()`, with cancellation expressed
+    as an `error_code` comparing equal to
+    `errc::operation_canceled` (see @ref awaitable_sender_base's
+    Dual-Protocol Contract). Syntax alone cannot check this.
+
+    @note Distinct from `std::execution::with_awaitable_senders`,
+        which adapts in the opposite direction (senders made
+        awaitable inside a coroutine).
+
+    @tparam S The op type.
+*/
+template<class S>
+concept AwaitableSender =
+    IoAwaitable<S> &&
+    ex::sender<S>;
+
+namespace detail {
+
+// Shared by both connect() overloads so const& connect gets the
+// same check as the rvalue overload. Scaffolding-era check, DROP
+// AT GRADUATION with the beman-compat query forms: the bundled
+// implementation probes decomposable senders by aggregate
+// brace-init and hard-errors on arity mismatches. Under the
+// adopted wording the durable opt-out is private data members
+// (see the class note); aggregates merely take a guarded
+// fallback path there.
+template<class Derived>
+constexpr void check_awaitable_sender()
+{
+    static_assert(
+        !std::is_aggregate_v<Derived>,
+        "Derived must not be an aggregate; keep its data "
+        "members private (or declare a constructor) so "
+        "sender decomposition cannot claim it");
+}
+
+} // namespace detail
+
+/** CRTP mixin that makes an IoAwaitable a sender.
+
+    Deriving from this base adds the C++26 sender interface to
+    an I/O awaitable, so the same op type works under `co_await`
+    in capy coroutines and under `connect`/`start` in sender
+    pipelines. The completion signatures are deduced from
+    `Derived::await_resume()` with the same rules as `as_sender`;
+    deduction runs inside the signature query, where `Derived` is
+    complete, so nothing is stated twice and the advertised
+    signatures cannot drift from the implementation.
+
+    @par Dual-Protocol Contract
+    Both protocols drive the identical awaitable members, so the
+    two views cannot diverge as long as `Derived` reports its full
+    disposition in-band: `await_resume()`'s result carries the
+    outcome, with cancellation expressed as an `error_code`
+    comparing equal to `errc::operation_canceled`. The sender
+    machinery derives completion channels from that result alone.
+    Results without an error channel (`void`, plain values) are
+    dual-use-safe only for ops that cannot be canceled.
+
+    @note Derived ops should keep their data members private.
+        The standard imposes no access requirements on senders,
+        but `std::execution` sender decomposition claims any type
+        whose members admit a structured binding and treats the
+        first member as a sender tag; private members make the
+        binding ill-formed, so the op is categorically
+        non-decomposable and independent of the dispatch
+        machinery's guarded fallbacks. (`connect` additionally
+        rejects aggregates; that check accommodates the bundled
+        pre-standard implementation and is dropped at
+        graduation.)
+
+    @par Example
+    @code
+    struct read_op : awaitable_sender_base<read_op>
+    {
+        // usual IoAwaitable members; the completion signatures
+        // follow await_resume()'s result type
+    };
+    @endcode
+
+    @tparam Derived The op type deriving from this base.
+*/
+template<class Derived>
+struct awaitable_sender_base
+{
+    using sender_concept = ex::sender_t;
+
+    /** Return the completion signatures deduced from `Derived`.
+
+        This is the C++26 mechanism: a static member function
+        template the implementation calls as
+        `Sndr::template get_completion_signatures<Sndr, Env...>()`
+        ([exec.getcomplsigs]; P3164R4 also removed the
+        nested-typedef mechanism from the standard).
+    */
+    template<class Sndr, class... Env>
+    static consteval auto get_completion_signatures() noexcept
+    {
+        return decltype(detail::make_sigs<Derived>()){};
+    }
+
+    /** Return the completion signatures deduced from `Derived`.
+
+        beman-compat form: DROP AT GRADUATION. beman::execution
+        still implements the pre-P3164 protocol and probes for an
+        instance member called with the environment as a function
+        argument; the adopted C++26 wording recognizes only the
+        static template form above.
+    */
+    template<class Env>
+    constexpr auto get_completion_signatures(
+        Env const&) const noexcept
+    {
+        return decltype(detail::make_sigs<Derived>()){};
+    }
+
+    /// Connect the op to a receiver, consuming it.
+    template<class Receiver>
+    auto connect(Receiver rcvr) &&
+        -> detail::awaitable_op_state<Derived, Receiver>
+    {
+        detail::check_awaitable_sender<Derived>();
+        return detail::awaitable_op_state<Derived, Receiver>(
+            static_cast<Derived&&>(*this), std::move(rcvr));
+    }
+
+    /// Connect a copy of the op to a receiver.
+    template<class Receiver>
+    auto connect(Receiver rcvr) const&
+        -> detail::awaitable_op_state<Derived, Receiver>
+    {
+        detail::check_awaitable_sender<Derived>();
+        return detail::awaitable_op_state<Derived, Receiver>(
+            static_cast<Derived const&>(*this),
+            std::move(rcvr));
+    }
+};
+
+} // namespace boost::capy
+
+#endif
diff --git a/example/awaitable-sender/awaitable_sender_detail.hpp b/example/awaitable-sender/awaitable_sender_detail.hpp
new file mode 100644
index 000000000..df29d66c0
--- /dev/null
+++ b/example/awaitable-sender/awaitable_sender_detail.hpp
@@ -0,0 +1,599 @@
+//
+// Copyright (c) 2026 Steve Gerbino
+//
+// 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)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#ifndef BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_DETAIL_HPP
+#define BOOST_CAPY_EXAMPLE_AWAITABLE_SENDER_DETAIL_HPP
+
+#include <boost/capy/concept/io_awaitable.hpp>
+#include <boost/capy/detail/await_suspend_helper.hpp>
+#include <boost/capy/ex/execution_context.hpp>
+#include <boost/capy/ex/executor_ref.hpp>
+#include <boost/capy/ex/io_env.hpp>
+#include <boost/capy/io_result.hpp>
+
+#include <beman/execution/execution.hpp>
+
+#include <concepts>
+#include <coroutine>
+#include <exception>
+#include <optional>
+#include <stop_token>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+
+namespace boost::capy {
+
+// On graduation this alias flips to std::execution, gated on
+// __cpp_lib_senders.
+namespace ex = beman::execution;
+
+// -------------------------------------------------------
+// CPO: query a receiver environment for a Capy executor
+// -------------------------------------------------------
+
+struct get_io_executor_t
+    : ex::forwarding_query_t
+{
+    // Derive from forwarding_query_t so env-wrapping adaptors
+    // like then forward this query: std::execution's adaptor
+    // environments only re-expose queries whose CPO is itself a
+    // forwarding_query, filtering out the rest by default.
+    template<class Env>
+    auto operator()(Env const& env) const noexcept
+        -> decltype(env.query(std::declval<get_io_executor_t const&>()))
+    {
+        return env.query(*this);
+    }
+};
+
+inline constexpr get_io_executor_t get_io_executor{};
+
+// -------------------------------------------------------
+// Environment that carries a Capy executor + stop token
+// -------------------------------------------------------
+
+struct io_sender_env
+{
+    executor_ref io_executor;
+    std::stop_token stop_token;
+
+    auto query(
+        get_io_executor_t const&) const noexcept
+        -> executor_ref
+    {
+        return io_executor;
+    }
+
+    auto query(
+        ex::get_stop_token_t const&) const noexcept
+        -> std::stop_token
+    {
+        return stop_token;
+    }
+};
+
+namespace detail {
+
+// Channel splitting keys on capy's io_result, the type that
+// declares "element 0 is an error_code" as intent. A generic
+// tuple-like that merely happens to lead with an error_code is a
+// value, not an outcome to split — shape alone cannot tell a
+// result protocol from a payload. The arity trait avoids naming
+// std::tuple_size on foreign types, which would hard-error for
+// non-tuples (&& does not short-circuit instantiation).
+template<class T>
+struct io_result_arity
+    : std::integral_constant<std::size_t, 0> {};
+
+template<class... Ts>
+struct io_result_arity<io_result<Ts...>>
+    : std::integral_constant<std::size_t, 1 + sizeof...(Ts)> {};
+
+template<class T>
+constexpr bool is_ec_outcome_v =
+    std::is_same_v<T, std::error_code> ||
+    io_result_arity<T>::value == 1;
+
+template<class T>
+constexpr bool is_compound_ec_result_v =
+    io_result_arity<T>::value >= 2;
+
+// -------------------------------------------------------
+// frame_cb: synthetic coroutine frame for callback handles
+//
+// The first two members match the coroutine frame layout
+// used by MSVC, GCC, and Clang. from_address produces a
+// coroutine_handle whose .resume() calls our function
+// pointer and whose .destroy() is a no-op.
+// -------------------------------------------------------
+
+struct frame_cb
+{
+    void (*resume)(frame_cb*);
+    void (*destroy)(frame_cb*);
+    void* data;
+};
+
+// declared only for decltype composition
+template<class... A, class... B>
+auto concat_sigs(
+    ex::completion_signatures<A...>,
+    ex::completion_signatures<B...>)
+    -> ex::completion_signatures<A..., B...>;
+
+template<class R, std::size_t... I>
+auto compound_value_sig(std::index_sequence<I...>)
+    -> ex::completion_signatures<
+        ex::set_value_t(std::tuple_element_t<I + 1, R>...),
+        ex::set_error_t(std::error_code)>;
+
+// Deduce completion signatures from an awaitable's result type.
+template<class Aw>
+auto make_sigs()
+{
+    using A = std::decay_t<Aw>;
+    using R = awaitable_result_t<A>;
+    constexpr bool nothrow_resume =
+        noexcept(std::declval<A&>().await_resume());
+
+    auto base = []{
+        if constexpr (std::is_void_v<R>)
+            return ex::completion_signatures<
+                ex::set_value_t()>{};
+        else if constexpr (is_ec_outcome_v<R>)
+            return ex::completion_signatures<
+                ex::set_value_t(),
+                ex::set_error_t(std::error_code)>{};
+        else if constexpr (is_compound_ec_result_v<R>)
+            return decltype(compound_value_sig<R>(
+                std::make_index_sequence<
+                    std::tuple_size_v<R> - 1>{})){};
+        else
+            return ex::completion_signatures<
+                ex::set_value_t(R)>{};
+    }();
+
+    auto tail = []{
+        if constexpr (nothrow_resume)
+            return ex::completion_signatures<
+                ex::set_stopped_t()>{};
+        else
+            return ex::completion_signatures<
+                ex::set_error_t(std::exception_ptr),
+                ex::set_stopped_t()>{};
+    }();
+
+    return decltype(concat_sigs(base, tail)){};
+}
+
+// beman's get_stop_token CPO requires the C++26 stoppable_token
+// concept, which today's std::stop_token fails (no callback_type);
+// query the environment directly so std::stop_token environments
+// keep cancellation until stdlibs catch up.
+template<class Env, class = void>
+struct env_stop_token
+{
+    using type = ex::never_stop_token;
+};
+
+template<class Env>
+struct env_stop_token<Env, std::void_t<
+    decltype(std::declval<Env const&>().query(
+        ex::get_stop_token_t{}))>>
+{
+    // remove_cvref_t so an env returning `std::stop_token const&`
+    // still matches token_passthrough below.
+    using type = std::remove_cvref_t<decltype(
+        std::declval<Env const&>().query(
+            ex::get_stop_token_t{}))>;
+};
+
+template<class T>
+struct is_polymorphic_allocator : std::false_type {};
+
+template<class T>
+struct is_polymorphic_allocator<
+    std::pmr::polymorphic_allocator<T>> : std::true_type {};
+
+// Bridges a std::execution scheduler to capy's Executor concept so
+// ops can run under receivers that provide only get_scheduler
+// (e.g. sync_wait's run_loop). Each post allocates a small holder
+// for the schedule operation; receivers supplying get_io_executor
+// never pay this.
+template<class Sched>
+class scheduler_executor
+{
+    Sched sched_;
+
+    struct holder;
+
+    struct resume_receiver
+    {
+        using receiver_concept = ex::receiver_t;
+
+        holder* self_;
+        std::coroutine_handle<> h_;
+
+        void set_value() && noexcept
+        {
+            auto h = h_;
+            delete self_;
+            h.resume();
+        }
+
+        // The scheduler refused the work (its context has
+        // finished): abandon without resuming, like a post to a
+        // stopped execution context.
+        template<class E>
+        void set_error(E&&) && noexcept
+        {
+            delete self_;
+        }
+
+        void set_stopped() && noexcept
+        {
+            delete self_;
+        }
+    };
+
+    struct holder
+    {
+        // schedule() wants a non-const scheduler; schedulers are
+        // cheap copyable handles, so take one by value.
+        using op_t = decltype(ex::connect(
+            ex::schedule(std::declval<Sched>()),
+            std::declval<resume_receiver>()));
+
+        op_t op_;
+
+        holder(Sched s, std::coroutine_handle<> h)
+            : op_(ex::connect(
+                ex::schedule(std::move(s)),
+                resume_receiver{this, h}))
+        {
+        }
+    };
+
+public:
+    explicit scheduler_executor(Sched sched)
+        : sched_(std::move(sched))
+    {
+    }
+
+    // Foreign schedulers have no capy execution_context; ops in
+    // the sender path must not rely on context services.
+    execution_context& context() const noexcept
+    {
+        static execution_context ctx;
+        return ctx;
+    }
+
+    void on_work_started() const noexcept {}
+    void on_work_finished() const noexcept {}
+
+    bool operator==(
+        scheduler_executor const& other) const noexcept
+    {
+        return sched_ == other.sched_;
+    }
+
+    void post(continuation& c) const
+    {
+        // the holder frees itself from the receiver on completion
+        auto* p = new holder(sched_, c.h);
+        ex::start(p->op_);
+    }
+
+    // A generic scheduler cannot answer "is the current thread
+    // yours", so dispatch degrades to post.
+    std::coroutine_handle<> dispatch(continuation& c) const
+    {
+        post(c);
+        return std::noop_coroutine();
+    }
+};
+
+template<class Aw, class Receiver>
+struct awaitable_op_state
+{
+    using operation_state_concept = ex::operation_state_t;
+    using result_type = awaitable_result_t<Aw>;
+
+    using env_type =
+        decltype(ex::get_env(std::declval<Receiver const&>()));
+    using stop_token_type =
+        typename env_stop_token<env_type>::type;
+
+    // io_env only carries std::stop_token, so any other
+    // stoppable token needs an owned std::stop_source relayed
+    // through a stop_callback; unstoppable tokens need neither.
+    // The stoppable_token conjunct keeps stop_callback_for_t
+    // instantiable in the bridged branch.
+    static constexpr bool token_passthrough =
+        std::is_same_v<stop_token_type, std::stop_token>;
+    static constexpr bool token_bridged =
+        !token_passthrough &&
+        ex::stoppable_token<stop_token_type> &&
+        !ex::unstoppable_token<stop_token_type>;
+
+    struct on_stop
+    {
+        std::stop_source& src;
+
+        void operator()() const noexcept
+        {
+            src.request_stop();
+        }
+    };
+
+    // stop_callback_for_t needs the token's callback_type alias,
+    // which std::stop_token lacks today; select the storage type
+    // lazily so only bridged tokens ever instantiate it.
+    struct no_stop_callback {};
+    static auto stop_cb_storage()
+    {
+        if constexpr (token_bridged)
+            return std::type_identity<
+                ex::stop_callback_for_t<stop_token_type, on_stop>>{};
+        else
+            return std::type_identity<no_stop_callback>{};
+    }
+
+    static constexpr bool has_io_executor =
+        requires(env_type const& e) { get_io_executor(e); };
+    static constexpr bool has_scheduler =
+        requires(env_type const& e) { ex::get_scheduler(e); };
+
+    // Lazy type selection, as with stop_cb_storage: the scheduler
+    // bridge type only exists for envs that need it.
+    struct no_scheduler_bridge {};
+    static auto sched_ex_storage()
+    {
+        if constexpr (!has_io_executor && has_scheduler)
+            return std::type_identity<scheduler_executor<
+                std::remove_cvref_t<decltype(ex::get_scheduler(
+                    std::declval<env_type const&>()))>>>{};
+        else
+            return std::type_identity<no_scheduler_bridge>{};
+    }
+
+    Aw aw_;
+    Receiver rcvr_;
+    io_env env_;
+    frame_cb cb_;
+    // stop_cb_ declared after stop_src_ so it is destroyed first:
+    // on_stop holds a reference to stop_src_.
+    std::stop_source stop_src_;
+    std::optional<
+        typename decltype(stop_cb_storage())::type>
+            stop_cb_;
+    // env_.executor may reference this member; op_states are
+    // immovable, so the address is stable.
+    std::optional<
+        typename decltype(sched_ex_storage())::type>
+            sched_ex_;
+
+    // std::stop_source allocates shared state; only pay for it
+    // when the receiver's token actually needs bridging.
+    static std::stop_source make_stop_source()
+    {
+        if constexpr (token_bridged)
+            return std::stop_source();
+        else
+            return std::stop_source(std::nostopstate);
+    }
+
+    awaitable_op_state(Aw aw, Receiver rcvr)
+        : aw_(std::move(aw))
+        , rcvr_(std::move(rcvr))
+        , cb_{}
+        , stop_src_(make_stop_source())
+    {
+    }
+
+    awaitable_op_state(awaitable_op_state const&) = delete;
+    awaitable_op_state(awaitable_op_state&&) = delete;
+    awaitable_op_state& operator=(awaitable_op_state const&) = delete;
+    awaitable_op_state& operator=(awaitable_op_state&&) = delete;
+
+    static void on_resume(frame_cb* p) noexcept
+    {
+        auto* self = static_cast<awaitable_op_state*>(p->data);
+        self->complete();
+    }
+
+    static void on_destroy(frame_cb*) noexcept
+    {
+    }
+
+    static constexpr bool nothrow_resume =
+        noexcept(std::declval<Aw&>().await_resume());
+
+    // A receiver connected against nothrow signatures has no
+    // exception_ptr overload, so the try/catch only exists when
+    // await_resume() can actually throw.
+    void complete() noexcept
+    {
+        if constexpr (nothrow_resume)
+        {
+            do_complete();
+        }
+        else
+        {
+            try
+            {
+                do_complete();
+            }
+            catch(...)
+            {
+                ex::set_error(
+                    std::move(rcvr_),
+                    std::current_exception());
+            }
+        }
+    }
+
+    void do_complete()
+    {
+        if constexpr (std::is_void_v<result_type>)
+        {
+            // void and plain-value results carry no error_code, so
+            // there is no in-band disposition; token state is the
+            // only cancellation signal available for these rows.
+            aw_.await_resume();
+            if(env_.stop_token.stop_requested())
+                ex::set_stopped(
+                    std::move(rcvr_));
+            else
+                ex::set_value(
+                    std::move(rcvr_));
+        }
+        else if constexpr (is_ec_outcome_v<result_type>)
+        {
+            // The op's own disposition picks the channel: a canceled
+            // completion surfaces as stopped even if the env token
+            // never fired, and a successful result survives a stop
+            // request that merely landed by completion time.
+            auto result = aw_.await_resume();
+            std::error_code ec;
+            if constexpr (std::is_same_v<
+                result_type, std::error_code>)
+                ec = result;
+            else
+                ec = get<0>(result);
+            // Success first: comparing ec against a condition goes
+            // through the category's virtual default_error_condition,
+            // which the hot path should not pay.
+            if(!ec)
+                ex::set_value(
+                    std::move(rcvr_));
+            else if(ec == std::errc::operation_canceled)
+                ex::set_stopped(
+                    std::move(rcvr_));
+            else
+                ex::set_error(
+                    std::move(rcvr_), ec);
+        }
+        else if constexpr (is_compound_ec_result_v<result_type>)
+        {
+            auto result = aw_.await_resume();
+            auto ec = get<0>(result);
+            if(ec)
+            {
+                if(ec == std::errc::operation_canceled)
+                    ex::set_stopped(
+                        std::move(rcvr_));
+                else
+                    ex::set_error(
+                        std::move(rcvr_), ec);
+            }
+            else
+            {
+                [&]<std::size_t... I>(std::index_sequence<I...>)
+                {
+                    ex::set_value(
+                        std::move(rcvr_),
+                        get<I + 1>(std::move(result))...);
+                }(std::make_index_sequence<
+                    std::tuple_size_v<result_type> - 1>{});
+            }
+        }
+        else
+        {
+            auto result = aw_.await_resume();
+            if(env_.stop_token.stop_requested())
+                ex::set_stopped(
+                    std::move(rcvr_));
+            else
+                ex::set_value(
+                    std::move(rcvr_),
+                    std::move(result));
+        }
+    }
+
+    void start() noexcept
+    {
+        auto renv = ex::get_env(rcvr_);
+
+        static_assert(has_io_executor || has_scheduler,
+            "the receiver's environment must provide "
+            "capy::get_io_executor or a scheduler via "
+            "std::execution::get_scheduler");
+
+        executor_ref io_ex = [&]() -> executor_ref
+        {
+            if constexpr (has_io_executor)
+            {
+                return get_io_executor(renv);
+            }
+            else
+            {
+                sched_ex_.emplace(ex::get_scheduler(renv));
+                return executor_ref(*sched_ex_);
+            }
+        }();
+
+        std::stop_token st;
+        if constexpr (token_passthrough)
+        {
+            st = renv.query(
+                ex::get_stop_token_t{});
+        }
+        else if constexpr (token_bridged)
+        {
+            stop_cb_.emplace(
+                renv.query(
+                    ex::get_stop_token_t{}),
+                on_stop{stop_src_});
+            st = stop_src_.get_token();
+        }
+        // Map the environment's allocator into io_env when it
+        // speaks pmr; other allocator types have no
+        // memory_resource to lend.
+        std::pmr::memory_resource* fa = nullptr;
+        if constexpr (requires { ex::get_allocator(renv); })
+        {
+            auto alloc = ex::get_allocator(renv);
+            if constexpr (is_polymorphic_allocator<
+                decltype(alloc)>::value)
+                fa = alloc.resource();
+        }
+
+        env_ = io_env{io_ex, st, fa};
+
+        if(aw_.await_ready())
+        {
+            complete();
+            return;
+        }
+
+        cb_.resume = &on_resume;
+        cb_.destroy = &on_destroy;
+        cb_.data = this;
+
+        auto h = std::coroutine_handle<>::from_address(
+            static_cast<void*>(&cb_));
+
+        // Not a real coroutine caller, so symmetric transfer
+        // must be driven by hand: any non-noop handle (our own
+        // frame on immediate completion, or a wrapped task's
+        // handle that still needs to run) has to be resumed
+        // explicitly or nothing ever completes.
+        auto resumed = detail::call_await_suspend(&aw_, h, &env_);
+        if(resumed != std::noop_coroutine())
+            resumed.resume();
+    }
+};
+
+} // namespace detail
+
+} // namespace boost::capy
+
+#endif
diff --git a/example/awaitable-sender/read_op.hpp b/example/awaitable-sender/read_op.hpp
new file mode 100644
index 000000000..738d79d22
--- /dev/null
+++ b/example/awaitable-sender/read_op.hpp
@@ -0,0 +1,135 @@
+//
+// Copyright (c) 2026 Steve Gerbino
+//
+// 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)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#ifndef BOOST_CAPY_EXAMPLE_READ_OP_HPP
+#define BOOST_CAPY_EXAMPLE_READ_OP_HPP
+
+#include "awaitable_sender_base.hpp"
+
+#include <boost/capy/continuation.hpp>
+#include <boost/capy/ex/async_waker.hpp>
+#include <boost/capy/ex/io_env.hpp>
+#include <boost/capy/io_result.hpp>
+#include <boost/capy/task.hpp>
+
+#include <coroutine>
+#include <cstddef>
+#include <optional>
+#include <system_error>
+#include <utility>
+
+namespace boost::capy {
+
+/** Mock I/O op with a scripted outcome, modeled on the real
+    corosio op shape: starts in `await_suspend`, completes
+    through the environment's executor, and is a sender via
+    @ref awaitable_sender_base. Its stop-wait mode composes
+    `async_waker` rather than hand-rolling stop-callback arming:
+    the waker's arbiter already resolves the races between
+    arming, stop requests, and completion.
+*/
+class read_op : public awaitable_sender_base<read_op>
+{
+    // Data members are private so the op is invisible to
+    // std::execution sender decomposition: tag_of_t claims any
+    // type whose members admit a structured binding, and
+    // inaccessible members make that binding ill-formed (it also
+    // keeps the type a non-aggregate, which is the condition
+    // today's implementations actually probe).
+    std::error_code ec_{};
+    std::size_t n_ = 0;
+    bool immediate_ = false;
+
+    io_env const* env_ = nullptr;
+    continuation cont_{};
+
+    // Stop-wait mode: a coroutine owning a waker nobody wakes, so
+    // it completes only when the environment's stop token fires
+    // (with error::canceled). The waker lives in the coroutine
+    // frame; the task is movable, keeping read_op movable.
+    std::optional<task<io_result<>>> stop_task_;
+
+    static task<io_result<>> stop_wait()
+    {
+        async_waker waker;
+        co_return co_await waker.wait();
+    }
+
+public:
+    /// Construct an op that completes successfully with zero bytes.
+    read_op() = default;
+
+    // Scripted construction goes through factories, not multi-arg
+    // constructors — a scaffolding accommodation, unnecessary at
+    // graduation: the bundled pre-standard implementation probes
+    // decomposable senders by brace-initializing with 2..6
+    // placeholder arguments, and any constructor of that arity
+    // matches the probe and re-enters its decomposition machinery.
+
+    /// Create an op with a scripted `(ec, n)` completion.
+    static read_op result(std::error_code ec, std::size_t n)
+    {
+        read_op op;
+        op.ec_ = ec;
+        op.n_ = n;
+        return op;
+    }
+
+    /// Create an op that completes inline via `await_ready`.
+    static read_op immediate(std::error_code ec, std::size_t n)
+    {
+        read_op op = result(ec, n);
+        op.immediate_ = true;
+        return op;
+    }
+
+    /// Create an op that never self-completes; it finishes only
+    /// on a stop request.
+    static read_op waits_for_stop()
+    {
+        read_op op;
+        op.stop_task_ = stop_wait();
+        return op;
+    }
+
+    bool await_ready() const noexcept
+    {
+        return immediate_;
+    }
+
+    std::coroutine_handle<> await_suspend(
+        std::coroutine_handle<> h,
+        io_env const* env)
+    {
+        if(stop_task_)
+            return stop_task_->await_suspend(h, env);
+        env_ = env;
+        cont_ = continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    io_result<std::size_t> await_resume() noexcept
+    {
+        // The op reports its own disposition in-band: a stop-wait
+        // ends with the waker's error::canceled, which the sender
+        // machinery routes to set_stopped() and a co_await caller
+        // observes as operation_canceled.
+        if(stop_task_)
+        {
+            auto [ec] = stop_task_->await_resume();
+            return {ec ? ec : ec_, n_};
+        }
+        return {ec_, n_};
+    }
+};
+
+} // namespace boost::capy
+
+#endif
diff --git a/example/awaitable-sender/tests.cpp b/example/awaitable-sender/tests.cpp
new file mode 100644
index 000000000..dc22f1c29
--- /dev/null
+++ b/example/awaitable-sender/tests.cpp
@@ -0,0 +1,894 @@
+//
+// Copyright (c) 2026 Steve Gerbino
+//
+// 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)
+//
+// Official repository: https://github.com/cppalliance/capy
+//
+
+#include "awaitable_sender.hpp"
+#include "awaitable_sender_base.hpp"
+#include "read_op.hpp"
+
+#include <boost/capy.hpp>
+
+#include <beman/execution/execution.hpp>
+
+#include <chrono>
+#include <coroutine>
+#include <cstddef>
+#include <cstdlib>
+#include <iostream>
+#include <latch>
+#include <stdexcept>
+#include <stop_token>
+#include <system_error>
+#include <thread>
+#include <memory_resource>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+
+namespace capy = boost::capy;
+namespace ex = beman::execution;
+
+static int failures = 0;
+
+#define CHECK(cond) \
+    do { \
+        if(!(cond)) \
+        { \
+            ++failures; \
+            std::cerr << "FAIL " << __FILE__ << ":" \
+                << __LINE__ << ": " #cond "\n"; \
+        } \
+    } while(0)
+
+enum class channel { none, value, error, stopped };
+
+struct test_outcome
+{
+    channel ch = channel::none;
+    std::size_t n = 0;
+    int i = 0;
+    std::error_code ec{};
+};
+
+template<class Token>
+struct test_env
+{
+    capy::executor_ref ex_;
+    Token tok_;
+
+    auto query(capy::get_io_executor_t const&) const noexcept
+        -> capy::executor_ref
+    {
+        return ex_;
+    }
+
+    auto query(ex::get_stop_token_t const&) const noexcept
+        -> Token
+    {
+        return tok_;
+    }
+};
+
+template<class Token>
+struct test_receiver
+{
+    using receiver_concept = ex::receiver_t;
+
+    test_env<Token> env_;
+    test_outcome* out_;
+    std::latch* done_;
+
+    auto get_env() const noexcept -> test_env<Token>
+    {
+        return env_;
+    }
+
+    void set_value() && noexcept
+    {
+        out_->ch = channel::value;
+        done_->count_down();
+    }
+
+    void set_value(std::size_t n) && noexcept
+    {
+        out_->ch = channel::value;
+        out_->n = n;
+        done_->count_down();
+    }
+
+    void set_value(int i) && noexcept
+    {
+        out_->ch = channel::value;
+        out_->i = i;
+        done_->count_down();
+    }
+
+    void set_error(std::error_code ec) && noexcept
+    {
+        out_->ch = channel::error;
+        out_->ec = ec;
+        done_->count_down();
+    }
+
+    void set_error(std::exception_ptr) && noexcept
+    {
+        out_->ch = channel::error;
+        done_->count_down();
+    }
+
+    void set_stopped() && noexcept
+    {
+        out_->ch = channel::stopped;
+        done_->count_down();
+    }
+};
+
+// Connect, start, and wait for any sender against a fresh
+// single-thread pool; returns the observed completion.
+template<class Sender, class Token = std::stop_token>
+test_outcome run(Sender&& sndr, Token tok = {})
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    std::latch done(1);
+    test_outcome out;
+    auto op = ex::connect(
+        std::forward<Sender>(sndr),
+        test_receiver<Token>{
+            {pool_ex, tok}, &out, &done});
+    ex::start(op);
+    done.wait();
+    return out;
+}
+
+// Poll a latch with a deadline so a cancellation regression fails
+// the suite instead of hanging it. Returns false on timeout; the
+// caller must then exit the process, because abandoning a pending
+// operation whose state lives on its stack would be use-after-free.
+bool await_latch(
+    std::latch& l,
+    std::chrono::milliseconds limit)
+{
+    auto deadline = std::chrono::steady_clock::now() + limit;
+    while(!l.try_wait())
+    {
+        if(std::chrono::steady_clock::now() > deadline)
+            return false;
+        std::this_thread::sleep_for(
+            std::chrono::milliseconds(1));
+    }
+    return true;
+}
+
+#define WAIT_OR_DIE(latch) \
+    do { \
+        if(!await_latch(latch, std::chrono::seconds(2))) \
+        { \
+            ++failures; \
+            std::cerr << "FAIL " << __FILE__ << ":" \
+                << __LINE__ \
+                << ": timeout waiting for completion\n"; \
+            std::_Exit(1); \
+        } \
+    } while(0)
+
+// Mock IoAwaitable with a scripted error_code result,
+// completing through the environment's executor.
+struct ec_op
+{
+    std::error_code ec_{};
+
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    std::error_code await_resume() noexcept { return ec_; }
+};
+
+// Mock IoAwaitable with a scripted compound (ec, n) result.
+struct compound_op
+{
+    std::error_code ec_{};
+    std::size_t n_ = 0;
+
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    capy::io_result<std::size_t> await_resume() noexcept
+    {
+        return {ec_, n_};
+    }
+};
+
+// Mock IoAwaitable that completes with no payload at all
+// (the `void` row of make_sigs, distinct from io_result<>'s
+// 1-tuple-ec row exercised by async_waker).
+struct void_op
+{
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    void await_resume() noexcept {}
+};
+
+// Mock IoAwaitable returning a plain value type, unrelated to
+// error_code (the "other T" row of make_sigs).
+struct value_op
+{
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    int await_resume() noexcept { return 17; }
+};
+
+// Mock IoAwaitable whose await_resume() can throw, exercising
+// awaitable_op_state's conditional try/catch -> set_error(exception_ptr).
+struct throwing_op
+{
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    std::size_t await_resume()
+    {
+        throw std::runtime_error("throwing_op");
+    }
+};
+
+void test_compound_success()
+{
+    auto out = run(capy::as_sender(
+        compound_op{.ec_ = {}, .n_ = 42}));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 42);
+}
+
+void test_compound_error()
+{
+    auto expected = std::make_error_code(
+        std::errc::connection_reset);
+    auto out = run(capy::as_sender(
+        compound_op{.ec_ = expected, .n_ = 3}));
+    CHECK(out.ch == channel::error);
+    CHECK(out.ec == expected);
+}
+
+void test_canceled_ec_routes_stopped()
+{
+    // A canceled disposition surfaces on the stopped channel even
+    // though the environment's token was never triggered.
+    auto out = run(capy::as_sender(compound_op{
+        .ec_ = std::make_error_code(
+            std::errc::operation_canceled),
+        .n_ = 0}));
+    CHECK(out.ch == channel::stopped);
+}
+
+void test_value_survives_racing_stop()
+{
+    // The op ignores tokens and completes with a value; a stop
+    // request that merely landed by completion time must not
+    // discard the result.
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    ex::inplace_stop_source ss;
+    ss.request_stop();
+    std::latch done(1);
+    test_outcome out;
+    auto op = ex::connect(
+        capy::as_sender(compound_op{.ec_ = {}, .n_ = 7}),
+        test_receiver<ex::inplace_stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    ex::start(op);
+    done.wait();
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 7);
+}
+
+void test_empty_io_result_success()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    capy::async_waker waker;
+    std::latch done(1);
+    test_outcome out;
+    auto op = ex::connect(
+        capy::as_sender(waker.wait()),
+        test_receiver<std::stop_token>{
+            {pool_ex, {}}, &out, &done});
+    ex::start(op);
+    waker.wake();
+    done.wait();
+    CHECK(out.ch == channel::value);
+}
+
+void test_void_op_success()
+{
+    auto out = run(capy::as_sender(void_op{}));
+    CHECK(out.ch == channel::value);
+}
+
+void test_value_op_success()
+{
+    auto out = run(capy::as_sender(value_op{}));
+    CHECK(out.ch == channel::value);
+    CHECK(out.i == 17);
+}
+
+void test_throwing_op_error()
+{
+    auto out = run(capy::as_sender(throwing_op{}));
+    CHECK(out.ch == channel::error);
+}
+
+void test_ec_success()
+{
+    auto out = run(capy::as_sender(ec_op{}));
+    CHECK(out.ch == channel::value);
+}
+
+void test_ec_error()
+{
+    auto expected = std::make_error_code(
+        std::errc::connection_reset);
+    auto out = run(capy::as_sender(ec_op{expected}));
+    CHECK(out.ch == channel::error);
+    CHECK(out.ec == expected);
+}
+
+void test_cancel_std_token()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    capy::async_waker waker;
+    std::stop_source ss;
+    std::latch done(1);
+    test_outcome out;
+    auto op = ex::connect(
+        capy::as_sender(waker.wait()),
+        test_receiver<std::stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    ex::start(op);
+    ss.request_stop();
+    WAIT_OR_DIE(done);
+    CHECK(out.ch == channel::stopped);
+}
+
+void test_cancel_inplace_token()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    capy::async_waker waker;
+    ex::inplace_stop_source ss;
+    std::latch done(1);
+    test_outcome out;
+
+    // Rescue thread: if the bridge is broken the waker never
+    // observes the stop; wake it late so the test fails on the
+    // channel check instead of hanging.
+    std::thread rescue([&] {
+        std::this_thread::sleep_for(
+            std::chrono::milliseconds(200));
+        waker.wake();
+    });
+
+    auto op = ex::connect(
+        capy::as_sender(waker.wait()),
+        test_receiver<ex::inplace_stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    ex::start(op);
+    ss.request_stop();
+    done.wait();
+    rescue.join();
+    CHECK(out.ch == channel::stopped);
+}
+
+void test_then_pipeline()
+{
+    auto out = run(ex::then(
+        capy::as_sender(compound_op{.ec_ = {}, .n_ = 41}),
+        [](std::size_t n) { return n + 1; }));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 42);
+}
+
+void test_read_op_native_success()
+{
+    auto out = run(capy::read_op::result({}, 7));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 7);
+}
+
+void test_read_op_native_error()
+{
+    auto expected = std::make_error_code(
+        std::errc::broken_pipe);
+    auto out = run(capy::read_op::result(expected, 0));
+    CHECK(out.ch == channel::error);
+    CHECK(out.ec == expected);
+}
+
+void test_read_op_immediate()
+{
+    auto out = run(capy::read_op::immediate({}, 9));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 9);
+}
+
+void test_read_op_stopped()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    ex::inplace_stop_source ss;
+    std::latch done(1);
+    test_outcome out;
+    auto o = ex::connect(
+        capy::read_op::waits_for_stop(),
+        test_receiver<ex::inplace_stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    ex::start(o);
+    ss.request_stop();
+    WAIT_OR_DIE(done);
+    CHECK(out.ch == channel::stopped);
+}
+
+void test_read_op_stopped_before_start()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    ex::inplace_stop_source ss;
+    std::latch done(1);
+    test_outcome out;
+    auto o = ex::connect(
+        capy::read_op::waits_for_stop(),
+        test_receiver<ex::inplace_stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    // Stop before start: exercises read_op's pre-check, which
+    // must resume inline instead of arming a stop_callback that
+    // would fire synchronously during construction.
+    ss.request_stop();
+    ex::start(o);
+    WAIT_OR_DIE(done);
+    CHECK(out.ch == channel::stopped);
+}
+
+void test_read_op_through_adaptor()
+{
+    auto out = run(capy::as_sender(capy::read_op::result({}, 5)));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 5);
+}
+
+// Symmetric with test_cancel_std_token, but for the native
+// awaitable_sender_base path (token_passthrough) instead of as_sender.
+void test_read_op_cancel_std_token()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    std::stop_source ss;
+    std::latch done(1);
+    test_outcome out;
+    auto o = ex::connect(
+        capy::read_op::waits_for_stop(),
+        test_receiver<std::stop_token>{
+            {pool_ex, ss.get_token()}, &out, &done});
+    ex::start(o);
+    ss.request_stop();
+    WAIT_OR_DIE(done);
+    CHECK(out.ch == channel::stopped);
+}
+
+// Symmetric with test_read_op_immediate, but through as_sender().
+void test_adaptor_immediate()
+{
+    auto out = run(capy::as_sender(
+        capy::read_op::immediate({}, 3)));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 3);
+}
+
+void test_read_op_then_pipeline()
+{
+    auto out = run(ex::then(
+        capy::read_op::result({}, 41),
+        [](std::size_t n) { return n + 1; }));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 42);
+}
+
+// sync_wait's environment provides get_scheduler but neither
+// get_io_executor nor a stop token; these exercise the scheduler
+// bridge end-to-end on the run_loop.
+void test_sync_wait_value()
+{
+    auto r = ex::sync_wait(capy::read_op::result({}, 7));
+    CHECK(r.has_value());
+    CHECK(std::get<0>(*r) == 7);
+}
+
+void test_sync_wait_pipeline()
+{
+    auto r = ex::sync_wait(ex::then(
+        capy::read_op::result({}, 41),
+        [](std::size_t n) { return n + 1; }));
+    CHECK(r.has_value());
+    CHECK(std::get<0>(*r) == 42);
+}
+
+void test_sync_wait_error_throws()
+{
+    bool caught = false;
+    try
+    {
+        (void)ex::sync_wait(capy::read_op::result(
+            std::make_error_code(std::errc::broken_pipe), 0));
+    }
+    catch(...)
+    {
+        caught = true;
+    }
+    CHECK(caught);
+}
+
+// Records the frame allocator the machinery hands to the op.
+struct alloc_probe_op
+{
+    std::pmr::memory_resource** out_;
+
+    capy::io_env const* env_ = nullptr;
+    capy::continuation cont_{};
+
+    bool await_ready() const noexcept { return false; }
+
+    auto await_suspend(
+        std::coroutine_handle<> h,
+        capy::io_env const* env)
+    {
+        *out_ = env->frame_allocator;
+        env_ = env;
+        cont_ = capy::continuation{h};
+        env_->executor.post(cont_);
+        return std::noop_coroutine();
+    }
+
+    std::error_code await_resume() noexcept { return {}; }
+};
+
+struct pmr_env
+{
+    capy::executor_ref ex_;
+    std::pmr::memory_resource* mr_;
+
+    auto query(capy::get_io_executor_t const&) const noexcept
+        -> capy::executor_ref
+    {
+        return ex_;
+    }
+
+    auto query(ex::get_allocator_t const&) const noexcept
+        -> std::pmr::polymorphic_allocator<>
+    {
+        return std::pmr::polymorphic_allocator<>(mr_);
+    }
+};
+
+struct pmr_receiver
+{
+    using receiver_concept = ex::receiver_t;
+
+    pmr_env env_;
+    std::latch* done_;
+
+    auto get_env() const noexcept -> pmr_env { return env_; }
+
+    void set_value() && noexcept { done_->count_down(); }
+    void set_error(std::error_code) && noexcept
+    {
+        done_->count_down();
+    }
+    void set_error(std::exception_ptr) && noexcept
+    {
+        done_->count_down();
+    }
+    void set_stopped() && noexcept { done_->count_down(); }
+};
+
+void test_env_allocator_reaches_io_env()
+{
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    std::pmr::monotonic_buffer_resource mr;
+    std::pmr::memory_resource* seen = nullptr;
+    std::latch done(1);
+    auto op = ex::connect(
+        capy::as_sender(alloc_probe_op{&seen}),
+        pmr_receiver{{pool_ex, &mr}, &done});
+    ex::start(op);
+    done.wait();
+    CHECK(seen == &mr);
+}
+
+// Cross-protocol consistency: drive the SAME op through co_await
+// and through connect/start, and require identical outcomes. The
+// dual-protocol contract (see awaitable_sender_base) makes
+// divergence impossible when the op reports its disposition
+// in-band; these tests pin that.
+capy::task<void> record_outcome(
+    capy::read_op op,
+    test_outcome* out,
+    std::latch* done)
+{
+    auto [ec, n] = co_await std::move(op);
+    out->ec = ec;
+    out->n = n;
+    out->ch = !ec ? channel::value
+        : ec == std::errc::operation_canceled
+            ? channel::stopped
+            : channel::error;
+    done->count_down();
+}
+
+void test_coawait_matches_sender_success()
+{
+    auto sender_out = run(capy::read_op::result({}, 7));
+
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    test_outcome coawait_out;
+    std::latch done(1);
+    capy::run_async(pool_ex)(record_outcome(
+        capy::read_op::result({}, 7), &coawait_out, &done));
+    WAIT_OR_DIE(done);
+
+    CHECK(coawait_out.ch == channel::value);
+    CHECK(sender_out.ch == coawait_out.ch);
+    CHECK(sender_out.n == coawait_out.n);
+}
+
+void test_coawait_matches_sender_error()
+{
+    auto expected = std::make_error_code(
+        std::errc::broken_pipe);
+    auto sender_out = run(capy::read_op::result(expected, 0));
+
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    test_outcome coawait_out;
+    std::latch done(1);
+    capy::run_async(pool_ex)(record_outcome(
+        capy::read_op::result(expected, 0),
+        &coawait_out, &done));
+    WAIT_OR_DIE(done);
+
+    CHECK(coawait_out.ch == channel::error);
+    CHECK(sender_out.ch == coawait_out.ch);
+    CHECK(sender_out.ec == coawait_out.ec);
+}
+
+void test_coawait_stopped_before_start()
+{
+    // co_await twin of test_read_op_stopped_before_start: the
+    // pre-stopped token must yield a canceled disposition.
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    std::stop_source ss;
+    ss.request_stop();
+    test_outcome coawait_out;
+    std::latch done(1);
+    capy::run_async(pool_ex, ss.get_token())(record_outcome(
+        capy::read_op::waits_for_stop(),
+        &coawait_out, &done));
+    WAIT_OR_DIE(done);
+    CHECK(coawait_out.ch == channel::stopped);
+}
+
+void test_coawait_stopped_midflight()
+{
+    // co_await twin of test_read_op_stopped.
+    capy::thread_pool pool(1);
+    auto pool_ex = pool.get_executor();
+    std::stop_source ss;
+    test_outcome coawait_out;
+    std::latch done(1);
+    capy::run_async(pool_ex, ss.get_token())(record_outcome(
+        capy::read_op::waits_for_stop(),
+        &coawait_out, &done));
+    ss.request_stop();
+    WAIT_OR_DIE(done);
+    CHECK(coawait_out.ch == channel::stopped);
+}
+
+// Channel splitting is keyed on io_result, not tuple shape: a
+// std::tuple that happens to lead with error_code is a value.
+struct tuple_result_op
+{
+    bool await_ready() const noexcept { return false; }
+    auto await_suspend(
+        std::coroutine_handle<>, capy::io_env const*)
+    {
+        return std::noop_coroutine();
+    }
+    std::tuple<std::error_code, int> await_resume() noexcept
+    {
+        return {};
+    }
+};
+static_assert(std::is_same_v<
+    decltype(capy::detail::make_sigs<tuple_result_op>()),
+    ex::completion_signatures<
+        ex::set_value_t(std::tuple<std::error_code, int>),
+        ex::set_stopped_t()>>);
+
+// AwaitableSender partitions the world correctly: read_op models
+// both protocols; compound_op is awaitable-only; the as_sender
+// wrapper is sender-only (no await_suspend).
+static_assert(capy::AwaitableSender<capy::read_op>);
+static_assert(capy::IoAwaitable<compound_op> &&
+    !capy::AwaitableSender<compound_op>);
+static_assert(
+    !capy::IoAwaitable<capy::awaitable_sender<compound_op>> &&
+    !capy::AwaitableSender<capy::awaitable_sender<compound_op>>);
+
+// ensure_sender: identity for dual-protocol ops, lifting for
+// awaitable-only ops.
+static_assert(std::is_same_v<
+    decltype(capy::ensure_sender(
+        capy::read_op::result({}, 0))),
+    capy::read_op>);
+static_assert(std::is_same_v<
+    decltype(capy::ensure_sender(compound_op{})),
+    capy::awaitable_sender<compound_op>>);
+
+void test_ensure_sender_passthrough()
+{
+    auto out = run(capy::ensure_sender(
+        capy::read_op::result({}, 7)));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 7);
+}
+
+void test_ensure_sender_lifts()
+{
+    auto out = run(capy::ensure_sender(
+        compound_op{.ec_ = {}, .n_ = 5}));
+    CHECK(out.ch == channel::value);
+    CHECK(out.n == 5);
+}
+
+// beman never calls the C++26 static-template signature query, so
+// nothing else instantiates its body; pin it here so it stays
+// callable in constant evaluation and agrees with the beman-compat
+// instance form.
+static_assert(std::is_same_v<
+    decltype(capy::read_op::
+        get_completion_signatures<capy::read_op>()),
+    decltype(std::declval<capy::read_op const&>()
+        .get_completion_signatures(0))>);
+
+// Compile-fail probe: uncomment to verify the non-aggregate
+// static_assert in awaitable_sender_base fires. (Completion
+// signatures are deduced from await_resume(), so a signature
+// mismatch is no longer expressible; the aggregate requirement
+// is the remaining connect-time check.)
+// Expected diagnostic: "Derived must declare a constructor".
+//
+// struct aggregate_op
+//     : capy::awaitable_sender_base<aggregate_op>
+// {
+//     bool await_ready() const noexcept { return false; }
+//     auto await_suspend(
+//         std::coroutine_handle<>, capy::io_env const*)
+//     {
+//         return std::noop_coroutine();
+//     }
+//     capy::io_result<std::size_t> await_resume() noexcept
+//     {
+//         return {};
+//     }
+// };
+// void probe() { (void)run(aggregate_op{}); }
+
+int main()
+{
+    test_empty_io_result_success();
+    test_void_op_success();
+    test_value_op_success();
+    test_throwing_op_error();
+    test_ec_success();
+    test_ec_error();
+    test_compound_success();
+    test_compound_error();
+    test_canceled_ec_routes_stopped();
+    test_value_survives_racing_stop();
+    test_cancel_std_token();
+    test_cancel_inplace_token();
+    test_then_pipeline();
+    test_read_op_native_success();
+    test_read_op_native_error();
+    test_read_op_immediate();
+    test_read_op_stopped();
+    test_read_op_stopped_before_start();
+    test_read_op_through_adaptor();
+    test_read_op_then_pipeline();
+    test_sync_wait_value();
+    test_sync_wait_pipeline();
+    test_sync_wait_error_throws();
+    test_env_allocator_reaches_io_env();
+    test_coawait_matches_sender_success();
+    test_coawait_matches_sender_error();
+    test_coawait_stopped_before_start();
+    test_coawait_stopped_midflight();
+    test_ensure_sender_passthrough();
+    test_ensure_sender_lifts();
+    test_read_op_cancel_std_token();
+    test_adaptor_immediate();
+
+    if(failures == 0)
+        std::cout << "all tests passed\n";
+    else
+        std::cerr << failures << " test(s) failed\n";
+    return failures == 0 ? 0 : 1;
+}