From 72e93c8144b54e3d01289f910515ce8a73b8cb54 Mon Sep 17 00:00:00 2001
From: Eric Niebler <eniebler@nvidia.com>
Date: Thu, 5 Mar 2026 10:11:40 -0800
Subject: [PATCH] change nvexec::let_* to tell the secondary sender where it is
 executing

---
 include/nvexec/stream/common.cuh        | 200 +++++++++--------
 include/nvexec/stream/let_xxx.cuh       | 280 ++++++++++++++----------
 include/nvexec/stream/repeat_n.cuh      |  41 ++--
 include/nvexec/stream/schedule_from.cuh |   2 +-
 include/stdexec/__detail/__meta.hpp     |   6 +
 5 files changed, 303 insertions(+), 226 deletions(-)

diff --git a/include/nvexec/stream/common.cuh b/include/nvexec/stream/common.cuh
index ed014a478..0bb263cef 100644
--- a/include/nvexec/stream/common.cuh
+++ b/include/nvexec/stream/common.cuh
@@ -488,27 +488,31 @@ namespace nv::execution
     template <class Env, class Variant>
     struct stream_enqueue_receiver
     {
-      Env*              env_;
-      Variant*          variant_;
-      queue::task_base* task_;
-      queue::producer   producer_;
-
-     public:
       using receiver_concept = STDEXEC::receiver_t;
 
+      explicit stream_enqueue_receiver(Env const *       env,
+                                       Variant*          variant,
+                                       queue::task_base* task,
+                                       queue::producer   producer)
+        : env_(env)
+        , variant_(variant)
+        , task_(task)
+        , producer_(producer)
+      {}
+
       template <class... Args>
       STDEXEC_ATTRIBUTE(host, device)
       void set_value(Args&&... args) noexcept
       {
-        variant_->template emplace<decayed_tuple_t<set_value_t, Args...>>(set_value_t(),
-                                                                          static_cast<Args&&>(
-                                                                            args)...);
+        using tuple_t = decayed_tuple_t<set_value_t, Args...>;
+        variant_->template emplace<tuple_t>(set_value_t(), static_cast<Args&&>(args)...);
         producer_(task_);
       }
 
       STDEXEC_ATTRIBUTE(host, device) void set_stopped() noexcept
       {
-        variant_->template emplace<decayed_tuple_t<set_stopped_t>>(set_stopped_t());
+        using tuple_t = decayed_tuple_t<set_stopped_t>;
+        variant_->template emplace<tuple_t>(set_stopped_t());
         producer_(task_);
       }
 
@@ -519,14 +523,13 @@ namespace nv::execution
         if constexpr (__decays_to<Error, std::exception_ptr>)
         {
           // What is `exception_ptr` but death pending
-          variant_->template emplace<decayed_tuple_t<set_error_t, cudaError_t>>(STDEXEC::set_error,
-                                                                                cudaErrorUnknown);
+          using tuple_t = decayed_tuple_t<set_error_t, cudaError_t>;
+          variant_->template emplace<tuple_t>(STDEXEC::set_error, cudaErrorUnknown);
         }
         else
         {
-          variant_->template emplace<decayed_tuple_t<set_error_t, Error>>(set_error_t(),
-                                                                          static_cast<Error&&>(
-                                                                            err));
+          using tuple_t = decayed_tuple_t<set_error_t, Error>;
+          variant_->template emplace<tuple_t>(set_error_t(), static_cast<Error&&>(err));
         }
         producer_(task_);
       }
@@ -536,15 +539,11 @@ namespace nv::execution
         return *env_;
       }
 
-      stream_enqueue_receiver(Env*              env,
-                              Variant*          variant,
-                              queue::task_base* task,
-                              queue::producer   producer)
-        : env_(env)
-        , variant_(variant)
-        , task_(task)
-        , producer_(producer)
-      {}
+     private:
+      Env const *       env_;
+      Variant*          variant_;
+      queue::task_base* task_;
+      queue::producer   producer_;
     };
 
     template <class Receiver, class... Args, class Tag>
@@ -558,16 +557,10 @@ namespace nv::execution
     template <class Receiver, class Variant>
     struct continuation_task : queue::task_base
     {
-      Receiver                   rcvr_;
-      Variant*                   variant_;
-      cudaStream_t               stream_{};
-      std::pmr::memory_resource* pinned_resource_{};
-      cudaError_t                status_{cudaSuccess};
-
-      continuation_task(Receiver                   rcvr,
-                        Variant*                   variant,
-                        cudaStream_t               stream,
-                        std::pmr::memory_resource* pinned_resource) noexcept
+      explicit continuation_task(Receiver                   rcvr,
+                                 Variant*                   variant,
+                                 cudaStream_t               stream,
+                                 std::pmr::memory_resource* pinned_resource) noexcept
         : rcvr_{rcvr}
         , variant_{variant}
         , stream_{stream}
@@ -606,6 +599,18 @@ namespace nv::execution
           status_ = STDEXEC_LOG_CUDA_API(cudaMemsetAsync(this->atom_next_, 0, ptr_size, stream_));
         }
       }
