From be5eceb4a727adf760ddc3a2b8e52aab022fef24 Mon Sep 17 00:00:00 2001
From: gabewillen <gabewillen@gmail.com>
Date: Thu, 2 Jul 2026 15:23:52 +0000
Subject: [PATCH 1/4] Add external-completion scheduler policy for co_sm

Adds utility/external_completion.hpp: completion_source (an atomic flag a
worker thread fires via a trampoline compatible with
thread_pool_scheduler::try_submit_with_completion) and
external_completion_scheduler<SourceCount>, a single-consumer scheduler
policy that lets a co_sm dispatch suspend awaiting completions fired by
other threads while still returning only when the whole dispatch is done.

co_sm gains a structural external_completion_scheduler_contract concept
(duck-typed so co_sm.hpp stays independent of the opt-in module) and an
external dispatch path: (1) commit sweep - completions that fired between
dispatches are delivered as explicit utility::completion events in
ascending source order before the trigger event runs; (2) the trigger
event dispatches, its actions may arm sources, hand them to workers, and
mark them required; (3) the dispatch coroutine awaits each required
source (lowest fired index first) and re-enters it as a completion
event; the top-level process_event parks on the scheduler's semaphore
and resumes the coroutine on the dispatching thread until the root task
is ready.

Producers interact only through completion_source::fire (atomic store +
semaphore release, never a coroutine resumption), so single-consumer
dispatch holds structurally. Delivery order is deterministic regardless
of physical fire order. Run-to-completion holds: required completions
drain before dispatch returns; unrequired background fires persist only
as passive flags swept by the next dispatch.

Module is opt-in like the thread pool scheduler: requires C++20
coroutines plus <semaphore>, compiles to nothing otherwise.
---
 include/boost/sml/utility/co_sm.hpp           |  85 ++++++-
 .../boost/sml/utility/external_completion.hpp | 224 +++++++++++++++++
 .../sml/utility/external_completion.hpp       |   7 +
 test/ft/CMakeLists.txt                        |   4 +
 test/ft/external_completion.cpp               | 232 ++++++++++++++++++
 5 files changed, 541 insertions(+), 11 deletions(-)
 create mode 100644 include/boost/sml/utility/external_completion.hpp
 create mode 100644 include/stateforward/sml/utility/external_completion.hpp
 create mode 100644 test/ft/external_completion.cpp

diff --git a/include/boost/sml/utility/co_sm.hpp b/include/boost/sml/utility/co_sm.hpp
index 65ddc486..4f645402 100644
--- a/include/boost/sml/utility/co_sm.hpp
+++ b/include/boost/sml/utility/co_sm.hpp
@@ -303,6 +303,26 @@ concept has_try_run_immediate = requires(TScheduler scheduler) {
   } -> std::same_as<bool>;
 };
 
+// Structural contract for schedulers that deliver externally fired completions
+// (see utility/external_completion.hpp). Detected duck-typed so this header
+// stays independent of the opt-in module. Such schedulers still honor the
+// strict ordering contract: required completions are drained before the
+// top-level dispatch returns, and delivery order is deterministic.
+template <class TScheduler>
+concept external_completion_scheduler_contract =
+    requires {
+      { TScheduler::external_completion } -> std::convertible_to<bool>;
+      { TScheduler::npos } -> std::convertible_to<std::size_t>;
+      typename TScheduler::completion_event;
+    } && static_cast<bool>(TScheduler::external_completion) &&
+    requires(TScheduler scheduler) {
+      { scheduler.has_required() } -> std::convertible_to<bool>;
+      { scheduler.sweep_next_fired() } -> std::same_as<std::size_t>;
+      scheduler.next_fired_required();
+      scheduler.wait_any();
+      { scheduler.try_resume_parked() } -> std::same_as<bool>;
+    };
+
 template <class TAllocatorPolicy>
 concept valid_coroutine_allocator_policy = requires { typename TAllocatorPolicy::allocator_type; };
 
@@ -578,7 +598,20 @@ class co_sm {
 
   template <class TEvent>
   bool process_event(const TEvent& event) {
-    return state_machine_.process_event(event);
+    if constexpr (policy::external_completion_scheduler_contract<scheduler_type>) {
+      // The dispatch coroutine may suspend awaiting externally fired
+      // completions; drive it to completion here so the run-to-completion
+      // contract holds: nothing observable escapes this call, and resumption
+      // happens only on this (the dispatching) thread.
+      bool_task task = process_event_external_impl(std::allocator_arg, allocator_, *this, event);
+      while (!task.await_ready()) {
+        scheduler_.wait_any();
+        (void)scheduler_.try_resume_parked();
+      }
+      return task.result();
+    } else {
+      return state_machine_.process_event(event);
+    }
   }
 
   template <class TEvent>
@@ -586,19 +619,21 @@ class co_sm {
     using event_t = typename std::decay<TEvent>::type;
     event_t event_copy(static_cast<TEvent&&>(event));
 
-    if constexpr (std::is_same_v<scheduler_type, policy::inline_scheduler>) {
+    if constexpr (policy::external_completion_scheduler_contract<scheduler_type>) {
+      return bool_task::from_value(process_event(event_copy));
+    } else if constexpr (std::is_same_v<scheduler_type, policy::inline_scheduler>) {
       return bool_task::from_value(state_machine_.process_event(event_copy));
-    }
-
-    if constexpr (policy::has_try_run_immediate<scheduler_type>) {
-      bool accepted = false;
-      if (scheduler_.try_run_immediate(
-              [this, &event_copy, &accepted]() { accepted = state_machine_.process_event(event_copy); })) {
-        return bool_task::from_value(accepted);
+    } else {
+      if constexpr (policy::has_try_run_immediate<scheduler_type>) {
+        bool accepted = false;
+        if (scheduler_.try_run_immediate(
+                [this, &event_copy, &accepted]() { accepted = state_machine_.process_event(event_copy); })) {
+          return bool_task::from_value(accepted);
+        }
       }
-    }
 
-    return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast<event_t&&>(event_copy));
+      return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast<event_t&&>(event_copy));
+    }
   }
 
   template <class TState>
@@ -627,6 +662,34 @@ class co_sm {
     co_return co_await process_event_awaitable<typename std::decay<TEvent>::type>{self, static_cast<TEvent&&>(event)};
   }  // GCOVR_EXCL_LINE
 
