Do not dispatch thread-pool bulk for negative shapes

include/exec/static_thread_pool.hpp

@@ -1378,8 +1378,10 @@ namespace experimental::execution
       {
         if constexpr (Parallelize)
         {
-          return static_cast<std::uint32_t>(
-            __umin({std::size_t(shape_), std::size_t(pool_.available_parallelism())}));
+          return Shape{} < shape_
+               ? static_cast<std::uint32_t>(
+                   __umin({std::size_t(shape_), std::size_t(pool_.available_parallelism())}))
+               : 0;
         }
         else
         {
@@ -1491,7 +1493,7 @@ namespace experimental::execution
           }
         }
 
-        if (shared_state_.shape_)
+        if (Shape{} < shared_state_.shape_)
         {
           enqueue();
         }

include/exec/thread_pool_base.hpp

@@ -207,8 +207,10 @@ namespace experimental::execution
         {
           // With work stealing, is std::min necessary, or can we feel free to ask for more agents (tasks)
           // than we can actually deal with at one time?
-          return static_cast<std::uint32_t>(
-            (std::min) (shape_, static_cast<Shape>(pool_.available_parallelism())));
+          return Shape{} < shape_
+               ? static_cast<std::uint32_t>(
+                   (std::min) (shape_, static_cast<Shape>(pool_.available_parallelism())))
+               : 0;
         }
 
         template <class F>
@@ -329,7 +331,7 @@ namespace experimental::execution
             state.data_.template emplace<tuple_t>(static_cast<As&&>(as)...);
           }
 
-          if (state.shape_)
+          if (Shape{} < state.shape_)
           {
             enqueue();
           }

test/exec/test_thread_pool_base.cpp

@@ -125,5 +125,24 @@ namespace
     CHECK(bulk_calls == 0);
     CHECK(pool.enqueued_ == 1);
   }
+
+  TEST_CASE("thread_pool_base bulk does not invoke the function with a negative shape",
+            "[thread_pool_base][bulk]")
+  {
+    inline_test_thread_pool pool;
+    completion_state        state;
+    int                     bulk_calls = 0;
+
+    auto sndr = ex::schedule(pool.get_scheduler())
+              | ex::bulk_chunked(ex::par, -1, [&](int, int) noexcept { ++bulk_calls; });
+    auto op = ex::connect(std::move(sndr), counting_receiver{&state});
+
+    ex::start(op);
+
+    CHECK(state.completions_ == 1);
+    CHECK_FALSE(state.error_);
+    CHECK(bulk_calls == 0);
+    CHECK(pool.enqueued_ == 1);
+  }
 }  // namespace
 #endif  // !STDEXEC_NO_STDCPP_EXCEPTIONS()

test/stdexec/algos/adaptors/test_bulk.cpp

@@ -27,6 +27,7 @@
 #if STDEXEC_USE_MODULES()
 import std;
 #else
+#  include <atomic>
 #  include <exception>
 #  include <numeric>
 #  include <vector>
@@ -721,6 +722,21 @@ namespace
     CHECK(called == 0);
   }
 
+  TEST_CASE("bulk_chunked function is not called with a negative shape on a static thread pool",
+            "[adaptors][bulk]")
+  {
+    exec::static_thread_pool pool{4};
+    std::atomic<int>         called{};
+
+    auto snd = ex::just() | ex::continues_on(pool.get_scheduler())
+             | ex::bulk_chunked(ex::par,
+                                -1,
+                                [&called](int, int) { called.fetch_add(1); });
+    ex::sync_wait(std::move(snd));
+
+    CHECK(called == 0);
+  }
+
   TEST_CASE("bulk_unchunked function in not called on stop", "[adaptors][bulk]")
   {
     constexpr int n = 2;