+
+      cudaError_t status() const noexcept
+      {
+        return status_;
+      }
+
+     private:
+      Receiver                   rcvr_;
+      Variant*                   variant_;
+      cudaStream_t               stream_{};
+      std::pmr::memory_resource* pinned_resource_{};
+      cudaError_t                status_{cudaSuccess};
     };
 
     template <class Env>
@@ -695,6 +700,7 @@ namespace nv::execution
         }
       }
 
+      [[nodiscard]]
       auto make_env() const noexcept -> env_t
       {
         return make_stream_env(get_env(rcvr_), get_stream_provider());
@@ -738,10 +744,12 @@ namespace nv::execution
       stream_provider stream_provider_;
     };
 
-    template <class OuterReceiver>
+    template <class OpState, class Env = decltype(__declval<OpState&>().make_env())>
     struct propagate_receiver : stream_receiver_base
     {
-      opstate_base<OuterReceiver>& opstate_;
+      explicit propagate_receiver(OpState& opstate) noexcept
+        : opstate_(opstate)
+      {}
 
       template <class... Args>
       void set_value(Args&&... args) noexcept
@@ -760,10 +768,14 @@ namespace nv::execution
         opstate_.propagate_completion_signal(set_stopped_t());
       }
 
-      auto get_env() const noexcept -> decltype(auto)
+      [[nodiscard]]
+      auto get_env() const noexcept -> Env
       {
         return opstate_.make_env();
       }
+
+     private:
+      OpState& opstate_;
     };
 
     template <class CvSender, class InnerReceiver, class OuterReceiver>
@@ -780,38 +792,6 @@ namespace nv::execution
         __if_c<stream_sender<CvSender, env_t>, InnerReceiver, stream_enqueue_receiver_t>;
       using inner_opstate_t = connect_result_t<CvSender, intermediate_receiver_t>;
 