+  // External-completion dispatch: (1) commit sweep — completions fired between
+  // dispatches are delivered as explicit completion events in ascending source
+  // order before the trigger runs; (2) the trigger event dispatches; (3) any
+  // sources the machine marked required are awaited (suspending this coroutine)
+  // and re-entered as completion events until none remain. The caller's drive
+  // loop in process_event resumes the coroutine on the dispatching thread only.
+  template <class TEvent>
+  static bool_task process_event_external_impl(std::allocator_arg_t, allocator_type& allocator, co_sm& self, TEvent event)
+    requires policy::external_completion_scheduler_contract<scheduler_type>
+  {
+    (void)allocator;
+    using completion_event_t = typename scheduler_type::completion_event;
+
+    for (std::size_t index = self.scheduler_.sweep_next_fired(); index != scheduler_type::npos;
+         index = self.scheduler_.sweep_next_fired()) {
+      (void)self.state_machine_.process_event(completion_event_t{index});
+    }
+
+    bool accepted = self.state_machine_.process_event(event);
+
+    while (self.scheduler_.has_required()) {
+      const std::size_t index = co_await self.scheduler_.next_fired_required();
+      accepted = self.state_machine_.process_event(completion_event_t{index}) && accepted;
+    }
+
+    co_return accepted;
+  }  // GCOVR_EXCL_LINE
+
   template <class TEvent>
   struct process_event_awaitable {
     co_sm& self;
diff --git a/include/boost/sml/utility/external_completion.hpp b/include/boost/sml/utility/external_completion.hpp
new file mode 100644
index 00000000..eb59594f
--- /dev/null
+++ b/include/boost/sml/utility/external_completion.hpp
@@ -0,0 +1,224 @@
+//
+// Copyright (c) 2026 stateforward
+//
+// 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)
+//
+#ifndef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP
+#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP
+
+#include "boost/sml.hpp"
+
+#if defined(_MSVC_LANG)
+#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG _MSVC_LANG
+#else
+#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG __cplusplus
+#endif
+
+// Opt-in module: external completion delivery needs C++20 coroutines plus a
+// hosted std::counting_semaphore, so it compiles to nothing on freestanding or
+// pre-C++20 toolchains, keeping the header-only core dependency-free.
+#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG >= 202002L && (defined(__cpp_impl_coroutine) || defined(__cpp_coroutines))
+#if defined(__has_include)
+#if __has_include(<semaphore>)
+#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED 1
+#endif
+#endif
+#endif
+#ifndef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED
+#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED 0
+#endif
+
+#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED
+#include <array>
+#include <atomic>
+#include <coroutine>
+#include <cstddef>
+#include <cstdint>
+#include <semaphore>
+
+BOOST_SML_NAMESPACE_BEGIN
+
+namespace utility {
+
+// Generic completion trigger delivered into a machine's transition table by a
+// co_sm running an external-completion scheduler policy. source_index names the
+// completion_source that fired; machines map it onto domain state (for example
+// a buffer slot) via their own guards and actions.
+struct completion {
+  std::size_t source_index{};
+};
+
+namespace policy {
+
+// One externally fired completion flag. Producers (worker threads, device
+// callbacks) touch only fire(); every other member is single-consumer and runs
+// on the dispatching thread. fire() never resumes a coroutine and never touches
+// the state machine: it flips the flag and releases the scheduler's semaphore,
+// so single-writer dispatch holds structurally no matter which thread fires.
+class completion_source {
+ public:
+  // Trampoline compatible with thread_pool_scheduler::try_submit_with_completion.
+  static void fire(void* ctx) noexcept {
+    auto* source = static_cast<completion_source*>(ctx);
+    source->state_.store(state_fired, std::memory_order_release);
+    source->wakeup_->release();
+  }
+
+  void arm() noexcept { state_.store(state_pending, std::memory_order_relaxed); }
+  void reset() noexcept { state_.store(state_idle, std::memory_order_relaxed); }
+  bool is_fired() const noexcept { return state_.load(std::memory_order_acquire) == state_fired; }
+  bool is_idle() const noexcept { return state_.load(std::memory_order_acquire) == state_idle; }
+
+ private:
+  template <std::size_t>
+  friend class external_completion_scheduler;
+
+  static constexpr std::uint8_t state_idle = 0;
+  static constexpr std::uint8_t state_pending = 1;
+  static constexpr std::uint8_t state_fired = 2;
+
+  std::atomic<std::uint8_t> state_{state_idle};
+  std::counting_semaphore<>* wakeup_ = nullptr;
+};
+
+// Scheduler policy that lets a co_sm dispatch suspend on completions fired by
+// other threads while still returning only when the whole dispatch is done:
+//
+//   1. commit sweep — completions that fired between dispatches are delivered
+//      as explicit `utility::completion` events, ascending index order, before
+//      the trigger event runs;
+//   2. the trigger event dispatches; its actions may arm sources, hand them to
+//      workers, and mark some of them required for this dispatch;
+//   3. required-wait drain — the dispatch coroutine awaits each required
+//      source (lowest fired index first) and re-enters it as a
+//      `utility::completion` event; the top-level process_event parks on the
+//      semaphore and resumes the coroutine on the calling thread until the
+//      root task is ready.
+//
+// All coroutine machinery (park, resume, await bookkeeping) is single-consumer
+// on the dispatching thread; producers interact only through
+// completion_source::fire. The run-to-completion guarantees are honored because
+// nothing escapes the top-level dispatch: required completions are drained
+// before it returns, and unrequired (background) fires persist only as passive
+// flags swept by the next dispatch.
+//
+// Contract: a source marked required must have been handed to a producer that
+// will eventually fire it, or the drain loop cannot finish. Destroying the
+// scheduler while producers still hold source pointers is undefined; owners
+// drain or join in-flight work first.
+template <std::size_t SourceCount = 8>
+class external_completion_scheduler {
+ public:
+  static_assert(SourceCount > 0, "external_completion_scheduler needs at least one source");
+  static_assert(SourceCount <= 64, "required-source bookkeeping is a 64-bit mask");
+
+  static constexpr bool guarantees_fifo = true;
+  static constexpr bool single_consumer = true;
+  static constexpr bool run_to_completion = true;
+  static constexpr bool external_completion = true;
+  static constexpr std::size_t npos = static_cast<std::size_t>(-1);
+
+  using completion_event = completion;
+
+  external_completion_scheduler() noexcept {
+    for (auto& source : sources_) {
+      source.wakeup_ = &wakeup_;
+    }
+  }
+
+  external_completion_scheduler(const external_completion_scheduler&) = delete;
+  external_completion_scheduler& operator=(const external_completion_scheduler&) = delete;
+  external_completion_scheduler(external_completion_scheduler&&) = delete;
+  external_completion_scheduler& operator=(external_completion_scheduler&&) = delete;
+
+  // Continuations run inline on the dispatching thread.
+  template <class TFn>
+  void schedule(TFn&& fn) noexcept(noexcept(static_cast<TFn&&>(fn)())) {
+    static_cast<TFn&&>(fn)();
+  }
+
+  completion_source& source(const std::size_t index) noexcept { return sources_[index]; }
+
+  // Marks a source as blocking the current dispatch. Consumer thread only
+  // (machine actions via an injected scheduler pointer).
+  void require(const std::size_t index) noexcept { required_ |= bit(index); }
+
+  bool has_required() const noexcept { return required_ != 0u; }
+
+  // Lowest fired source not marked required, consumed and returned for
+  // delivery as a completion event; npos when none. Consumer thread only.
+  std::size_t sweep_next_fired() noexcept {
+    for (std::size_t index = 0; index < SourceCount; ++index) {
+      if ((required_ & bit(index)) == 0u && sources_[index].is_fired()) {
+        sources_[index].reset();
+        return index;
+      }
+    }
+    return npos;
+  }
+
+  // Awaitable used by the dispatch coroutine: ready when any required source
+  // has fired; resuming consumes the lowest fired required index.
+  auto next_fired_required() noexcept {
+    struct awaitable {
+      external_completion_scheduler& self;
+      bool await_ready() const noexcept { return self.lowest_required_fired() != npos; }
+      void await_suspend(std::coroutine_handle<> handle) noexcept { self.parked_ = handle; }
+      std::size_t await_resume() noexcept { return self.consume_lowest_required_fired(); }
+    };
+    return awaitable{*this};
+  }
+
+  // Blocks the dispatching thread until some source fires. Permits accumulate,
+  // so a fire that lands before the wait is never lost.
+  void wait_any() noexcept { wakeup_.acquire(); }
+
+  // Resumes the parked dispatch coroutine iff its awaited condition holds.
+  // A permit consumed for a background (unrequired) fire leaves the coroutine
+  // parked; that fire stays set for a later sweep.
+  bool try_resume_parked() noexcept {
+    if (parked_ == nullptr || lowest_required_fired() == npos) {
+      return false;
+    }
+    const std::coroutine_handle<> handle = parked_;
+    parked_ = nullptr;
+    handle.resume();
+    return true;
+  }
+
+ private:
+  static constexpr std::uint64_t bit(const std::size_t index) noexcept { return std::uint64_t{1} << index; }
+
+  std::size_t lowest_required_fired() const noexcept {
+    for (std::size_t index = 0; index < SourceCount; ++index) {
+      if ((required_ & bit(index)) != 0u && sources_[index].is_fired()) {
+        return index;
+      }
+    }
+    return npos;
+  }
+
+  std::size_t consume_lowest_required_fired() noexcept {
+    const std::size_t index = lowest_required_fired();
+    required_ &= ~bit(index);
+    sources_[index].reset();
+    return index;
+  }
+
+  std::array<completion_source, SourceCount> sources_{};
+  std::counting_semaphore<> wakeup_{0};
+  std::uint64_t required_ = 0;
+  std::coroutine_handle<> parked_ = nullptr;
+};
+
+}  // namespace policy
+}  // namespace utility
+
+BOOST_SML_NAMESPACE_END
+#endif
+
+#undef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG
+
+#endif  // BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP
diff --git a/include/stateforward/sml/utility/external_completion.hpp b/include/stateforward/sml/utility/external_completion.hpp
new file mode 100644
index 00000000..2f9d294a
--- /dev/null
+++ b/include/stateforward/sml/utility/external_completion.hpp
@@ -0,0 +1,7 @@
+#ifndef STATEFORWARD_SML_UTILITY_EXTERNAL_COMPLETION_HPP
+#define STATEFORWARD_SML_UTILITY_EXTERNAL_COMPLETION_HPP
+
+#include <boost/sml/utility/external_completion.hpp>
+#include <stateforward/sml.hpp>
+
+#endif
diff --git a/test/ft/CMakeLists.txt b/test/ft/CMakeLists.txt
index 5ac48a60..be82b57c 100644
--- a/test/ft/CMakeLists.txt
+++ b/test/ft/CMakeLists.txt
@@ -23,6 +23,10 @@ add_executable(test_thread_pool_scheduler thread_pool_scheduler.cpp)
 add_test(test_thread_pool_scheduler test_thread_pool_scheduler)
 target_link_libraries(test_thread_pool_scheduler -lpthread)
 
+add_executable(test_external_completion external_completion.cpp)
+add_test(test_external_completion test_external_completion)
+target_link_libraries(test_external_completion -lpthread)
+
 add_executable(test_constexpr constexpr.cpp)
 add_test(test_constexpr test_constexpr)
 
diff --git a/test/ft/external_completion.cpp b/test/ft/external_completion.cpp
new file mode 100644
index 00000000..a1e86986
--- /dev/null
+++ b/test/ft/external_completion.cpp
@@ -0,0 +1,232 @@
+//
+// Copyright (c) 2026 stateforward
+//
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+//
+#include <array>
+#include <atomic>
+#include <boost/sml.hpp>
+#include <boost/sml/utility/co_sm.hpp>
+#include <boost/sml/utility/external_completion.hpp>
+#include <boost/sml/utility/thread_pool_scheduler.hpp>
+#include <chrono>
+#include <cstddef>
+#include <thread>
+
+namespace sml = boost::sml;
+
+#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED && BOOST_SML_UTILITY_CO_SM_ENABLED && \
+    BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED
+namespace utility = sml::utility;
+
+using scheduler_t = utility::policy::external_completion_scheduler<8>;
+using pool_t = utility::policy::thread_pool_scheduler<1>;
+
+static_assert(utility::policy::external_completion_scheduler_contract<scheduler_t>,
+              "external_completion_scheduler must satisfy the external completion contract");
+static_assert(!utility::policy::external_completion_scheduler_contract<utility::policy::inline_scheduler>,
+              "inline_scheduler must not satisfy the external completion contract");
+static_assert(!utility::policy::external_completion_scheduler_contract<utility::policy::fifo_scheduler<>>,
+              "fifo_scheduler must not satisfy the external completion contract");
+static_assert(utility::policy::strict_ordering_scheduler_contract<scheduler_t>,
+              "external completion drains before return, honoring the strict ordering contract");
+
+namespace {
+
+constexpr std::size_t k_probe_marker = 99;
+
+struct test_context {
+  scheduler_t* scheduler = nullptr;
+  pool_t* pool = nullptr;
+  std::array<std::size_t, 16> log{};
+  std::size_t log_count = 0;
+  std::thread::id completion_thread{};
+  bool worker_delay = false;
+};
+
+struct e_require_via_worker {
+  std::size_t count{};
+};
+struct e_require_fired_inline {
+  std::size_t count{};
+};
+struct e_hand_to_worker_unrequired {
+  std::size_t index{};
+};
+struct e_probe {};
+
+void fire_after_optional_delay(test_context& context, const std::size_t index) {
+  const bool delay = context.worker_delay;
+  (void)context.pool->try_submit_with_completion(
+      [delay]() noexcept {
+        if (delay) {
+          const auto until = std::chrono::steady_clock::now() + std::chrono::milliseconds(2);
+          while (std::chrono::steady_clock::now() < until) {
+          }
+        }
+      },
+      &context.scheduler->source(index), &utility::policy::completion_source::fire);
+}
+
+struct machine {
+  auto operator()() const {
+    using namespace sml;
+
+    const auto a_require_via_worker = [](const e_require_via_worker& ev, test_context& context) {
+      for (std::size_t index = 0; index < ev.count; ++index) {
+        context.scheduler->source(index).arm();
+        fire_after_optional_delay(context, index);
+      }
+      for (std::size_t index = 0; index < ev.count; ++index) {
+        context.scheduler->require(index);
+      }
+    };
+
+    const auto a_require_fired_inline = [](const e_require_fired_inline& ev, test_context& context) {
+      for (std::size_t index = 0; index < ev.count; ++index) {
+        context.scheduler->source(index).arm();
+      }
+      // Fire in descending order on this thread: delivery must still ascend.
+      for (std::size_t reverse = ev.count; reverse > 0; --reverse) {
+        utility::policy::completion_source::fire(&context.scheduler->source(reverse - 1));
+      }
+      for (std::size_t index = 0; index < ev.count; ++index) {
+        context.scheduler->require(index);
+      }
+    };
+
+    const auto a_hand_to_worker_unrequired = [](const e_hand_to_worker_unrequired& ev, test_context& context) {
+      context.scheduler->source(ev.index).arm();
+      fire_after_optional_delay(context, ev.index);
+    };
+
+    const auto a_probe = [](const e_probe&, test_context& context) { context.log[context.log_count++] = k_probe_marker; };
+
+    const auto a_record = [](const utility::completion& ev, test_context& context) {
+      context.log[context.log_count++] = ev.source_index;
+      context.completion_thread = std::this_thread::get_id();
+    };
+
+    // clang-format off
+    return make_transition_table(
+      *"ready"_s + event<e_require_via_worker> / a_require_via_worker
+      , "ready"_s + event<e_require_fired_inline> / a_require_fired_inline
+      , "ready"_s + event<e_hand_to_worker_unrequired> / a_hand_to_worker_unrequired
+      , "ready"_s + event<e_probe> / a_probe
+      , "ready"_s + event<utility::completion> / a_record
+    );
+    // clang-format on
+  }
+};
+
+using co_sm_t = utility::co_sm<machine, utility::policy::coroutine_scheduler<scheduler_t>,
+                               utility::policy::coroutine_allocator<utility::policy::pooled_coroutine_allocator<>>,
+                               test_context>;
+
+struct fixture {
+  pool_t pool{};
+  test_context context{};
+  co_sm_t machine_instance;
+
+  fixture() : machine_instance{context} {
+    context.scheduler = &machine_instance.scheduler();
+    context.pool = &pool;
+  }
+};
+
+}  // namespace
+
+test external_completion_required_drained_before_return = [] {
+  fixture f{};
+  f.context.worker_delay = true;  // force the dispatch coroutine to park
+
+  expect(f.machine_instance.process_event(e_require_via_worker{3}));
+
+  // Everything observable happened inside the dispatch: all three completions
+  // were delivered, nothing remains required, and the sources are reusable.
+  expect(3u == f.context.log_count);
+  expect(!f.machine_instance.scheduler().has_required());
+  for (std::size_t index = 0; index < 3; ++index) {
+    expect(f.machine_instance.scheduler().source(index).is_idle());
+  }
+};
+
+test external_completion_delivers_on_dispatching_thread = [] {
+  fixture f{};
+  f.context.worker_delay = true;
+
+  expect(f.machine_instance.process_event(e_require_via_worker{1}));
+
+  expect(1u == f.context.log_count);
+  expect(std::this_thread::get_id() == f.context.completion_thread);
+};
+
+test external_completion_prefired_sources_deliver_ascending_without_park = [] {
+  fixture f{};
+
+  // Sources fire 2,1,0 on this thread before the drain runs: no suspension,
+  // and delivery is deterministic ascending regardless of fire order.
+  expect(f.machine_instance.process_event(e_require_fired_inline{3}));
+
+  expect(3u == f.context.log_count);
+  expect(0u == f.context.log[0]);
+  expect(1u == f.context.log[1]);
+  expect(2u == f.context.log[2]);
+};
+
+test external_completion_background_fire_swept_before_next_trigger = [] {
+  fixture f{};
+
+  // Dispatch 1 hands source 5 to a worker without requiring it; the dispatch
+  // returns while the fire is still pending.
+  expect(f.machine_instance.process_event(e_hand_to_worker_unrequired{5}));
+  expect(0u == f.context.log_count);
+
+  while (!f.machine_instance.scheduler().source(5).is_fired()) {
+    std::this_thread::yield();
+  }
+
+  // Dispatch 2: the commit sweep delivers completion(5) before the trigger.
+  expect(f.machine_instance.process_event(e_probe{}));
+  expect(2u == f.context.log_count);
+  expect(5u == f.context.log[0]);
+  expect(k_probe_marker == f.context.log[1]);
+  expect(f.machine_instance.scheduler().source(5).is_idle());
+};
+
+test external_completion_dispatch_without_requires_never_parks = [] {
+  fixture f{};
+
+  expect(f.machine_instance.process_event(e_probe{}));
+
+  expect(1u == f.context.log_count);
+  expect(k_probe_marker == f.context.log[0]);
+};
+
+test external_completion_async_wrapper_completes_inline = [] {
+  fixture f{};
+  f.context.worker_delay = true;
+
+  utility::bool_task task = f.machine_instance.process_event_async(e_require_via_worker{2});
+
+  expect(task.await_ready());
+  expect(task.result());
+  expect(2u == f.context.log_count);
+};
+
+test external_completion_sources_reusable_across_dispatches = [] {
+  fixture f{};
+
+  for (std::size_t round = 0; round < 3; ++round) {
+    f.context.log_count = 0;
+    expect(f.machine_instance.process_event(e_require_fired_inline{2}));
+    expect(2u == f.context.log_count);
+    expect(0u == f.context.log[0]);
+    expect(1u == f.context.log[1]);
+  }
+};
+#else
+test external_completion_disabled_without_cxx20_coroutines_and_semaphore = [] { expect(true); };
+#endif

From 8968306bc8ec59602e6863a5d45cd87a6cead9cd Mon Sep 17 00:00:00 2001
From: gabewillen <gabewillen@gmail.com>
Date: Thu, 2 Jul 2026 21:03:47 +0000
Subject: [PATCH 2/4] Address review: fire idempotence, abort-safe dispatch
 state, by-ref trigger

- completion_source::fire releases a wakeup permit only on the transition
  into fired (exchange-gated), so a double-firing producer cannot accumulate
  permits or overflow the semaphore max count.
- external_completion_scheduler::reset_dispatch_state clears required bits
  and any parked handle at the top of every external dispatch: bookkeeping
  from a dispatch aborted mid-drain (for example a machine action that threw
  after require()) can no longer block the next dispatch's drain. Added to
  the scheduler contract and covered by a regression test that aborts after
  require() and proves the next dispatch completes.
- process_event_external_impl takes the trigger event by const reference:
  the coroutine always completes inside the enclosing process_event, so the
  caller's event outlives the frame and non-copyable events stay supported.
- clang-format over the touched files (fixes the Quality gates CI job).
---
 include/boost/sml/utility/co_sm.hpp           | 35 +++++----
 .../boost/sml/utility/external_completion.hpp | 17 +++-
 test/ft/external_completion.cpp               | 78 ++++++++++++++++++-
 3 files changed, 110 insertions(+), 20 deletions(-)

diff --git a/include/boost/sml/utility/co_sm.hpp b/include/boost/sml/utility/co_sm.hpp
index 4f645402..becef6d3 100644
--- a/include/boost/sml/utility/co_sm.hpp
+++ b/include/boost/sml/utility/co_sm.hpp
@@ -309,19 +309,18 @@ concept has_try_run_immediate = requires(TScheduler scheduler) {
 // strict ordering contract: required completions are drained before the
 // top-level dispatch returns, and delivery order is deterministic.
 template <class TScheduler>
-concept external_completion_scheduler_contract =
-    requires {
-      { TScheduler::external_completion } -> std::convertible_to<bool>;
-      { TScheduler::npos } -> std::convertible_to<std::size_t>;
-      typename TScheduler::completion_event;
-    } && static_cast<bool>(TScheduler::external_completion) &&
-    requires(TScheduler scheduler) {
-      { scheduler.has_required() } -> std::convertible_to<bool>;
-      { scheduler.sweep_next_fired() } -> std::same_as<std::size_t>;
-      scheduler.next_fired_required();
-      scheduler.wait_any();
-      { scheduler.try_resume_parked() } -> std::same_as<bool>;
-    };
+concept external_completion_scheduler_contract = requires {
+  { TScheduler::external_completion } -> std::convertible_to<bool>;
+  { TScheduler::npos } -> std::convertible_to<std::size_t>;
+  typename TScheduler::completion_event;
+} && static_cast<bool>(TScheduler::external_completion) && requires(TScheduler scheduler) {
+  { scheduler.has_required() } -> std::convertible_to<bool>;
+  { scheduler.sweep_next_fired() } -> std::same_as<std::size_t>;
+  scheduler.next_fired_required();
+  scheduler.wait_any();
+  { scheduler.try_resume_parked() } -> std::same_as<bool>;
+  scheduler.reset_dispatch_state();
+};
 
 template <class TAllocatorPolicy>
 concept valid_coroutine_allocator_policy = requires { typename TAllocatorPolicy::allocator_type; };
@@ -602,7 +601,9 @@ class co_sm {
       // The dispatch coroutine may suspend awaiting externally fired
       // completions; drive it to completion here so the run-to-completion
       // contract holds: nothing observable escapes this call, and resumption
-      // happens only on this (the dispatching) thread.
+      // happens only on this (the dispatching) thread. Any required bits or
+      // parked handle left by a previous aborted dispatch are cleared first.
+      scheduler_.reset_dispatch_state();
       bool_task task = process_event_external_impl(std::allocator_arg, allocator_, *this, event);
       while (!task.await_ready()) {
         scheduler_.wait_any();
@@ -668,8 +669,12 @@ class co_sm {
   // sources the machine marked required are awaited (suspending this coroutine)
   // and re-entered as completion events until none remain. The caller's drive
   // loop in process_event resumes the coroutine on the dispatching thread only.
+  // The trigger event is taken by reference: the coroutine is always driven
+  // to completion inside the enclosing process_event call, so the caller's
+  // event outlives the frame and non-copyable event types stay supported.
   template <class TEvent>
-  static bool_task process_event_external_impl(std::allocator_arg_t, allocator_type& allocator, co_sm& self, TEvent event)
+  static bool_task process_event_external_impl(std::allocator_arg_t, allocator_type& allocator, co_sm& self,
+                                               const TEvent& event)
     requires policy::external_completion_scheduler_contract<scheduler_type>
   {
     (void)allocator;
diff --git a/include/boost/sml/utility/external_completion.hpp b/include/boost/sml/utility/external_completion.hpp
index eb59594f..854df95c 100644
--- a/include/boost/sml/utility/external_completion.hpp
+++ b/include/boost/sml/utility/external_completion.hpp
@@ -60,10 +60,14 @@ namespace policy {
 class completion_source {
  public:
   // Trampoline compatible with thread_pool_scheduler::try_submit_with_completion.
+  // Only the transition into fired releases a wakeup permit, so a misbehaving
+  // producer that fires twice cannot accumulate permits or overflow the
+  // semaphore's max count.
   static void fire(void* ctx) noexcept {
     auto* source = static_cast<completion_source*>(ctx);
-    source->state_.store(state_fired, std::memory_order_release);
-    source->wakeup_->release();
+    if (source->state_.exchange(state_fired, std::memory_order_release) != state_fired) {
+      source->wakeup_->release();
+    }
   }
 
   void arm() noexcept { state_.store(state_pending, std::memory_order_relaxed); }
@@ -141,6 +145,15 @@ class external_completion_scheduler {
 
   completion_source& source(const std::size_t index) noexcept { return sources_[index]; }
 
+  // Clears bookkeeping an aborted dispatch may have left behind (for example
+  // a state-machine action that threw after require()): required bits and a
+  // parked handle are meaningless outside the dispatch that created them, and
+  // stale required bits would otherwise block the next drain forever.
+  void reset_dispatch_state() noexcept {
+    required_ = 0;
+    parked_ = nullptr;
+  }
+
   // Marks a source as blocking the current dispatch. Consumer thread only
   // (machine actions via an injected scheduler pointer).
   void require(const std::size_t index) noexcept { required_ |= bit(index); }
diff --git a/test/ft/external_completion.cpp b/test/ft/external_completion.cpp
index a1e86986..9820d023 100644
--- a/test/ft/external_completion.cpp
+++ b/test/ft/external_completion.cpp
@@ -13,6 +13,7 @@
 #include <boost/sml/utility/thread_pool_scheduler.hpp>
 #include <chrono>
 #include <cstddef>
+#include <stdexcept>
 #include <thread>
 
 namespace sml = boost::sml;
@@ -121,9 +122,9 @@ struct machine {
   }
 };
 
-using co_sm_t = utility::co_sm<machine, utility::policy::coroutine_scheduler<scheduler_t>,
-                               utility::policy::coroutine_allocator<utility::policy::pooled_coroutine_allocator<>>,
-                               test_context>;
+using co_sm_t =
+    utility::co_sm<machine, utility::policy::coroutine_scheduler<scheduler_t>,
+                   utility::policy::coroutine_allocator<utility::policy::pooled_coroutine_allocator<>>, test_context>;
 
 struct fixture {
   pool_t pool{};
@@ -216,6 +217,77 @@ test external_completion_async_wrapper_completes_inline = [] {
   expect(2u == f.context.log_count);
 };
 
+test external_completion_double_fire_releases_one_permit = [] {
+  fixture f{};
+
+  // A misbehaving producer fires the same source twice: only the transition
+  // into fired may release a permit, so exactly one completion is delivered
+  // and no stray permit disturbs a later dispatch.
+  f.context.scheduler->source(2).arm();
+  utility::policy::completion_source::fire(&f.context.scheduler->source(2));
+  utility::policy::completion_source::fire(&f.context.scheduler->source(2));
+
+  expect(f.machine_instance.process_event(e_probe{}));
+  expect(2u == f.context.log_count);
+  expect(2u == f.context.log[0]);
+  expect(k_probe_marker == f.context.log[1]);
+
+  f.context.log_count = 0;
+  expect(f.machine_instance.process_event(e_require_fired_inline{1}));
+  expect(1u == f.context.log_count);
+  expect(0u == f.context.log[0]);
+};
+
+#if !defined(BOOST_SML_DISABLE_EXCEPTIONS) || !BOOST_SML_DISABLE_EXCEPTIONS
+struct e_require_then_throw {};
+
+test external_completion_aborted_dispatch_does_not_poison_next = [] {
+  fixture f{};
+
+  struct throwing_machine {
+    auto operator()() const {
+      using namespace sml;
+      const auto a_require_then_throw = [](const e_require_then_throw&, test_context& context) {
+        context.scheduler->require(4);  // never armed, never fired
+        throw std::runtime_error{"abort after require"};
+      };
+      const auto a_probe = [](const e_probe&, test_context& context) { context.log[context.log_count++] = k_probe_marker; };
+      const auto a_record = [](const utility::completion& ev, test_context& context) {
+        context.log[context.log_count++] = ev.source_index;
+      };
+      // clang-format off
+      return make_transition_table(
+        *"ready"_s + event<e_require_then_throw> / a_require_then_throw
+        , "ready"_s + event<e_probe> / a_probe
+        , "ready"_s + event<utility::completion> / a_record
+      );
+      // clang-format on
+    }
+  };
+
+  using throwing_co_sm =
+      utility::co_sm<throwing_machine, utility::policy::coroutine_scheduler<scheduler_t>,
+                     utility::policy::coroutine_allocator<utility::policy::pooled_coroutine_allocator<>>, test_context>;
+  test_context context{};
+  throwing_co_sm machine{context};
+  context.scheduler = &machine.scheduler();
+
+  bool threw = false;
+  try {
+    (void)machine.process_event(e_require_then_throw{});
+  } catch (const std::runtime_error&) {
+    threw = true;
+  }
+  expect(threw);
+
+  // Pre-fix, the stale required bit for the never-fired source 4 would make
+  // this dispatch's drain block forever; post-fix it completes immediately.
+  expect(machine.process_event(e_probe{}));
+  expect(1u == context.log_count);
+  expect(k_probe_marker == context.log[0]);
+};
+#endif
+
 test external_completion_sources_reusable_across_dispatches = [] {
   fixture f{};
 

From 2b416907601b3c039382414b5359be08d7d72b08 Mon Sep 17 00:00:00 2001
From: gabewillen <gabewillen@gmail.com>
Date: Thu, 2 Jul 2026 21:17:26 +0000
Subject: [PATCH 3/4] Address review: nested-dispatch safety; replace coroutine
 drive with plain loop

Bugbot (high): a nested synchronous process_event from a transition action
cleared the outer dispatch's required bits via reset_dispatch_state, letting
the outer drain exit before its completions arrived. Only the outermost
dispatch now owns reset/sweep/drain (depth-guarded); nested calls dispatch
their trigger plainly and anything they require is drained by the enclosing
dispatch. Regression test covers a nested probe issued after require().

Performance: the synchronous drive needs no coroutine at all - process_event
blocks until done by contract, so suspend/park/resume machinery was pure
overhead. The external path is now a plain loop (sweep -> trigger -> drain
via consume_next_fired_required + wait_any) with identical semantics: RTC,
ascending deterministic delivery, dispatch-thread-only execution. The
awaitable, parked handle, and per-dispatch coroutine frame are gone, and the
scheduler contract shrinks accordingly.

Measured (Ryzen 5950X, -O3, 10M dispatches): inline baseline 0.2 ns/op,
external fast path (no completions) 2.6 ns/op - one 8-source sweep scan plus
a mask check - and a full arm+fire+require+drain round trip 5.7 ns/op. The
blocking path is semaphore-wait dominated, i.e. the external work being
awaited, not scheduler overhead.
---
 include/boost/sml/utility/co_sm.hpp           | 84 +++++++++++--------
 .../boost/sml/utility/external_completion.hpp | 81 +++++++-----------
 test/ft/external_completion.cpp               | 65 ++++++++++++++
 3 files changed, 141 insertions(+), 89 deletions(-)

diff --git a/include/boost/sml/utility/co_sm.hpp b/include/boost/sml/utility/co_sm.hpp
index becef6d3..cbe6c780 100644
--- a/include/boost/sml/utility/co_sm.hpp
+++ b/include/boost/sml/utility/co_sm.hpp
@@ -316,9 +316,8 @@ concept external_completion_scheduler_contract = requires {
 } && static_cast<bool>(TScheduler::external_completion) && requires(TScheduler scheduler) {
   { scheduler.has_required() } -> std::convertible_to<bool>;
   { scheduler.sweep_next_fired() } -> std::same_as<std::size_t>;
-  scheduler.next_fired_required();
+  { scheduler.consume_next_fired_required() } -> std::same_as<std::size_t>;
   scheduler.wait_any();
-  { scheduler.try_resume_parked() } -> std::same_as<bool>;
   scheduler.reset_dispatch_state();
 };
 
@@ -598,18 +597,50 @@ class co_sm {
   template <class TEvent>
   bool process_event(const TEvent& event) {
     if constexpr (policy::external_completion_scheduler_contract<scheduler_type>) {
-      // The dispatch coroutine may suspend awaiting externally fired
-      // completions; drive it to completion here so the run-to-completion
-      // contract holds: nothing observable escapes this call, and resumption
-      // happens only on this (the dispatching) thread. Any required bits or
-      // parked handle left by a previous aborted dispatch are cleared first.
+      // Only the outermost dispatch owns the completion machinery: a nested
+      // synchronous process_event from a transition action dispatches its
+      // trigger plainly, and any sources it marks required are drained by the
+      // enclosing dispatch before that dispatch returns. Resetting or
+      // draining from the nested call would clear required bits the outer
+      // drain still needs.
+      if (external_dispatch_depth_ > 0) {
+        return state_machine_.process_event(event);
+      }
+
+      struct dispatch_depth_scope {
+        std::size_t& depth;
+        ~dispatch_depth_scope() noexcept { --depth; }
+      };
+      ++external_dispatch_depth_;
+      dispatch_depth_scope depth_scope{external_dispatch_depth_};
+
+      // Synchronous drive with zero coroutine machinery: (1) commit sweep -
+      // completions that fired between dispatches are delivered as explicit
+      // completion events in ascending source order before the trigger runs;
+      // (2) the trigger dispatches, its actions may arm sources, hand them to
+      // workers, and mark them required; (3) the drain loop blocks on the
+      // scheduler's semaphore and delivers each required completion (lowest
+      // fired index first) until none remain, so nothing observable escapes
+      // this call and delivery happens only on this (the dispatching)
+      // thread. Required bits left by a previous aborted dispatch are
+      // cleared first. The fast path (no fired and no required sources) costs
+      // one sweep scan plus one mask check over the inline dispatch.
       scheduler_.reset_dispatch_state();
-      bool_task task = process_event_external_impl(std::allocator_arg, allocator_, *this, event);
-      while (!task.await_ready()) {
-        scheduler_.wait_any();
-        (void)scheduler_.try_resume_parked();
+      using completion_event_t = typename scheduler_type::completion_event;
+      for (std::size_t index = scheduler_.sweep_next_fired(); index != scheduler_type::npos;
+           index = scheduler_.sweep_next_fired()) {
+        (void)state_machine_.process_event(completion_event_t{index});
+      }
+      bool accepted = state_machine_.process_event(event);
+      while (scheduler_.has_required()) {
+        const std::size_t index = scheduler_.consume_next_fired_required();
+        if (index == scheduler_type::npos) {
+          scheduler_.wait_any();
+          continue;
+        }
+        accepted = state_machine_.process_event(completion_event_t{index}) && accepted;
       }
-      return task.result();
+      return accepted;
     } else {
       return state_machine_.process_event(event);
     }
@@ -669,31 +700,6 @@ class co_sm {
   // sources the machine marked required are awaited (suspending this coroutine)
   // and re-entered as completion events until none remain. The caller's drive
   // loop in process_event resumes the coroutine on the dispatching thread only.
-  // The trigger event is taken by reference: the coroutine is always driven
-  // to completion inside the enclosing process_event call, so the caller's
-  // event outlives the frame and non-copyable event types stay supported.
-  template <class TEvent>
-  static bool_task process_event_external_impl(std::allocator_arg_t, allocator_type& allocator, co_sm& self,
-                                               const TEvent& event)
-    requires policy::external_completion_scheduler_contract<scheduler_type>
-  {
-    (void)allocator;
-    using completion_event_t = typename scheduler_type::completion_event;
-
-    for (std::size_t index = self.scheduler_.sweep_next_fired(); index != scheduler_type::npos;
-         index = self.scheduler_.sweep_next_fired()) {
-      (void)self.state_machine_.process_event(completion_event_t{index});
-    }
-
-    bool accepted = self.state_machine_.process_event(event);
-
-    while (self.scheduler_.has_required()) {
-      const std::size_t index = co_await self.scheduler_.next_fired_required();
-      accepted = self.state_machine_.process_event(completion_event_t{index}) && accepted;
-    }
-
-    co_return accepted;
-  }  // GCOVR_EXCL_LINE
 
   template <class TEvent>
   struct process_event_awaitable {
@@ -738,6 +744,10 @@ class co_sm {
   state_machine_type state_machine_{};
   scheduler_type scheduler_{};
   allocator_type allocator_{};
+  // Depth of synchronous external-completion dispatches on this machine;
+  // nested calls (actions re-entering process_event) skip reset/sweep/drain
+  // so the outermost dispatch's run-to-completion drain stays intact.
+  std::size_t external_dispatch_depth_ = 0;
 };
 
 }  // namespace utility
diff --git a/include/boost/sml/utility/external_completion.hpp b/include/boost/sml/utility/external_completion.hpp
index 854df95c..15bc0c1f 100644
--- a/include/boost/sml/utility/external_completion.hpp
+++ b/include/boost/sml/utility/external_completion.hpp
@@ -33,7 +33,6 @@
 #if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED
 #include <array>
 #include <atomic>
-#include <coroutine>
 #include <cstddef>
 #include <cstdint>
 #include <semaphore>
@@ -87,7 +86,7 @@ class completion_source {
   std::counting_semaphore<>* wakeup_ = nullptr;
 };
 
-// Scheduler policy that lets a co_sm dispatch suspend on completions fired by
+// Scheduler policy that lets a co_sm dispatch block on completions fired by
 // other threads while still returning only when the whole dispatch is done:
 //
 //   1. commit sweep — completions that fired between dispatches are delivered
@@ -95,18 +94,17 @@ class completion_source {
 //      the trigger event runs;
 //   2. the trigger event dispatches; its actions may arm sources, hand them to
 //      workers, and mark some of them required for this dispatch;
-//   3. required-wait drain — the dispatch coroutine awaits each required
-//      source (lowest fired index first) and re-enters it as a
-//      `utility::completion` event; the top-level process_event parks on the
-//      semaphore and resumes the coroutine on the calling thread until the
-//      root task is ready.
+//   3. required-wait drain — the dispatching thread consumes each required
+//      source (lowest fired index first) as a `utility::completion` event,
+//      blocking on the semaphore between fires, until none remain.
 //
-// All coroutine machinery (park, resume, await bookkeeping) is single-consumer
-// on the dispatching thread; producers interact only through
-// completion_source::fire. The run-to-completion guarantees are honored because
-// nothing escapes the top-level dispatch: required completions are drained
-// before it returns, and unrequired (background) fires persist only as passive
-// flags swept by the next dispatch.
+// All bookkeeping is single-consumer on the dispatching thread; producers
+// interact only through completion_source::fire. The run-to-completion
+// guarantees are honored because nothing escapes the top-level dispatch:
+// required completions are drained before it returns, and unrequired
+// (background) fires persist only as passive flags swept by the next dispatch.
+// The drive is a plain loop, so the no-completion fast path adds only one
+// sweep scan and one mask check over an inline dispatch.
 //
 // Contract: a source marked required must have been handed to a producer that
 // will eventually fire it, or the drain loop cannot finish. Destroying the
@@ -146,13 +144,10 @@ class external_completion_scheduler {
   completion_source& source(const std::size_t index) noexcept { return sources_[index]; }
 
   // Clears bookkeeping an aborted dispatch may have left behind (for example
-  // a state-machine action that threw after require()): required bits and a
-  // parked handle are meaningless outside the dispatch that created them, and
-  // stale required bits would otherwise block the next drain forever.
-  void reset_dispatch_state() noexcept {
-    required_ = 0;
-    parked_ = nullptr;
-  }
+  // a state-machine action that threw after require()): required bits are
+  // meaningless outside the dispatch that created them, and stale bits would
+  // otherwise block the next drain forever.
+  void reset_dispatch_state() noexcept { required_ = 0; }
 
   // Marks a source as blocking the current dispatch. Consumer thread only
   // (machine actions via an injected scheduler pointer).
@@ -172,35 +167,25 @@ class external_completion_scheduler {
     return npos;
   }
 
-  // Awaitable used by the dispatch coroutine: ready when any required source
-  // has fired; resuming consumes the lowest fired required index.
-  auto next_fired_required() noexcept {
-    struct awaitable {
-      external_completion_scheduler& self;
-      bool await_ready() const noexcept { return self.lowest_required_fired() != npos; }
-      void await_suspend(std::coroutine_handle<> handle) noexcept { self.parked_ = handle; }
-      std::size_t await_resume() noexcept { return self.consume_lowest_required_fired(); }
-    };
-    return awaitable{*this};
+  // Lowest fired source marked required, consumed (required bit cleared,
+  // source reset) and returned for delivery as a completion event; npos when
+  // none has fired yet. Consumer thread only.
+  std::size_t consume_next_fired_required() noexcept {
+    const std::size_t index = lowest_required_fired();
+    if (index == npos) {
+      return npos;
+    }
+    required_ &= ~bit(index);
+    sources_[index].reset();
+    return index;
   }
 
   // Blocks the dispatching thread until some source fires. Permits accumulate,
-  // so a fire that lands before the wait is never lost.
+  // so a fire that lands before the wait is never lost; a permit consumed for
+  // a background (unrequired) fire simply re-checks and waits again, leaving
+  // that fire set for a later sweep.
   void wait_any() noexcept { wakeup_.acquire(); }
 
-  // Resumes the parked dispatch coroutine iff its awaited condition holds.
-  // A permit consumed for a background (unrequired) fire leaves the coroutine
-  // parked; that fire stays set for a later sweep.
-  bool try_resume_parked() noexcept {
-    if (parked_ == nullptr || lowest_required_fired() == npos) {
-      return false;
-    }
-    const std::coroutine_handle<> handle = parked_;
-    parked_ = nullptr;
-    handle.resume();
-    return true;
-  }
-
  private:
   static constexpr std::uint64_t bit(const std::size_t index) noexcept { return std::uint64_t{1} << index; }
 
@@ -213,17 +198,9 @@ class external_completion_scheduler {
     return npos;
   }
 
-  std::size_t consume_lowest_required_fired() noexcept {
-    const std::size_t index = lowest_required_fired();
-    required_ &= ~bit(index);
-    sources_[index].reset();
-    return index;
-  }
-
   std::array<completion_source, SourceCount> sources_{};
   std::counting_semaphore<> wakeup_{0};
   std::uint64_t required_ = 0;
-  std::coroutine_handle<> parked_ = nullptr;
 };
 
 }  // namespace policy
diff --git a/test/ft/external_completion.cpp b/test/ft/external_completion.cpp
index 9820d023..66365e94 100644
--- a/test/ft/external_completion.cpp
+++ b/test/ft/external_completion.cpp
@@ -288,6 +288,71 @@ test external_completion_aborted_dispatch_does_not_poison_next = [] {
 };
 #endif
 
+struct e_require_then_nested_probe {};
+
+test external_completion_nested_dispatch_preserves_outer_drain = [] {
+  struct nested_context {
+    scheduler_t* scheduler = nullptr;
+    pool_t* pool = nullptr;
+    std::array<std::size_t, 16> log{};
+    std::size_t log_count = 0;
+    void* self = nullptr;
+    bool (*nested_probe)(void*) = nullptr;
+  };
+
+  struct nested_machine {
+    auto operator()() const {
+      using namespace sml;
+      const auto a_require_then_nested_probe = [](const e_require_then_nested_probe&, nested_context& context) {
+        context.scheduler->source(0).arm();
+        (void)context.pool->try_submit_with_completion(
+            []() noexcept {
+              const auto until = std::chrono::steady_clock::now() + std::chrono::milliseconds(2);
+              while (std::chrono::steady_clock::now() < until) {
+              }
+            },
+            &context.scheduler->source(0), &utility::policy::completion_source::fire);
+        context.scheduler->require(0);
+        // Synchronous re-entry: the nested dispatch must not clear the
+        // required bit the outer drain is about to wait on.
+        (void)context.nested_probe(context.self);
+      };
+      const auto a_probe = [](const e_probe&, nested_context& context) { context.log[context.log_count++] = k_probe_marker; };
+      const auto a_record = [](const utility::completion& ev, nested_context& context) {
+        context.log[context.log_count++] = ev.source_index;
+      };
+      // clang-format off
+      return make_transition_table(
+        *"ready"_s + event<e_require_then_nested_probe> / a_require_then_nested_probe
+        , "ready"_s + event<e_probe> / a_probe
+        , "ready"_s + event<utility::completion> / a_record
+      );
+      // clang-format on
+    }
+  };
+
+  using nested_co_sm =
+      utility::co_sm<nested_machine, utility::policy::coroutine_scheduler<scheduler_t>,
+                     utility::policy::coroutine_allocator<utility::policy::pooled_coroutine_allocator<>>, nested_context>;
+  pool_t pool{};
+  nested_context context{};
+  nested_co_sm machine{context};
+  context.scheduler = &machine.scheduler();
+  context.pool = &pool;
+  context.self = &machine;
+  context.nested_probe = [](void* self) { return static_cast<nested_co_sm*>(self)->process_event(e_probe{}); };
+
+  expect(machine.process_event(e_require_then_nested_probe{}));
+
+  // The nested probe ran inside the outer dispatch, and the outer drain still
+  // delivered the required completion before returning.
+  expect(2u == context.log_count);
+  expect(k_probe_marker == context.log[0]);
+  expect(0u == context.log[1]);
+  expect(!machine.scheduler().has_required());
+  expect(machine.scheduler().source(0).is_idle());
+};
+
 test external_completion_sources_reusable_across_dispatches = [] {
   fixture f{};
 

From 2508a25ac9e61a4466733f7b4c4242df2c772b45 Mon Sep 17 00:00:00 2001
From: gabewillen <gabewillen@gmail.com>
Date: Thu, 2 Jul 2026 23:03:35 +0000
Subject: [PATCH 4/4] Address review: burn wakeup permits when consuming fired
 flags

Every successful fire() releases one wakeup permit, but the commit sweep
consumed fired flags without their permits, so background fires swept
between dispatches leaked permits for the scheduler's lifetime (spurious
drain wakeups, and in the theoretical extreme a noexcept release() past the
semaphore max). consume_fired() now resets the flag and try_acquires the
permit in one place, shared by the sweep and the required-drain consume.
Fired flags remain the ground truth - the drain blocks only after finding no
consumable flag - so permit accounting can never cause a hang. Regression
test runs 10k background fire+sweep cycles and proves a worker-fired
required dispatch still drains correctly.
---
 .../boost/sml/utility/external_completion.hpp | 15 ++++++++++--
 test/ft/external_completion.cpp               | 23 +++++++++++++++++++
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/include/boost/sml/utility/external_completion.hpp b/include/boost/sml/utility/external_completion.hpp
index 15bc0c1f..c47d6984 100644
--- a/include/boost/sml/utility/external_completion.hpp
+++ b/include/boost/sml/utility/external_completion.hpp
@@ -160,7 +160,7 @@ class external_completion_scheduler {
   std::size_t sweep_next_fired() noexcept {
     for (std::size_t index = 0; index < SourceCount; ++index) {
       if ((required_ & bit(index)) == 0u && sources_[index].is_fired()) {
-        sources_[index].reset();
+        consume_fired(index);
         return index;
       }
     }
@@ -176,7 +176,7 @@ class external_completion_scheduler {
       return npos;
     }
     required_ &= ~bit(index);
-    sources_[index].reset();
+    consume_fired(index);
     return index;
   }
 
@@ -189,6 +189,17 @@ class external_completion_scheduler {
  private:
   static constexpr std::uint64_t bit(const std::size_t index) noexcept { return std::uint64_t{1} << index; }
 
+  // Consuming a fired flag also burns the wakeup permit its fire released
+  // (non-blocking: the permit may already have been spent by a blocking wait
+  // in the drain loop). Fired flags are the ground truth - the drain only
+  // blocks after finding no consumable flag - so permit accounting cannot
+  // cause a hang; without this burn, background fires swept between
+  // dispatches would leak permits for the scheduler's lifetime.
+  void consume_fired(const std::size_t index) noexcept {
+    sources_[index].reset();
+    (void)wakeup_.try_acquire();
+  }
+
   std::size_t lowest_required_fired() const noexcept {
     for (std::size_t index = 0; index < SourceCount; ++index) {
       if ((required_ & bit(index)) != 0u && sources_[index].is_fired()) {
diff --git a/test/ft/external_completion.cpp b/test/ft/external_completion.cpp
index 66365e94..7682b2a9 100644
--- a/test/ft/external_completion.cpp
+++ b/test/ft/external_completion.cpp
@@ -353,6 +353,29 @@ test external_completion_nested_dispatch_preserves_outer_drain = [] {
   expect(machine.scheduler().source(0).is_idle());
 };
 
+test external_completion_background_fires_do_not_leak_permits = [] {
+  fixture f{};
+
+  // Each background fire releases one wakeup permit; the sweep that consumes
+  // its flag must burn that permit too, or permits accumulate for the
+  // scheduler's lifetime. After many cycles a worker-fired required dispatch
+  // must still drain correctly (a leak would surface as the drain loop
+  // spinning on stale permits or, at the extreme, fire() terminating on
+  // semaphore overflow).
+  for (std::size_t cycle = 0; cycle < 10000; ++cycle) {
+    f.context.scheduler->source(3).arm();
+    utility::policy::completion_source::fire(&f.context.scheduler->source(3));
+    expect(f.machine_instance.process_event(e_probe{}));
+    f.context.log_count = 0;
+  }
+
+  f.context.worker_delay = true;
+  expect(f.machine_instance.process_event(e_require_via_worker{2}));
+  expect(2u == f.context.log_count);
+  expect(0u == f.context.log[0]);
+  expect(1u == f.context.log[1]);
+};
+
 test external_completion_sources_reusable_across_dispatches = [] {
   fixture f{};