fix(coro): revalidate task-group final wake

CHANGELOG.md

@@ -17,6 +17,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Task-group final-wake eligibility**: A child completion that previously
+  observed an outstanding count of zero now revalidates that zero while
+  selecting the join waiter. A child accepted before join closes admission can
+  no longer be bypassed by an older zero transition (#1056).
 - **Completion waiter wake lifetime**: Join handles, task handles, task groups,
   object-cache release waits, and RDMA pump-exit waits now retain a slot-owned
   selected-wake lease between dequeue and scheduling. Destroying an awaiting

include/elio/coro/detail/completion_waiter.hpp

@@ -115,8 +115,15 @@ class completion_waiter_slot {
     }
 
     completion_wake_lease take() noexcept {
+        return take_if([] { return true; });
+    }
+
+    /// Select the registered waiter only while `ready` is still true under
+    /// the same slot lock used by waiter registration.
+    template<typename Ready>
+    completion_wake_lease take_if(Ready&& ready) noexcept {
         std::lock_guard<std::mutex> lock(mutex_);
-        if (!waiter_ || selected()) {
+        if (!waiter_ || selected() || !std::forward<Ready>(ready)()) {
             return {};
         }
 

include/elio/coro/task_group.hpp

@@ -65,6 +65,8 @@ namespace detail {
 inline std::atomic<bool> pause_before_task_group_completion_for_test{false};
 inline std::atomic<bool> task_group_completion_paused_for_test{false};
 inline std::atomic<bool> task_group_join_observed_pending_for_test{false};
+inline std::atomic<bool> pause_before_task_group_final_wake_for_test{false};
+inline std::atomic<bool> task_group_final_wake_paused_for_test{false};
 #endif
 
 [[nodiscard]] inline bool is_scheduler_worker(
@@ -166,7 +168,29 @@ class task_group_completion_state final {
         }
 
         if (completed) {
-            auto wake = all_done_waiter_.take();
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+            if (pause_before_task_group_final_wake_for_test.load(
+                    std::memory_order_acquire)) {
+                task_group_final_wake_paused_for_test.store(
+                    true, std::memory_order_release);
+                task_group_final_wake_paused_for_test.notify_all();
+                while (pause_before_task_group_final_wake_for_test.load(
+                    std::memory_order_acquire)) {
+                    pause_before_task_group_final_wake_for_test.wait(
+                        true, std::memory_order_acquire);
+                }
+                task_group_final_wake_paused_for_test.store(
+                    false, std::memory_order_release);
+                task_group_final_wake_paused_for_test.notify_all();
+            }
+#endif
+            // Admission can legally reopen the count from zero before join
+            // closes the group. A registered waiter means public admission is
+            // already closed. Revalidate under that waiter-slot lock so an
+            // older zero transition cannot select a later join waiter.
+            auto wake = all_done_waiter_.take_if([this] {
+                return outstanding_children() == 0;
+            });
             if (wake) {
                 return {scheduler_, std::move(wake)};
             }

tests/unit/test_task.cpp

@@ -662,6 +662,17 @@ TEST_CASE("completion wake lease preserves slot ownership transitions",
         REQUIRE_FALSE(slot.take());
     }
 
+    SECTION("conditional selection leaves a registered waiter linked") {
+        completion_waiter_slot slot;
+        completion_waiter waiter(slot);
+        const auto handle = std::noop_coroutine();
+
+        REQUIRE(slot.register_waiter(waiter, handle, [] { return false; }));
+        REQUIRE_FALSE(slot.take_if([] { return false; }));
+        auto wake = slot.take_if([] { return true; });
+        REQUIRE(wake.claim() == handle);
+    }
+
     SECTION("an abandoned generation cannot claim a reused slot") {
         completion_waiter_slot slot;
         std::optional<completion_waiter> first(std::in_place, slot);

tests/unit/test_task_group.cpp

@@ -832,6 +832,80 @@ TEST_CASE("task_group completion skips a join waiter destroyed after selection",
     REQUIRE(drained);
 }
 
+TEST_CASE("task_group ignores a stale zero transition after new admission",
+          "[task_group][structured][lifetime][regression]") {
+    using namespace elio::coro::detail;
+
+    scheduler sched(1);
+    auto state = std::make_shared<task_group_completion_state>(sched);
+    state->register_child();
+    pause_before_task_group_final_wake_for_test.store(
+        true, std::memory_order_release);
+    task_group_final_wake_paused_for_test.store(
+        false, std::memory_order_release);
+    std::atomic<bool> old_selected{false};
+
+    std::thread old_child([&] {
+        auto completion = state->child_finished();
+        old_selected.store(static_cast<bool>(completion.wake),
+                           std::memory_order_release);
+        if (auto selected = completion.wake.claim()) {
+            selected.resume();
+        }
+    });
+    const bool old_final_paused = wait_for_flag(
+        task_group_final_wake_paused_for_test);
+    if (!old_final_paused) {
+        pause_before_task_group_final_wake_for_test.store(
+            false, std::memory_order_release);
+        pause_before_task_group_final_wake_for_test.notify_all();
+        old_child.join();
+        REQUIRE(old_final_paused);
+        return;
+    }
+
+    state->register_child();
+    std::atomic<bool> resumed{false};
+    auto waiter_task = [state, &resumed]() -> task<void> {
+        co_await task_group_completion_state::all_done_awaitable(state);
+        resumed.store(true, std::memory_order_release);
+    };
+    auto waiter = waiter_task();
+    auto handle = task_access::handle(waiter);
+    handle.resume();
+    const bool waiter_registered = !handle.done();
+
+    pause_before_task_group_final_wake_for_test.store(
+        false, std::memory_order_release);
+    pause_before_task_group_final_wake_for_test.notify_all();
+    old_child.join();
+    task_group_final_wake_paused_for_test.store(
+        false, std::memory_order_release);
+
+    const bool resumed_by_old = resumed.load(std::memory_order_acquire);
+    const auto outstanding_after_old = state->outstanding_children();
+    auto final_completion = state->child_finished();
+    const bool final_selected = static_cast<bool>(final_completion.wake);
+    auto selected = final_completion.wake.claim();
+    const bool selected_expected_handle =
+        selected && selected.address() == handle.address();
+    if (selected) {
+        selected.resume();
+    }
+    const bool resumed_by_final = resumed.load(std::memory_order_acquire);
+    const bool waiter_done = handle.done();
+
+    REQUIRE(waiter_registered);
+    REQUIRE_FALSE(old_selected.load(std::memory_order_acquire));
+    REQUIRE_FALSE(resumed_by_old);
+    REQUIRE(outstanding_after_old == 1);
+    REQUIRE(final_selected);
+    REQUIRE(selected_expected_handle);
+    REQUIRE(resumed_by_final);
+    REQUIRE(waiter_done);
+    REQUIRE(state->outstanding_children() == 0);
+}
+
 TEST_CASE("task_scope survives rejected handoff after external body wakeup",
           "[task_group][task_scope][structured][scheduler_domain][failure]") {
     scheduler sched(1);

wiki/API-Contracts.md

@@ -48,7 +48,7 @@ by a broad module heading without checking the feature page or header comment.
 | `coro::task<T>` | A task is a move-only, single-shot lazy owner. Moving transfers the unstarted coroutine frame and leaves the source empty. Awaiting binds virtual-stack ancestry to the actual awaiter. An Elio-to-Elio direct await also shares the actual awaiter's execution context, cancellation token, affinity, and active-I/O ownership for the logical vthread; creation-time ancestry is irrelevant. A foreign promise remains a boundary and may materialize a distinct context. Context materialization can allocate and propagate allocation failure from `co_await` before the child starts. Destroying a non-empty task destroys its frame unless ownership was transferred to the runtime. Runtime policy remains in the bound promise context, not in the task object. | Do not await an empty or already-awaited task. Do not treat moving a task as migration, cancellation authority, or running work. Use an independent `spawn`/`go` root when a child needs a distinct cancellation or affinity domain. Treat a foreign coroutine promise as a boundary unless adapter code deliberately bridges a token. Keep objects referenced by the coroutine alive across suspension points. |
 | Direct `co_await` on Elio awaitables | Awaitables resume through the scheduler or backend path documented by that awaitable. | Do not destroy the awaited object, scheduler, stream, buffer, or synchronization primitive while an await is still registered unless the API documents safe teardown. |
 | `join_handle<T>` | Awaiting a join handle returns the spawned task result or rethrows its exception. `is_ready()` is a non-blocking result-readiness check and can precede frame destruction. `is_destroyed()` observes frame release; after `wait_destroyed()` returns for a normally completed spawn, the root frame and its parameters have been destroyed, including any callable wrapper used by the callable overload. `request_cancel()` publishes a cooperative request through the spawned root's shared execution context and remains safe after frame destruction; it does not rewrite a result that already completed. The request propagates down direct lazy-task awaits between Elio tasks. | Await or intentionally discard join handles according to the task lifetime you need. Call `wait_destroyed()` only from a non-coroutine thread that may block. Do not use a moved-from handle. Pass `this_coro::cancel_token()` into cancellation-aware waits inside the task. Do not assume `request_cancel()` forcibly destroys the task, stops it promptly, crosses a foreign coroutine promise, or cancels a separate explicit token. Registered callback exceptions can propagate from `request_cancel()`. |
-| `coro::task_group` | Owns registration, cancellation, failure collection, and completion accounting for children submitted to one scheduler. The default failure policy records the first child failure, requests sibling cancellation, waits for every registered child frame to leave the group, and then rethrows that failure from the direct single-use `join()` awaitable. `join()` does not create or link a nested task. `collect_all` records failures without cancelling siblings and reports them together through `task_group_error`. A nonzero `max_concurrency` bounds executing child bodies while preserving every accepted child registration. `outstanding_children()` counts accepted child submissions, not an active `task_scope()` body. A group created on its scheduler inherits the current task's cancellation token. Its join continuation always resumes on a worker of the selected scheduler. An ordinary external thread, including the thread that called `scheduler::start()`, synchronously dispatches cancellation there; a worker in another scheduler posts cancellation asynchronously to avoid cross-scheduler deadlock. | Call `join()` exactly once while executing on a worker of the group's scheduler, and keep the group alive until it completes. Pass `this_coro::cancel_token()` into cancellation-aware child operations; cancellation is cooperative and cannot finish a child blocked in work that ignores its token. Do not spawn after joining starts. Keep child captures alive through join. Treat an explicitly selected scheduler as the group's only execution domain and keep it running until the group drains. Account for synchronous callback execution and possible blocking on ordinary external threads; do not assume cancellation has run when a request returns on another scheduler's worker. The destructor requests cancellation but does not synchronously join. |
+| `coro::task_group` | Owns registration, cancellation, failure collection, and completion accounting for children submitted to one scheduler. The default failure policy records the first child failure, requests sibling cancellation, waits for every registered child frame to leave the group, and then rethrows that failure from the direct single-use `join()` awaitable. `join()` does not create or link a nested task. `collect_all` records failures without cancelling siblings and reports them together through `task_group_error`. A nonzero `max_concurrency` bounds executing child bodies while preserving every accepted child registration. `outstanding_children()` counts accepted child submissions, not an active `task_scope()` body; before join closes admission, a newly accepted child may follow another child's transition to zero. Join-wake selection revalidates that no accepted child remains under the same slot lock used to register the waiter. A group created on its scheduler inherits the current task's cancellation token. Its join continuation always resumes on a worker of the selected scheduler. An ordinary external thread, including the thread that called `scheduler::start()`, synchronously dispatches cancellation there; a worker in another scheduler posts cancellation asynchronously to avoid cross-scheduler deadlock. | Call `join()` exactly once while executing on a worker of the group's scheduler, and keep the group alive until it completes. Pass `this_coro::cancel_token()` into cancellation-aware child operations; cancellation is cooperative and cannot finish a child blocked in work that ignores its token. Do not spawn after joining starts. Keep child captures alive through join. Treat an explicitly selected scheduler as the group's only execution domain and keep it running until the group drains. Account for synchronous callback execution and possible blocking on ordinary external threads; do not assume cancellation has run when a request returns on another scheduler's worker. The destructor requests cancellation but does not synchronously join. |
 | `coro::task_scope()` | Creates a task group for a callback-shaped lexical scope and runs the body under an isolated group cancellation context. Caller cancellation propagates into the scope, but group or fail-fast cancellation does not leave the caller's token cancelled after the scope joins. Caller user affinity is inherited and the scope's final user-affinity value flows back; active I/O pins remain owned by the scope's operation context. Normal return joins all children. A fail-fast child can therefore wake a token-aware suspended body before the scope joins and reports the child failure. If a body awaitable resumes elsewhere, scope cleanup hands execution back to a worker of the selected scheduler. The body callable and its captures remain alive until the children join. If the body throws, the scope requests child cancellation and joins every child. Fail-fast then rethrows the body failure; `collect_all` reports the body failure together with every child failure in `task_group_error`. | Use it when lexical cancel-and-join is required. The body must return an Elio task, must not call `join()` on its own group, and must not allow token-ignoring work to outlive captured resources. The explicit-scheduler overload must initially be awaited on a worker of that scheduler; it does not migrate a foreign worker or external thread into the scope. Automatic local objects in the body coroutine are destroyed when that coroutine returns, so children that can outlive the body must not retain references to those locals. Choose `collect_all` or a concurrency bound explicitly when fail-fast and unlimited execution are not suitable. Keep the selected scheduler running through cleanup and caller resumption. |
 | `coro::generator<T>` | `next()` delivers yielded values in producer order and returns completion through the documented optional result. It binds producer virtual-stack ancestry to the active consumer and restores the consumer context before yield and completion transfers, including after internal asynchronous suspension. | Use it as a single-consumer stream unless a future generator API documents sharing. Do not retain yielded references or views beyond their documented lifetime. |
 | `coro::cancel_source` | Calling `cancel()` publishes cooperative cancellation and synchronously dispatches callbacks outside the state lock. Same-dispatch reentrant teardown may remove a later selected callback. After dispatch, `cancel()` rethrows the first callback exception. Concurrent calls publish one request. | Keep callback code reentry-safe and account for execution on the requesting thread. Do not assume cancellation forcibly destroys operations that do not observe the token. Handle callback exceptions where callbacks may throw. |