-      void start() & noexcept
-      {
-        started_.test_and_set(::cuda::std::memory_order::relaxed);
-
-        if (this->stream_provider_.status_ != cudaSuccess)
-        {
-          // Couldn't allocate memory for opstate state, complete with error
-          this->propagate_completion_signal(STDEXEC::set_error,
-                                            std::move(this->stream_provider_.status_));
-          return;
-        }
-
-        if constexpr (stream_receiver<InnerReceiver>)
-        {
-          if (InnerReceiver::memory_allocation_size())
-          {
-            STDEXEC_TRY
-            {
-              this->temp_storage_ = this->ctx_.managed_resource_->allocate(
-                InnerReceiver::memory_allocation_size());
-            }
-            STDEXEC_CATCH_ALL
-            {
-              this->propagate_completion_signal(STDEXEC::set_error, cudaErrorMemoryAllocation);
-              return;
-            }
-          }
-        }
-
-        STDEXEC::start(inner_op_);
-      }
-
       template <class ReceiverProvider>
         requires stream_sender<CvSender, env_t>
       opstate(CvSender&&       sender,
@@ -846,7 +826,7 @@ namespace nv::execution
       {
         if (this->stream_provider_.status_ == cudaSuccess)
         {
-          this->stream_provider_.status_ = task_->status_;
+          this->stream_provider_.status_ = task_->status();
         }
       }
 
@@ -870,6 +850,39 @@ namespace nv::execution
 
       STDEXEC_IMMOVABLE(opstate);
 
+      void start() & noexcept
+      {
+        started_.test_and_set(::cuda::std::memory_order::relaxed);
+
+        if (this->stream_provider_.status_ != cudaSuccess)
+        {
+          // Couldn't allocate memory for opstate state, complete with error
+          this->propagate_completion_signal(STDEXEC::set_error,
+                                            std::move(this->stream_provider_.status_));
+          return;
+        }
+
+        if constexpr (stream_receiver<InnerReceiver>)
+        {
+          if (InnerReceiver::memory_allocation_size())
+          {
+            STDEXEC_TRY
+            {
+              this->temp_storage_ = this->ctx_.managed_resource_->allocate(
+                InnerReceiver::memory_allocation_size());
+            }
+            STDEXEC_CATCH_ALL
+            {
+              this->propagate_completion_signal(STDEXEC::set_error, cudaErrorMemoryAllocation);
+              return;
+            }
+          }
+        }
+
+        STDEXEC::start(inner_op_);
+      }
+
+     private:
       host_ptr_t<variant_t>        storage_;
       task_t*                      task_{};
       ::cuda::std::atomic_flag     started_{};
@@ -880,26 +893,26 @@ namespace nv::execution
     template <class CvSender, class OuterReceiver>
       requires stream_receiver<OuterReceiver>
     using exit_opstate_t =
-      _strm::opstate<CvSender, propagate_receiver<OuterReceiver>, OuterReceiver>;
+      _strm::opstate<CvSender, propagate_receiver<opstate_base<OuterReceiver>>, OuterReceiver>;
 
-    template <class Sender, class OuterReceiver>
-    auto exit_opstate(Sender&& sndr, OuterReceiver rcvr, context ctx) noexcept
-      -> exit_opstate_t<Sender, OuterReceiver>
+    template <class CvSender, class OuterReceiver>
+    auto exit_opstate(CvSender&& sndr, OuterReceiver rcvr, context ctx) noexcept
+      -> exit_opstate_t<CvSender, OuterReceiver>
     {
-      return exit_opstate_t<Sender, OuterReceiver>(
-        static_cast<Sender&&>(sndr),
+      return exit_opstate_t<CvSender, OuterReceiver>(
+        static_cast<CvSender&&>(sndr),
         static_cast<OuterReceiver&&>(rcvr),
-        [](opstate_base<OuterReceiver>& op) -> propagate_receiver<OuterReceiver>
-        { return propagate_receiver<OuterReceiver>{{}, op}; },
+        [](opstate_base<OuterReceiver>& op) noexcept
+        { return propagate_receiver<opstate_base<OuterReceiver>>(op); },
         ctx);
     }
 
-    template <class Sender, class E>
+    template <class Sender, class Env>
     concept stream_completing_sender =
       sender<Sender>
       && gpu_stream_scheduler<
-        __result_of<get_completion_scheduler<set_value_t>, env_of_t<Sender>, E>,
-        E>;
+        __result_of<get_completion_scheduler<set_value_t>, env_of_t<Sender>, Env>,
+        Env>;
 
     template <class InnerReceiverProvider, class OuterReceiver>
     using inner_receiver_t = __call_result_t<InnerReceiverProvider, opstate_base<OuterReceiver>&>;
@@ -907,33 +920,34 @@ namespace nv::execution
     template <class CvSender, class InnerReceiver, class OuterReceiver>
     using stream_opstate_t = _strm::opstate<CvSender, InnerReceiver, OuterReceiver>;
 
-    template <class Sender, class OuterReceiver, class ReceiverProvider>
-      requires stream_completing_sender<Sender, env_of_t<OuterReceiver>>
-    auto
-    stream_opstate(Sender&& sndr, OuterReceiver&& out_receiver, ReceiverProvider receiver_provider)
-      -> stream_opstate_t<Sender, inner_receiver_t<ReceiverProvider, OuterReceiver>, OuterReceiver>
+    template <class CvSender, class OuterReceiver, class ReceiverProvider>
+      requires stream_completing_sender<CvSender, env_of_t<OuterReceiver>>
+    auto stream_opstate(CvSender&&       sndr,
+                        OuterReceiver&&  out_receiver,
+                        ReceiverProvider receiver_provider)
+      -> stream_opstate_t<CvSender, inner_receiver_t<ReceiverProvider, OuterReceiver>, OuterReceiver>
     {
       auto    sch = get_completion_scheduler<set_value_t>(get_env(sndr), get_env(out_receiver));
       context ctx = sch.ctx_;
 
-      return stream_opstate_t<Sender,
+      return stream_opstate_t<CvSender,
                               inner_receiver_t<ReceiverProvider, OuterReceiver>,
-                              OuterReceiver>(static_cast<Sender&&>(sndr),
+                              OuterReceiver>(static_cast<CvSender&&>(sndr),
                                              static_cast<OuterReceiver&&>(out_receiver),
                                              receiver_provider,
                                              ctx);
     }
 
-    template <class Sender, class OuterReceiver, class ReceiverProvider>
-    auto stream_opstate(Sender&&         sndr,
+    template <class CvSender, class OuterReceiver, class ReceiverProvider>
+    auto stream_opstate(CvSender&&       sndr,
                         OuterReceiver&&  out_receiver,
                         ReceiverProvider receiver_provider,
                         context          ctx)
-      -> stream_opstate_t<Sender, inner_receiver_t<ReceiverProvider, OuterReceiver>, OuterReceiver>
+      -> stream_opstate_t<CvSender, inner_receiver_t<ReceiverProvider, OuterReceiver>, OuterReceiver>
     {
-      return stream_opstate_t<Sender,
+      return stream_opstate_t<CvSender,
                               inner_receiver_t<ReceiverProvider, OuterReceiver>,
-                              OuterReceiver>(static_cast<Sender&&>(sndr),
+                              OuterReceiver>(static_cast<CvSender&&>(sndr),
                                              static_cast<OuterReceiver&&>(out_receiver),
                                              receiver_provider,
                                              ctx);
diff --git a/include/nvexec/stream/let_xxx.cuh b/include/nvexec/stream/let_xxx.cuh
index 37b8d62b8..5f567ec5e 100644
--- a/include/nvexec/stream/let_xxx.cuh
+++ b/include/nvexec/stream/let_xxx.cuh
@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include "../../stdexec/__detail/__variant.hpp"
 #include "../../stdexec/execution.hpp"
 
 #include "common.cuh"
@@ -31,117 +32,135 @@ STDEXEC_PRAGMA_IGNORE_EDG(cuda_compile)
 
 namespace nv::execution::_strm
 {
-  namespace let_xxx
+  namespace _let
   {
     using namespace STDEXEC;
 
     template <class... Args, class Fun, class ResultSenderT>
     STDEXEC_ATTRIBUTE(launch_bounds(1))
-    __global__ void _let_xxx_kernel(Fun fn, ResultSenderT* result, Args... args)
+    __global__ void _let_kernel(Fun fn, ResultSenderT* result, Args... args)
     {
       static_assert(trivially_copyable<Fun, Args...>);
       new (result) ResultSenderT(::cuda::std::move(fn)(static_cast<Args&&>(args)...));
     }
 
-    template <class Tp>
-    using __decay_ref_t = STDEXEC::__decay_t<Tp>&;
+    template <class Sender, class Receiver, class Fun, class SetTag>
+    struct _opstate;
 
     template <class Fun>
-    struct _mk_result_sender
+    struct _mk_result_sender_fn
     {
       template <class... Args>
-      using __f = __remove_rvalue_reference_t<__call_result_t<Fun, __decay_ref_t<Args>...>>;
+      using __f = __remove_rvalue_reference_t<__call_result_t<Fun, __decay_t<Args>&...>>;
     };
 
     template <class Sender, class PropagateReceiver, class Fun, class SetTag>
       requires sender_in<Sender, env_of_t<PropagateReceiver>>
-    struct __max_sender_size
+    struct _max_sender_size
     {
-      struct _result_sender_size
-      {
-        template <class... Args>
-        using __f = __msize_t<sizeof(__minvoke<_mk_result_sender<Fun>, Args...>)>;
-      };
-
-      static constexpr std::size_t value = __gather_completions_of_t<SetTag,
-                                                                     Sender,
-                                                                     env_of_t<PropagateReceiver>,
-                                                                     _result_sender_size,
-                                                                     maxsize>::value;
+      using _env_t              = env_of_t<PropagateReceiver>;
+      using _result_sender_size = __mcompose<__msizeof, _mk_result_sender_fn<Fun>>;
+      using _value_t =
+        __gather_completions_of_t<SetTag, Sender, _env_t, _result_sender_size, maxsize>;
+
+      static constexpr std::size_t value = _value_t::value;
+    };
+
+    // The environment of the receiver used to connect the secondary (result) sender must
+    // correctly report the scheduler and domain on which the sender's operation will be
+    // started.
+    inline constexpr auto _mk_sch_env =
+      []<class CvSender, class Receiver, class SetTag>(CvSender&& sndr, Receiver&& rcvr, SetTag)
+    {
+      using cv_fn = __copy_cvref_fn<CvSender>;
+      return __mk_secondary_env_t<SetTag>()(cv_fn{}, sndr, STDEXEC::get_env(rcvr));
+    };
+
+    template <class CvSender, class Receiver, class SetTag>
+    using _sch_env_t = __result_of<_mk_sch_env, CvSender, Receiver, SetTag>;
+
+    inline constexpr auto _mk_env2 =
+      []<class SchEnv, class Receiver>([[maybe_unused]]
+                                       SchEnv const &                        sch_env,
+                                       _strm::opstate_base<Receiver> const & opstate)
+    {
+      //return opstate.make_env();
+      return __env::__join(sch_env, opstate.make_env());
     };
 
-    template <class Receiver, class Fun>
-    using opstate_for_t = __mcompose<__mbind_back_q<connect_result_t, propagate_receiver<Receiver>>,
-                                     _mk_result_sender<Fun>>;
+    template <class CvSender, class Receiver, class SetTag>
+    using _env2_t = __result_of<_mk_env2,
+                                _sch_env_t<CvSender, Receiver, SetTag> const &,
+                                _strm::opstate_base<Receiver> const &>;
+
+    template <class CvSender, class Receiver, class Fun, class SetTag>
+    using _propagate_receiver_t = propagate_receiver<_opstate<CvSender, Receiver, Fun, SetTag>,
+                                                     _env2_t<CvSender, Receiver, SetTag>>;
 
-    template <class Set, class Sig>
-    struct __tfx_signal_fn
+    template <class Sender, class Receiver, class Fun, class SetTag>
+    using _mk_opstate_fn = __mcompose<
+      __mbind_back_q<connect_result_t, _propagate_receiver_t<Sender, Receiver, Fun, SetTag>>,
+      _mk_result_sender_fn<Fun>>;
+
+    template <class SetTag, class Sig>
+    struct _tfx_signal_fn
     {
       template <class, class...>
       using __f = completion_signatures<Sig>;
     };
 
-    template <class Set, class... Args>
-    struct __tfx_signal_fn<Set, Set(Args...)>
+    template <class SetTag, class... Args>
+    struct _tfx_signal_fn<SetTag, SetTag(Args...)>
     {
       template <class Fun, class... StreamEnv>
       using __f = __transform_completion_signatures_t<
-        __completion_signatures_of_t<__minvoke<_mk_result_sender<Fun>, Args...>, StreamEnv...>,
+        __completion_signatures_of_t<__minvoke<_mk_result_sender_fn<Fun>, Args...>, StreamEnv...>,
         completion_signatures<set_error_t(cudaError_t)>>;
     };
 
-    template <class Sig, class Fun, class Set, class... StreamEnv>
-    using __tfx_signal_t = __minvoke<__tfx_signal_fn<Set, Sig>, Fun, StreamEnv...>;
-
-    template <class Sender, class Receiver, class Fun, class Let>
-    struct opstate;
+    template <class Sig, class Fun, class SetTag, class... StreamEnv>
+    using _tfx_signal_t = __minvoke<_tfx_signal_fn<SetTag, Sig>, Fun, StreamEnv...>;
 
-    template <class Sender, class Receiver, class Fun, class Set, class... Tuples>
-    struct receiver : public stream_receiver_base
+    template <class Sender, class Receiver, class Fun, class SetTag, class... Tuples>
+    struct _receiver : public stream_receiver_base
     {
-      using env_t = _strm::opstate_base<Receiver>::env_t;
+      using _env_t           = _strm::opstate_base<Receiver>::env_t;
+      using _result_tuples_t = __mlist<Tuples...>;
+
       static constexpr std::size_t memory_allocation_size() noexcept
       {
-        return __max_sender_size<Sender, propagate_receiver<Receiver>, Fun, Set>::value;
+        using _propagate_receiver_t = _let::_propagate_receiver_t<Sender, Receiver, Fun, SetTag>;
+        return _max_sender_size<Sender, _propagate_receiver_t, Fun, SetTag>::value;
       }
 
-      template <__same_as<Set> Tag, class... Args>
+      template <__same_as<SetTag> Tag, class... Args>
       void _complete(Tag, Args&&... args) noexcept
       {
-        using result_sender_t = __minvoke<_mk_result_sender<Fun>, Args...>;
-        using opstate_t       = __minvoke<opstate_for_t<Receiver, Fun>, Args...>;
+        using result_sender_t = __minvoke<_mk_result_sender_fn<Fun>, Args...>;
 
         cudaStream_t stream        = opstate_->get_stream();
         auto*        result_sender = static_cast<result_sender_t*>(opstate_->temp_storage_);
-        _let_xxx_kernel<Args&&...><<<1, 1, 0, stream>>>(std::move(opstate_->fun_),
-                                                        result_sender,
-                                                        static_cast<Args&&>(args)...);
+        _let_kernel<Args&&...><<<1, 1, 0, stream>>>(std::move(opstate_->fun_),
+                                                    result_sender,
+                                                    static_cast<Args&&>(args)...);
 
-        if (cudaError_t status = STDEXEC_LOG_CUDA_API(cudaStreamSynchronize(stream));
-            status == cudaSuccess)
+        cudaError_t status = STDEXEC_LOG_CUDA_API(cudaStreamSynchronize(stream));
+        if (status == cudaSuccess)
         {
           opstate_->defer_temp_storage_destruction(result_sender);
-          auto& op = opstate_->opstate3_.template emplace<opstate_t>(__emplace_from{
-            [&]
-            {
-              return connect(std::move(*result_sender),
-                             propagate_receiver<Receiver>{
-                               {},
-                               static_cast<_strm::opstate_base<Receiver>&>(*opstate_)});
-            }});
+          auto& op = opstate_->_connect_result_sender(std::move(*result_sender));
           STDEXEC::start(op);
         }
         else
         {
-          opstate_->propagate_completion_signal(STDEXEC::set_error, std::move(status));
+          opstate_->propagate_completion_signal(STDEXEC::set_error, cudaError_t(status));
         }
       }
 
-      template <class Tag, class... Args>
-      void _complete(Tag, Args&&... args) noexcept
+      template <class SetTag2, class... Args>
+      void _complete(SetTag2, Args&&... args) noexcept
       {
-        static_assert(__nothrow_callable<Tag, Receiver, Args...>);
-        opstate_->propagate_completion_signal(Tag(), static_cast<Args&&>(args)...);
+        opstate_->propagate_completion_signal(SetTag2(), static_cast<Args&&>(args)...);
       }
 
       template <class... Args>
@@ -161,116 +180,151 @@ namespace nv::execution::_strm
         _complete(set_stopped_t());
       }
 
-      auto get_env() const noexcept -> env_t
+      auto get_env() const noexcept -> _env_t
       {
-        return opstate_->make_env();
+        return static_cast<_strm::opstate_base<Receiver>&>(*opstate_).make_env();
       }
 
-      using opstate_variant_t = __minvoke<
-        __mtransform<__muncurry<opstate_for_t<Receiver, Fun>>, __qq<__nullable_std_variant>>,
-        Tuples...>;
-
-      opstate<Sender, Receiver, Fun, Set>* opstate_;
+      _opstate<Sender, Receiver, Fun, SetTag>* opstate_;
     };
 
-    template <class Sender, class Receiver, class Fun, class Set>
-    using receiver_t =
-      __gather_completions_of_t<Set,
-                                Sender,
-                                stream_env_t<env_of_t<Receiver>>,
-                                __q<__decayed_std_tuple>,
-                                __munique<__mbind_front_q<receiver, Sender, Receiver, Fun, Set>>>;
+    template <class Sender, class Receiver, class Fun, class SetTag>
+    using _receiver_t = __gather_completions_of_t<
+      SetTag,
+      Sender,
+      stream_env_t<env_of_t<Receiver>>,
+      __q<__decayed_std_tuple>,
+      __munique<__mbind_front_q<_receiver, Sender, Receiver, Fun, SetTag>>>;
 
-    template <class Sender, class Receiver, class Fun, class Set>
-    using opstate_base_t = _strm::opstate<Sender, receiver_t<Sender, Receiver, Fun, Set>, Receiver>;
+    template <class Sender, class Receiver, class Fun, class SetTag>
+    using _opstate_base_t =
+      _strm::opstate<Sender, _receiver_t<Sender, Receiver, Fun, SetTag>, Receiver>;
 
-    template <class Sender, class Receiver, class Fun, class Set>
-    struct opstate : opstate_base_t<Sender, Receiver, Fun, Set>
+    template <class CvSender, class Receiver, class Fun, class SetTag>
+    struct _opstate : _opstate_base_t<CvSender, Receiver, Fun, SetTag>
     {
-      using receiver_t        = receiver_t<Sender, Receiver, Fun, Set>;
-      using opstate_variant_t = receiver_t::opstate_variant_t;
+      using _env2_t                = _sch_env_t<CvSender, Receiver, SetTag>;
+      using _receiver_t            = _let::_receiver_t<CvSender, Receiver, Fun, SetTag>;
+      using _result_tuples_t       = _receiver_t::_result_tuples_t;
+      using _mk_opstate_fn_t       = _mk_opstate_fn<CvSender, Receiver, Fun, SetTag>;
+      using _mk_opstate_variant_fn = __mtransform<__muncurry<_mk_opstate_fn_t>, __qq<__variant>>;
+      using _opstate_variant_t     = __mapply<_mk_opstate_variant_fn, _result_tuples_t>;
+      using _propagate_receiver_t  = _let::_propagate_receiver_t<CvSender, Receiver, Fun, SetTag>;
+
+      explicit _opstate(CvSender&& sndr, Receiver rcvr, Fun fun)
+        : _opstate(static_cast<CvSender&&>(sndr),
+                   static_cast<Receiver&&>(rcvr),
+                   static_cast<Fun&&>(fun),
+                   _mk_sch_env(sndr, rcvr, SetTag{}))
+      {}
 
-      opstate(Sender&& sndr, Receiver rcvr, Fun fun)
-        : opstate_base_t<Sender, Receiver, Fun, Set>(
-            static_cast<Sender&&>(sndr),
+      explicit _opstate(CvSender&& sndr, Receiver&& rcvr, Fun fun, _env2_t env2)
+        : _opstate_base_t<CvSender, Receiver, Fun, SetTag>(
+            static_cast<CvSender&&>(sndr),
             static_cast<Receiver&&>(rcvr),
-            [this](_strm::opstate_base<Receiver>&) -> receiver_t { return receiver_t{{}, this}; },
+            [this](__ignore) noexcept { return _receiver_t{{}, this}; },
             get_completion_scheduler<set_value_t>(get_env(sndr), get_env(rcvr)).ctx_)
         , fun_(static_cast<Fun&&>(fun))
+        , env2_(env2)
       {}
 
-      STDEXEC_IMMOVABLE(opstate);
+      STDEXEC_IMMOVABLE(_opstate);
 
-      Fun               fun_;
-      opstate_variant_t opstate3_;
+      [[nodiscard]]
+      auto make_env() const noexcept -> _let::_env2_t<CvSender, Receiver, SetTag>
+      {
+        return _let::_mk_env2(env2_, *this);
+      }
+
+      template <class ResultSender>
+      auto _connect_result_sender(ResultSender&& sndr)
+        -> connect_result_t<ResultSender, _propagate_receiver_t>&
+      {
+        return opstate3_.__emplace_from(STDEXEC::connect,
+                                        static_cast<ResultSender&&>(sndr),
+                                        _propagate_receiver_t(*this));
+      }
+
+      Fun                fun_;
+      _env2_t            env2_;
+      _opstate_variant_t opstate3_{__no_init};
     };
-  }  // namespace let_xxx
+  }  // namespace _let
 
-  template <class Sender, class Fun, class Set>
+  template <class Sender, class Fun, class SetTag>
   struct let_sender : public stream_sender_base
   {
+   private:
     template <class Self, class Receiver>
-    using opstate_t = let_xxx::opstate<__copy_cvref_t<Self, Sender>, Receiver, Fun, Set>;
+    using _opstate_t = _let::_opstate<__copy_cvref_t<Self, Sender>, Receiver, Fun, SetTag>;
 
     template <class Self, class Receiver>
-    using _receiver_t = let_xxx::receiver_t<__copy_cvref_t<Self, Sender>, Receiver, Fun, Set>;
+    using _receiver_t = _let::_receiver_t<__copy_cvref_t<Self, Sender>, Receiver, Fun, SetTag>;
 
     template <class CvSender, class... StreamEnv>
     using _completions_t =
-      __mapply<__mtransform<__mbind_back_q<let_xxx::__tfx_signal_t, Fun, Set, StreamEnv...>,
+      __mapply<__mtransform<__mbind_back_q<_let::_tfx_signal_t, Fun, SetTag, StreamEnv...>,
                             __mtry_q<__concat_completion_signatures_t>>,
                __completion_signatures_of_t<CvSender, StreamEnv...>>;
 
-    template <__decays_to<let_sender> Self, receiver Receiver>
-      requires receiver_of<
-        Receiver,
-        _completions_t<__copy_cvref_t<Self, Sender>, stream_env_t<env_of_t<Receiver>>>>
-    STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr)
-      -> opstate_t<Self, Receiver>
-    {
-      return opstate_t<Self, Receiver>{static_cast<Self&&>(self).sndr_,
-                                       static_cast<Receiver&&>(rcvr),
-                                       static_cast<Self&&>(self).fun_};
-    }
-    STDEXEC_EXPLICIT_THIS_END(connect)
+   public:
+    explicit let_sender(Sender sndr, Fun fun, SetTag)
+      noexcept(__nothrow_move_constructible<Sender, Fun>)
+      : sndr_(static_cast<Sender&&>(sndr))
+      , fun_(static_cast<Fun&&>(fun))
+    {}
 
+    [[nodiscard]]
     auto get_env() const noexcept -> stream_sender_attrs<Sender>
     {
       return {&sndr_};
     }
 
-    template <__decays_to<let_sender> Self, class... Env>
+    template <class Self, class... Env>
     static consteval auto get_completion_signatures()
-      -> _completions_t<__copy_cvref_t<Self, Sender>, stream_env_t<Env>...>
     {
-      return {};
+      return _completions_t<__copy_cvref_t<Self, Sender>, stream_env_t<Env>...>{};
+    }
+
+    template <class Self, receiver Receiver>
+    STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr)
+      -> _opstate_t<Self, Receiver>
+    {
+      return _opstate_t<Self, Receiver>{static_cast<Self&&>(self).sndr_,
+                                        static_cast<Receiver&&>(rcvr),
+                                        static_cast<Self&&>(self).fun_};
     }
+    STDEXEC_EXPLICIT_THIS_END(connect)
 
+   private:
     Sender sndr_;
     Fun    fun_;
   };
 
-  template <class Set>
-  struct _transform_let_xxx_sender
+  template <class Sender, class Fun, class SetTag>
+  STDEXEC_HOST_DEVICE_DEDUCTION_GUIDE
+    let_sender(Sender, Fun, SetTag) -> let_sender<Sender, Fun, SetTag>;
+
+  template <class SetTag>
+  struct _transform_let_sender
   {
     template <class Env, class Fun, stream_completing_sender<Env> Sender>
     auto operator()(Env const &, __ignore, Fun fn, Sender&& sndr) const
     {
-      using __sender_t = let_sender<__decay_t<Sender>, Fun, Set>;
-      return __sender_t{{}, static_cast<Sender&&>(sndr), static_cast<Fun&&>(fn)};
+      return let_sender{static_cast<Sender&&>(sndr), static_cast<Fun&&>(fn), SetTag{}};
     }
   };
 
   template <>
-  struct transform_sender_for<STDEXEC::let_value_t> : _transform_let_xxx_sender<set_value_t>
+  struct transform_sender_for<STDEXEC::let_value_t> : _transform_let_sender<set_value_t>
   {};
 
   template <>
-  struct transform_sender_for<STDEXEC::let_error_t> : _transform_let_xxx_sender<set_error_t>
+  struct transform_sender_for<STDEXEC::let_error_t> : _transform_let_sender<set_error_t>
   {};
 
   template <>
-  struct transform_sender_for<STDEXEC::let_stopped_t> : _transform_let_xxx_sender<set_stopped_t>
+  struct transform_sender_for<STDEXEC::let_stopped_t> : _transform_let_sender<set_stopped_t>
   {};
 }  // namespace nv::execution::_strm
 
@@ -278,9 +332,9 @@ namespace nvexec = nv::execution;
 
 namespace STDEXEC::__detail
 {
-  template <class Sender, class Fun, class Set>
-  extern __declfn_t<nvexec::_strm::let_sender<__demangle_t<Sender>, Fun, Set>>
-    __demangle_v<nvexec::_strm::let_sender<Sender, Fun, Set>>;
+  template <class Sender, class Fun, class SetTag>
+  extern __declfn_t<nvexec::_strm::let_sender<__demangle_t<Sender>, Fun, SetTag>>
+    __demangle_v<nvexec::_strm::let_sender<Sender, Fun, SetTag>>;
 }  // namespace STDEXEC::__detail
 
 STDEXEC_PRAGMA_POP()
diff --git a/include/nvexec/stream/repeat_n.cuh b/include/nvexec/stream/repeat_n.cuh
index e75f4894f..3ecbc01e3 100644
--- a/include/nvexec/stream/repeat_n.cuh
+++ b/include/nvexec/stream/repeat_n.cuh
@@ -34,7 +34,7 @@ namespace nv::execution::_strm
     template <class OpState>
     struct receiver : stream_receiver_base
     {
-      receiver(OpState& opstate) noexcept
+      explicit receiver(OpState& opstate) noexcept
         : opstate_(opstate)
       {}
 
@@ -54,6 +54,7 @@ namespace nv::execution::_strm
         opstate_._complete(STDEXEC::set_stopped);
       }
 
+      [[nodiscard]]
       auto get_env() const noexcept -> OpState::env_t
       {
         return opstate_.make_env();
@@ -62,24 +63,23 @@ namespace nv::execution::_strm
       OpState& opstate_;
     };
 
-    template <class CvSender, class Receiver>
+    template <class Sender, class Receiver>
     struct opstate : _strm::opstate_base<Receiver>
     {
       using operation_state_concept = STDEXEC::operation_state_t;
 
-      using sender_t = STDEXEC::__decay_t<CvSender>;
       using scheduler_t =
         STDEXEC::__result_of<STDEXEC::get_completion_scheduler<STDEXEC::set_value_t>,
-                             STDEXEC::env_of_t<sender_t>,
+                             STDEXEC::env_of_t<Sender>,
                              STDEXEC::env_of_t<Receiver>>;
 
       using inner_sender_t =
-        STDEXEC::__result_of<exec::sequence, STDEXEC::schedule_result_t<scheduler_t&>, sender_t&>;
+        STDEXEC::__result_of<exec::sequence, STDEXEC::schedule_result_t<scheduler_t&>, Sender&>;
       using inner_opstate_t = STDEXEC::connect_result_t<inner_sender_t, receiver<opstate>>;
 
-      explicit opstate(CvSender&& sndr, Receiver rcvr, std::size_t count, scheduler_t sched)
+      explicit opstate(Sender&& sndr, Receiver rcvr, std::size_t count, scheduler_t sched)
         : _strm::opstate_base<Receiver>(static_cast<Receiver&&>(rcvr), sched.ctx_)
-        , sndr_(static_cast<CvSender&&>(sndr))
+        , sndr_(static_cast<Sender&&>(sndr))
         , sched_(std::move(sched))
         , count_(count)
       {
@@ -90,6 +90,7 @@ namespace nv::execution::_strm
       {
         inner_opstate_.__emplace_from(STDEXEC::connect,
                                       exec::sequence(STDEXEC::schedule(sched_), sndr_),
+                                      //STDEXEC::on(sched_, sndr_),
                                       receiver{*this});
       }
 
@@ -146,7 +147,7 @@ namespace nv::execution::_strm
         }
       }
 
-      sender_t                             sndr_;
+      Sender                               sndr_;
       scheduler_t                          sched_;
       STDEXEC::__optional<inner_opstate_t> inner_opstate_;
       std::size_t                          count_{};
@@ -157,19 +158,22 @@ namespace nv::execution::_strm
     {
       using sender_concept = STDEXEC::sender_t;
 
-      using completion_signatures =
-        STDEXEC::completion_signatures<STDEXEC::set_value_t(),
-                                       STDEXEC::set_stopped_t(),
-                                       STDEXEC::set_error_t(std::exception_ptr),
-                                       STDEXEC::set_error_t(cudaError_t)>;
+      template <class>
+      static consteval auto get_completion_signatures()
+      {
+        return STDEXEC::completion_signatures<STDEXEC::set_value_t(),
+                                              STDEXEC::set_stopped_t(),
+                                              STDEXEC::set_error_t(std::exception_ptr),
+                                              STDEXEC::set_error_t(cudaError_t)>();
+      }
 
       template <STDEXEC::receiver Receiver>
-      auto connect(Receiver rcvr) &&  //
-        -> repeat_n::opstate<CvSender, Receiver>
+      auto connect(Receiver rcvr) && -> repeat_n::opstate<CvSender, Receiver>
       {
         auto sched =
           STDEXEC::get_completion_scheduler<STDEXEC::set_value_t>(STDEXEC::get_env(sndr_),
                                                                   STDEXEC::get_env(rcvr));
+
         return repeat_n::opstate<CvSender, Receiver>(static_cast<CvSender&&>(sndr_),
                                                      static_cast<Receiver&&>(rcvr),
                                                      count_,
@@ -190,11 +194,10 @@ namespace nv::execution::_strm
   template <>
   struct transform_sender_for<exec::repeat_n_t>
   {
-    template <class Env, class CvSender>
-    auto operator()(Env const &, STDEXEC::__ignore, size_t count, CvSender&& sndr) const
+    template <class Env, class Sender>
+    auto operator()(Env const &, STDEXEC::__ignore, size_t count, Sender sndr) const
     {
-      using sender_t = repeat_n::sender<CvSender>;
-      return sender_t{static_cast<CvSender&&>(sndr), count};
+      return repeat_n::sender<Sender>{static_cast<Sender&&>(sndr), count};
     }
   };
 }  // namespace nv::execution::_strm
diff --git a/include/nvexec/stream/schedule_from.cuh b/include/nvexec/stream/schedule_from.cuh
index 10e23a29e..b4792247c 100644
--- a/include/nvexec/stream/schedule_from.cuh
+++ b/include/nvexec/stream/schedule_from.cuh
@@ -93,7 +93,7 @@ namespace nv::execution
         {
           if (this->status_ == cudaSuccess)
           {
-            this->status_ = task_->status_;
+            this->status_ = task_->status();
           }
         }
 
diff --git a/include/stdexec/__detail/__meta.hpp b/include/stdexec/__detail/__meta.hpp
index 8f2424483..81748771d 100644
--- a/include/stdexec/__detail/__meta.hpp
+++ b/include/stdexec/__detail/__meta.hpp
@@ -681,6 +681,12 @@ namespace STDEXEC
     using __f = __msize_t<sizeof...(_Ts)>;
   };
 
+  struct __msizeof
+  {
+    template <class _Ty>
+    using __f = __msize_t<sizeof(_Ty)>;
+  };
+
   template <class _Ty>
   struct __mcount
   {