From abc6299849f7f6075a36f6b644d8f13736b37c4b Mon Sep 17 00:00:00 2001
From: Minh Vu <vuhoangminh97@gmail.com>
Date: Wed, 26 Aug 2026 11:52:03 +0200
Subject: [PATCH] Fix thread pool bulk handling of negative shapes

---
 include/exec/static_thread_pool.hpp       |  8 +++++---
 include/exec/thread_pool_base.hpp         |  8 +++++---
 test/exec/test_thread_pool_base.cpp       | 19 +++++++++++++++++++
 test/stdexec/algos/adaptors/test_bulk.cpp | 16 ++++++++++++++++
 4 files changed, 45 insertions(+), 6 deletions(-)

diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp
index f84d03851..d2c43e122 100644
--- a/include/exec/static_thread_pool.hpp
+++ b/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();
         }
diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp
index ed4f44c6f..058c1fb0c 100644
--- a/include/exec/thread_pool_base.hpp
+++ b/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();
           }
diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp
index b36097706..68fe40a08 100644
--- a/test/exec/test_thread_pool_base.cpp
+++ b/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()
diff --git a/test/stdexec/algos/adaptors/test_bulk.cpp b/test/stdexec/algos/adaptors/test_bulk.cpp
index c365e31ad..4b1e687df 100644
--- a/test/stdexec/algos/adaptors/test_bulk.cpp
+++ b/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;