From 206c3fbe66d7f6de61313bf1501cc636430b4489 Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 14:40:56 +0300
Subject: [PATCH 1/6] task/event_loop: fix TSAN race where event_task drifts to
 pool thread via cfuture

Root cause: task_final_awaiter did inline symmetric transfer to the event_task
continuation on whatever thread the inner task completed on (e.g. a pool thread
resuming a co_await cfuture). The event_task body then ran off the loop thread,
racing the loop thread's event_task.done() check in run_until_done().

Fix (Option A - loop-affine by default):

- task.h: add inline thread_local tl_loop_post_fn/tl_loop_ctx in rpp::detail.
  task_base::await_suspend captures them into the promise atomics (continuation,
  resume_via, resume_ctx) using a CAS handshake with task_final_awaiter so that
  both the "task finishes first" and "awaiter sets up first" races are safe.
  task_final_awaiter reads them with acquire semantics and posts the continuation
  back to the owner loop instead of doing symmetric transfer.

- event_loop.cpp: constructor registers loop_post_resume on the owner thread's
  TLS (covering eagerly-started event_tasks before the first process_event call).
  process_event saves/restores TLS around each resume so nested loops are correct.
  loop_post_resume is a static C-function trampoline that calls post_resume().

- tests/test_event_loop.cpp: add TestCase(tsan_task_await_resumes_on_loop_thread)
  that co_awaits a cfuture-completing task<int> from an event_task and asserts
  the continuation runs on the loop thread; passes TSAN clean 5/5 runs.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 src/rpp/event_loop.cpp    | 25 +++++++++-
 src/rpp/task.h            | 97 +++++++++++++++++++++++++++++++++------
 tests/test_event_loop.cpp | 42 +++++++++++++++++
 3 files changed, 148 insertions(+), 16 deletions(-)

diff --git a/src/rpp/event_loop.cpp b/src/rpp/event_loop.cpp
index 7e647b2..27f26d3 100644
--- a/src/rpp/event_loop.cpp
+++ b/src/rpp/event_loop.cpp
@@ -9,6 +9,8 @@
 
 namespace rpp
 {
+    static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept;
+
     event_loop::event_loop(rpp::uint64 main_thr_id,
                            rpp::thread_pool* background_task_pool,
                            rpp::AtomicTimeSource* warpable_clock) noexcept
@@ -16,6 +18,12 @@ namespace rpp
         , background_pool{background_task_pool ? *background_task_pool : rpp::thread_pool::global()}
         , time_source{warpable_clock}
     {
+        // Register this loop on the owner thread so that any rpp::task<T> constructed
+        // on the owner thread (including eagerly-started ones before the first
+        // process_event call) captures the correct post-back function.
+        // process_event saves/restores these, so nested calls stay correct.
+        detail::tl_loop_post_fn = &loop_post_resume;
+        detail::tl_loop_ctx     = this;
     }
 
     event_loop::~event_loop() noexcept
@@ -242,8 +250,20 @@ namespace rpp
         rpp::erase_if(fork_tasks, [](const event_task& t) { return t.done(); });
     }
 
+    static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept
+    {
+        static_cast<event_loop*>(ctx)->post_resume(h);
+    }
+
     void event_loop::process_event(resume_event& event) noexcept
     {
+        // Set thread-locals so any rpp::task co_awaited from within this
+        // coroutine posts its continuation back to this loop (Option A fix).
+        auto prev_fn  = detail::tl_loop_post_fn;
+        auto prev_ctx = detail::tl_loop_ctx;
+        detail::tl_loop_post_fn = &loop_post_resume;
+        detail::tl_loop_ctx     = this;
+
         if (event.handle)
         {
             try
@@ -265,7 +285,7 @@ namespace rpp
         else if (event.callback)
         {
             try
-            { 
+            {
                 event.callback();
             }
             catch (const std::exception& ex)
@@ -280,6 +300,9 @@ namespace rpp
                 else LogError("event_loop unhandled exception from delegate");
             }
         }
+
+        detail::tl_loop_post_fn = prev_fn;
+        detail::tl_loop_ctx     = prev_ctx;
     }
 
     void event_loop::invoke_loop_hook() noexcept
diff --git a/src/rpp/task.h b/src/rpp/task.h
index e21f110..302f792 100644
--- a/src/rpp/task.h
+++ b/src/rpp/task.h
@@ -13,6 +13,7 @@
 
 #if RPP_HAS_COROUTINES
 // NOTE: future_types.h already includes <coroutine> and exposes rpp::coro_handle / rpp::suspend_* / RPP_CORO_STD
+#include <atomic>
 #include <exception> // std::exception_ptr
 #include <type_traits> // std::conditional_t, std::is_void_v
 #include <utility> // std::exchange, std::move
@@ -66,17 +67,49 @@ namespace rpp
 
     namespace detail
     {
+        // Set by event_loop::process_event before each coroutine resume.
+        // task_base::await_suspend captures these so task_final_awaiter can
+        // post the continuation back to the loop instead of doing an inline
+        // symmetric transfer (which would cause the event_task to drift to
+        // whichever thread the inner task completed on).
+        using loop_post_fn = void (*)(void* ctx, rpp::coro_handle<>) noexcept;
+        inline thread_local loop_post_fn tl_loop_post_fn = nullptr;
+        inline thread_local void* tl_loop_ctx = nullptr;
+
+        // Sentinel: task_final_awaiter stores this when it runs before task_base::await_suspend,
+        // so that await_suspend can detect "already done" and handle the continuation itself.
+        inline void* task_done_sentinel() noexcept { return reinterpret_cast<void*>(uintptr_t{1}); }
+
         // Final-suspend awaiter shared by every task/deferred promise: hands control to the awaiting
         // coroutine via symmetric transfer (no thread hop, no spawn). Works on any promise that
-        // exposes a `.continuation` handle.
+        // exposes `.continuation`, `.resume_via`, `.resume_ctx` atomics.
         struct task_final_awaiter
         {
             bool await_ready() const noexcept { return false; }
             template <class Promise>
             rpp::coro_handle<> await_suspend(rpp::coro_handle<Promise> h) const noexcept
             {
-                rpp::coro_handle<> cont = h.promise().continuation;
-                return cont ? cont : RPP_CORO_STD::noop_coroutine();
+                auto& p = h.promise();
+                // Try to claim the continuation slot. If null → swap in sentinel (task finished first).
+                // release: pairs with the acquire in task_base::await_suspend's CAS/load.
+                void* expected = nullptr;
+                if (!p.continuation.compare_exchange_strong(expected, task_done_sentinel(),
+                        std::memory_order_release, std::memory_order_relaxed))
+                {
+                    // task_base::await_suspend already stored the continuation — resume it
+                    // acquire: sees the resume_via/resume_ctx stored by await_suspend
+                    void* cont_addr = expected; // CAS failure: expected = current value
+                    rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(cont_addr);
+                    if (auto fn = p.resume_via.load(std::memory_order_acquire))
+                    {
+                        void* ctx = p.resume_ctx.load(std::memory_order_relaxed);
+                        fn(ctx, cont); // post continuation to owner loop
+                        return RPP_CORO_STD::noop_coroutine();
+                    }
+                    return cont; // symmetric transfer (no loop context — standalone use)
+                }
+                // Sentinel stored — task_base::await_suspend will handle the continuation
+                return RPP_CORO_STD::noop_coroutine();
             }
             void await_resume() const noexcept {}
         };
@@ -105,7 +138,11 @@ namespace rpp
         struct task_promise
         {
             using value_type = T;
-            rpp::coro_handle<> continuation{}; // awaiter to resume when we finish
+            // Atomics: task_base::await_suspend writes these on the owner thread; task_final_awaiter
+            // reads them on the completing thread (which may be a pool thread for cfuture leaves).
+            std::atomic<void*> continuation{nullptr}; // coro_handle<>::address() of the awaiting coroutine
+            std::atomic<loop_post_fn> resume_via{nullptr}; // if set, post back via this instead of symmetric transfer
+            std::atomic<void*> resume_ctx{nullptr};         // context for resume_via
             std::variant<std::monostate, T, std::exception_ptr> result{};
 
             Owner get_return_object() noexcept { return Owner{rpp::coro_handle<task_promise>::from_promise(*this)}; }
@@ -119,7 +156,9 @@ namespace rpp
         struct task_promise<void, Lazy, Owner>
         {
             using value_type = void;
-            rpp::coro_handle<> continuation{};
+            std::atomic<void*> continuation{nullptr};
+            std::atomic<loop_post_fn> resume_via{nullptr};
+            std::atomic<void*> resume_ctx{nullptr};
             std::exception_ptr error{};
 
             Owner get_return_object() noexcept { return Owner{rpp::coro_handle<task_promise>::from_promise(*this)}; }
@@ -175,26 +214,54 @@ namespace rpp
             ///          lets drivers like event_loop::run_until_done check completion without awaiting.
             bool done() const noexcept { return !handle || handle.done(); }
 
-            bool await_ready() const noexcept { return !handle || handle.done(); }
+            // Returns true only if the task has already completed AND the continuation slot
+            // was claimed by this side (not by task_final_awaiter). In all other cases we
+            // go through await_suspend so the CAS handshake can coordinate safely.
+            bool await_ready() const noexcept { return false; }
 
-            // Register the awaiter, then hand off control:
-            //  - Eager: the body already started at construction -> just suspend (noop_coroutine).
-            //  - Deferred, not yet started: launch it via symmetric transfer (return its handle).
-            //  - Deferred, already started (e.g. start() then co_await): just suspend; resuming the
-            //    handle again would double-resume a coroutine already in flight. await_ready()
-            //    short-circuits when it is already done, so this only handles the in-flight case.
+            // Register the awaiter and hand off control.
+            // Uses a CAS on the continuation slot to coordinate with task_final_awaiter:
+            //   - CAS null→awaiting: task still in flight — it will post when done.
+            //   - CAS finds sentinel: task already finished — resume inline (or post to loop).
+            // Also handles deferred (Lazy) launch via symmetric transfer.
             rpp::coro_handle<> await_suspend(rpp::coro_handle<> awaiting) noexcept
             {
-                handle.promise().continuation = awaiting;
+                auto& p = handle.promise();
                 if constexpr (Lazy)
                 {
                     if (!started)
                     {
                         started = true;
-                        return handle;
+                        // For lazy tasks the body hasn't run yet: store continuation before launching
+                        p.resume_ctx.store(detail::tl_loop_ctx, std::memory_order_relaxed);
+                        p.resume_via.store(detail::tl_loop_post_fn, std::memory_order_relaxed);
+                        p.continuation.store(awaiting.address(), std::memory_order_release);
+                        return handle; // symmetric transfer launches the body
                     }
                 }
-                return RPP_CORO_STD::noop_coroutine();
+
+                // Store resume_via / resume_ctx first (relaxed), then CAS continuation (release).
+                // The release pairs with task_final_awaiter's CAS failure acquire.
+                p.resume_ctx.store(detail::tl_loop_ctx, std::memory_order_relaxed);
+                p.resume_via.store(detail::tl_loop_post_fn, std::memory_order_relaxed);
+
+                void* expected = nullptr;
+                if (p.continuation.compare_exchange_strong(expected, awaiting.address(),
+                        std::memory_order_release, std::memory_order_acquire))
+                {
+                    // CAS succeeded: continuation stored, task will call resume_via when done
+                    return RPP_CORO_STD::noop_coroutine();
+                }
+
+                // CAS failed: task_final_awaiter already ran and stored the sentinel —
+                // the task is done. Resume the awaiting coroutine inline or via the loop.
+                if (auto fn = p.resume_via.load(std::memory_order_relaxed))
+                {
+                    void* ctx = p.resume_ctx.load(std::memory_order_relaxed);
+                    fn(ctx, awaiting);
+                    return RPP_CORO_STD::noop_coroutine();
+                }
+                return awaiting; // symmetric transfer: no loop context, resume inline
             }
 
             typename Promise::value_type await_resume() { return detail::take_result<typename Promise::value_type>(handle); }
diff --git a/tests/test_event_loop.cpp b/tests/test_event_loop.cpp
index 9880746..d8fdc92 100644
--- a/tests/test_event_loop.cpp
+++ b/tests/test_event_loop.cpp
@@ -1313,6 +1313,48 @@ TestImpl(test_event_loop)
         AssertThat(off_thread, false); // ran on a different thread (one expected error log above)
     }
 
+    // Regression / TSAN repro: rpp::task<T> that completes on a pool thread
+    // (via a direct co_await cfuture, not via loop->run_async) must NOT cause
+    // the awaiting event_task to drift to the pool thread.
+    //
+    // Before the fix: task_final_awaiter did symmetric transfer to the event_task
+    // continuation on the completing pool thread; the event_task body then ran off
+    // the loop thread, racing the loop thread's event_task.done() check in
+    // run_until_done() — a genuine TSAN data race on the coroutine frame.
+    //
+    // After the fix: task_final_awaiter posts the continuation to the owner loop,
+    // so the event_task body always resumes on the loop thread.
+    TestCase(tsan_task_await_resumes_on_loop_thread)
+    {
+        std::atomic<rpp::uint64> continuation_tid{0};
+        const rpp::uint64 loop_tid = rpp::get_thread_id();
+
+        // A task whose body is resumed (and co_returns) on a pool thread:
+        // cfuture::await_suspend spawns a pool thread that calls cont.resume() inline,
+        // so the task body after co_await executes on that pool thread.
+        auto pool_completing_task = [&]() -> rpp::task<int>
+        {
+            // Sleep 1 ms so the future isn't ready before task_base::await_suspend
+            // writes continuation/resume_via, keeping the test race-free under TSAN.
+            rpp::cfuture<int> fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; });
+            int r = co_await fut; // cfuture resumes this task body on pool thread
+            co_return r;          // co_return fires on pool thread -> task_final_awaiter on pool thread
+        };
+
+        rpp::event_task et = [&]() -> rpp::event_task
+        {
+            int r = co_await pool_completing_task();
+            // Before fix: event_task body runs on pool thread -> races loop thread's done() check.
+            // After fix: always resumes on the loop thread.
+            continuation_tid = rpp::get_thread_id();
+            AssertThat(r, 42);
+        }();
+
+        loop->run_until_done(et);
+
+        AssertThat(continuation_tid.load(), loop_tid);
+    }
+
     // NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines)
 
 #endif // RPP_HAS_COROUTINES

From 198fcb9449116e8e261ccc77ed8ff2f69f22b2b3 Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 14:55:54 +0300
Subject: [PATCH 2/6] task: fix memory ordering in task_final_awaiter CAS
 failure path

The CAS failure in task_final_awaiter used memory_order_relaxed, so it did
not synchronize with the release CAS in task_base::await_suspend. On
weak-memory architectures (ARM) this means resume_via/resume_ctx stores
(written before the await_suspend release CAS) were not guaranteed visible
to task_final_awaiter after the relaxed failure, breaking the loop-post-back
mechanism.

Fix: use memory_order_acquire on the CAS failure so it pairs with
await_suspend's release and sees all prior relaxed stores. The subsequent
resume_via.load can then be relaxed (HB already established).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 src/rpp/task.h | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/src/rpp/task.h b/src/rpp/task.h
index 302f792..dde64cd 100644
--- a/src/rpp/task.h
+++ b/src/rpp/task.h
@@ -91,16 +91,19 @@ namespace rpp
             {
                 auto& p = h.promise();
                 // Try to claim the continuation slot. If null → swap in sentinel (task finished first).
-                // release: pairs with the acquire in task_base::await_suspend's CAS/load.
+                // release: ensures result writes are visible to the acquiring await_suspend.
+                // acquire on failure: synchronizes with await_suspend's release CAS so we see
+                // resume_via/resume_ctx written before that release.
                 void* expected = nullptr;
                 if (!p.continuation.compare_exchange_strong(expected, task_done_sentinel(),
-                        std::memory_order_release, std::memory_order_relaxed))
+                        std::memory_order_release, std::memory_order_acquire))
                 {
-                    // task_base::await_suspend already stored the continuation — resume it
-                    // acquire: sees the resume_via/resume_ctx stored by await_suspend
-                    void* cont_addr = expected; // CAS failure: expected = current value
+                    // await_suspend already stored the continuation — resume it.
+                    // The CAS failure acquire syncs with await_suspend's release CAS, so
+                    // resume_via/resume_ctx are visible here without a further acquire load.
+                    void* cont_addr = expected; // CAS failure: expected holds the current value
                     rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(cont_addr);
-                    if (auto fn = p.resume_via.load(std::memory_order_acquire))
+                    if (auto fn = p.resume_via.load(std::memory_order_relaxed))
                     {
                         void* ctx = p.resume_ctx.load(std::memory_order_relaxed);
                         fn(ctx, cont); // post continuation to owner loop

From 41bbd39d69989d4b3dadf3987677a1e21c09cca6 Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 16:19:31 +0300
Subject: [PATCH 3/6] task/event_loop: simplify CAS handshake after review

- Merge tl_loop_post_fn + tl_loop_ctx into a single loop_ctx struct
  (one TLS slot; cleaner save/restore in process_event)
- Hoist resume_via/ctx stores before the lazy branch in await_suspend
  (was duplicated in two places, now one store unconditionally)
- await_suspend CAS-failure path: use thread-locals directly instead of
  re-reading from atomics we just wrote on the same thread
- Trim WHAT comments per style guide

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 src/rpp/event_loop.cpp | 18 +++-------
 src/rpp/task.h         | 77 ++++++++++++------------------------------
 2 files changed, 26 insertions(+), 69 deletions(-)

diff --git a/src/rpp/event_loop.cpp b/src/rpp/event_loop.cpp
index 27f26d3..71ec0dd 100644
--- a/src/rpp/event_loop.cpp
+++ b/src/rpp/event_loop.cpp
@@ -18,12 +18,7 @@ namespace rpp
         , background_pool{background_task_pool ? *background_task_pool : rpp::thread_pool::global()}
         , time_source{warpable_clock}
     {
-        // Register this loop on the owner thread so that any rpp::task<T> constructed
-        // on the owner thread (including eagerly-started ones before the first
-        // process_event call) captures the correct post-back function.
-        // process_event saves/restores these, so nested calls stay correct.
-        detail::tl_loop_post_fn = &loop_post_resume;
-        detail::tl_loop_ctx     = this;
+        detail::tl_loop = { &loop_post_resume, this };
     }
 
     event_loop::~event_loop() noexcept
@@ -257,12 +252,8 @@ namespace rpp
 
     void event_loop::process_event(resume_event& event) noexcept
     {
-        // Set thread-locals so any rpp::task co_awaited from within this
-        // coroutine posts its continuation back to this loop (Option A fix).
-        auto prev_fn  = detail::tl_loop_post_fn;
-        auto prev_ctx = detail::tl_loop_ctx;
-        detail::tl_loop_post_fn = &loop_post_resume;
-        detail::tl_loop_ctx     = this;
+        auto prev = detail::tl_loop;
+        detail::tl_loop = { &loop_post_resume, this };
 
         if (event.handle)
         {
@@ -301,8 +292,7 @@ namespace rpp
             }
         }
 
-        detail::tl_loop_post_fn = prev_fn;
-        detail::tl_loop_ctx     = prev_ctx;
+        detail::tl_loop = prev;
     }
 
     void event_loop::invoke_loop_hook() noexcept
diff --git a/src/rpp/task.h b/src/rpp/task.h
index dde64cd..17c196c 100644
--- a/src/rpp/task.h
+++ b/src/rpp/task.h
@@ -67,22 +67,15 @@ namespace rpp
 
     namespace detail
     {
-        // Set by event_loop::process_event before each coroutine resume.
-        // task_base::await_suspend captures these so task_final_awaiter can
-        // post the continuation back to the loop instead of doing an inline
-        // symmetric transfer (which would cause the event_task to drift to
-        // whichever thread the inner task completed on).
+        // Set by event_loop on the owner thread; captured by task_base::await_suspend so
+        // task_final_awaiter posts the continuation back instead of doing an inline transfer.
         using loop_post_fn = void (*)(void* ctx, rpp::coro_handle<>) noexcept;
-        inline thread_local loop_post_fn tl_loop_post_fn = nullptr;
-        inline thread_local void* tl_loop_ctx = nullptr;
+        struct loop_ctx { loop_post_fn fn = nullptr; void* ctx = nullptr; };
+        inline thread_local loop_ctx tl_loop;
 
-        // Sentinel: task_final_awaiter stores this when it runs before task_base::await_suspend,
-        // so that await_suspend can detect "already done" and handle the continuation itself.
+        // Sentinel stored by task_final_awaiter to signal "task completed before await_suspend ran".
         inline void* task_done_sentinel() noexcept { return reinterpret_cast<void*>(uintptr_t{1}); }
 
-        // Final-suspend awaiter shared by every task/deferred promise: hands control to the awaiting
-        // coroutine via symmetric transfer (no thread hop, no spawn). Works on any promise that
-        // exposes `.continuation`, `.resume_via`, `.resume_ctx` atomics.
         struct task_final_awaiter
         {
             bool await_ready() const noexcept { return false; }
@@ -90,28 +83,20 @@ namespace rpp
             rpp::coro_handle<> await_suspend(rpp::coro_handle<Promise> h) const noexcept
             {
                 auto& p = h.promise();
-                // Try to claim the continuation slot. If null → swap in sentinel (task finished first).
-                // release: ensures result writes are visible to the acquiring await_suspend.
-                // acquire on failure: synchronizes with await_suspend's release CAS so we see
-                // resume_via/resume_ctx written before that release.
+                // release: result is visible to whoever acquires. acquire on failure: we see the
+                // resume_via/ctx stores done before await_suspend's release CAS.
                 void* expected = nullptr;
                 if (!p.continuation.compare_exchange_strong(expected, task_done_sentinel(),
                         std::memory_order_release, std::memory_order_acquire))
                 {
-                    // await_suspend already stored the continuation — resume it.
-                    // The CAS failure acquire syncs with await_suspend's release CAS, so
-                    // resume_via/resume_ctx are visible here without a further acquire load.
-                    void* cont_addr = expected; // CAS failure: expected holds the current value
-                    rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(cont_addr);
+                    rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(expected);
                     if (auto fn = p.resume_via.load(std::memory_order_relaxed))
                     {
-                        void* ctx = p.resume_ctx.load(std::memory_order_relaxed);
-                        fn(ctx, cont); // post continuation to owner loop
+                        fn(p.resume_ctx.load(std::memory_order_relaxed), cont);
                         return RPP_CORO_STD::noop_coroutine();
                     }
-                    return cont; // symmetric transfer (no loop context — standalone use)
+                    return cont;
                 }
-                // Sentinel stored — task_base::await_suspend will handle the continuation
                 return RPP_CORO_STD::noop_coroutine();
             }
             void await_resume() const noexcept {}
@@ -141,11 +126,9 @@ namespace rpp
         struct task_promise
         {
             using value_type = T;
-            // Atomics: task_base::await_suspend writes these on the owner thread; task_final_awaiter
-            // reads them on the completing thread (which may be a pool thread for cfuture leaves).
-            std::atomic<void*> continuation{nullptr}; // coro_handle<>::address() of the awaiting coroutine
-            std::atomic<loop_post_fn> resume_via{nullptr}; // if set, post back via this instead of symmetric transfer
-            std::atomic<void*> resume_ctx{nullptr};         // context for resume_via
+            std::atomic<void*> continuation{nullptr};
+            std::atomic<loop_post_fn> resume_via{nullptr};
+            std::atomic<void*> resume_ctx{nullptr};
             std::variant<std::monostate, T, std::exception_ptr> result{};
 
             Owner get_return_object() noexcept { return Owner{rpp::coro_handle<task_promise>::from_promise(*this)}; }
@@ -217,54 +200,38 @@ namespace rpp
             ///          lets drivers like event_loop::run_until_done check completion without awaiting.
             bool done() const noexcept { return !handle || handle.done(); }
 
-            // Returns true only if the task has already completed AND the continuation slot
-            // was claimed by this side (not by task_final_awaiter). In all other cases we
-            // go through await_suspend so the CAS handshake can coordinate safely.
+            // Always false: the CAS in await_suspend must run to coordinate with task_final_awaiter.
             bool await_ready() const noexcept { return false; }
 
-            // Register the awaiter and hand off control.
-            // Uses a CAS on the continuation slot to coordinate with task_final_awaiter:
-            //   - CAS null→awaiting: task still in flight — it will post when done.
-            //   - CAS finds sentinel: task already finished — resume inline (or post to loop).
-            // Also handles deferred (Lazy) launch via symmetric transfer.
             rpp::coro_handle<> await_suspend(rpp::coro_handle<> awaiting) noexcept
             {
                 auto& p = handle.promise();
+                // Capture loop context; release CAS below pairs with task_final_awaiter's acquire.
+                p.resume_via.store(detail::tl_loop.fn, std::memory_order_relaxed);
+                p.resume_ctx.store(detail::tl_loop.ctx, std::memory_order_relaxed);
+
                 if constexpr (Lazy)
                 {
                     if (!started)
                     {
                         started = true;
-                        // For lazy tasks the body hasn't run yet: store continuation before launching
-                        p.resume_ctx.store(detail::tl_loop_ctx, std::memory_order_relaxed);
-                        p.resume_via.store(detail::tl_loop_post_fn, std::memory_order_relaxed);
                         p.continuation.store(awaiting.address(), std::memory_order_release);
                         return handle; // symmetric transfer launches the body
                     }
                 }
 
-                // Store resume_via / resume_ctx first (relaxed), then CAS continuation (release).
-                // The release pairs with task_final_awaiter's CAS failure acquire.
-                p.resume_ctx.store(detail::tl_loop_ctx, std::memory_order_relaxed);
-                p.resume_via.store(detail::tl_loop_post_fn, std::memory_order_relaxed);
-
                 void* expected = nullptr;
                 if (p.continuation.compare_exchange_strong(expected, awaiting.address(),
                         std::memory_order_release, std::memory_order_acquire))
-                {
-                    // CAS succeeded: continuation stored, task will call resume_via when done
                     return RPP_CORO_STD::noop_coroutine();
-                }
 
-                // CAS failed: task_final_awaiter already ran and stored the sentinel —
-                // the task is done. Resume the awaiting coroutine inline or via the loop.
-                if (auto fn = p.resume_via.load(std::memory_order_relaxed))
+                // Task already done — resume using captured loop context.
+                if (detail::tl_loop.fn)
                 {
-                    void* ctx = p.resume_ctx.load(std::memory_order_relaxed);
-                    fn(ctx, awaiting);
+                    detail::tl_loop.fn(detail::tl_loop.ctx, awaiting);
                     return RPP_CORO_STD::noop_coroutine();
                 }
-                return awaiting; // symmetric transfer: no loop context, resume inline
+                return awaiting;
             }
 
             typename Promise::value_type await_resume() { return detail::take_result<typename Promise::value_type>(handle); }

From be3dc49b9b0fe1a3749a502d78e37e015e186354 Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 17:03:24 +0300
Subject: [PATCH 4/6] task/event_loop: demote resume-loop payload to a plain
 member
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adversarial review (5-lens propose + 3-skeptic verify) of the CAS handshake
confirmed resume_via/resume_ctx carried zero synchronization: they were two
relaxed atomics per promise specialization, written on the loop thread before
the release CAS on `continuation` and read on the completing thread only after
the acquire-on-failure CAS. That is the textbook "publish a plain payload via a
release flag" idiom, so they can be a single plain detail::loop_ctx member —
the `continuation` atomic provides all the happens-before. Removes 4 atomic
objects + 4 relaxed memory-order tokens, and makes the contract clearer (the
relaxed atomics implied an ordering they did not carry).

The four `continuation` memory orders and the done-sentinel are unchanged: the
demotion's safety rests entirely on the acquire-on-failure CAS staying acquire.

Also: move loop_post_resume above its first use, drop its forward declaration;
tighten the ordering WHY-comments to name both publication sites (the lazy
first-await release store is a second one the old wording omitted).

Verified on the worktree (not master) with gcc and clang TSAN:
test_event_loop 45/45 incl. tsan_task_await_resumes_on_loop_thread (30/30 under
test_until_failure), full suite green except the pre-existing basic_duration_coro
timing assertion, clang-tidy clean, only the 2 pre-existing showcase warnings.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/rpp/event_loop.cpp    | 10 ++++------
 src/rpp/task.h            | 18 ++++++++----------
 tests/test_event_loop.cpp |  2 +-
 3 files changed, 13 insertions(+), 17 deletions(-)

diff --git a/src/rpp/event_loop.cpp b/src/rpp/event_loop.cpp
index 71ec0dd..d73fb39 100644
--- a/src/rpp/event_loop.cpp
+++ b/src/rpp/event_loop.cpp
@@ -9,7 +9,10 @@
 
 namespace rpp
 {
-    static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept;
+    static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept
+    {
+        static_cast<event_loop*>(ctx)->post_resume(h);
+    }
 
     event_loop::event_loop(rpp::uint64 main_thr_id,
                            rpp::thread_pool* background_task_pool,
@@ -245,11 +248,6 @@ namespace rpp
         rpp::erase_if(fork_tasks, [](const event_task& t) { return t.done(); });
     }
 
-    static void loop_post_resume(void* ctx, rpp::coro_handle<> h) noexcept
-    {
-        static_cast<event_loop*>(ctx)->post_resume(h);
-    }
-
     void event_loop::process_event(resume_event& event) noexcept
     {
         auto prev = detail::tl_loop;
diff --git a/src/rpp/task.h b/src/rpp/task.h
index 17c196c..1fb5273 100644
--- a/src/rpp/task.h
+++ b/src/rpp/task.h
@@ -84,15 +84,15 @@ namespace rpp
             {
                 auto& p = h.promise();
                 // release: result is visible to whoever acquires. acquire on failure: we see the
-                // resume_via/ctx stores done before await_suspend's release CAS.
+                // captured loop context published before await_suspend's release store/CAS.
                 void* expected = nullptr;
                 if (!p.continuation.compare_exchange_strong(expected, task_done_sentinel(),
                         std::memory_order_release, std::memory_order_acquire))
                 {
                     rpp::coro_handle<> cont = rpp::coro_handle<>::from_address(expected);
-                    if (auto fn = p.resume_via.load(std::memory_order_relaxed))
+                    if (auto fn = p.resume_loop.fn)
                     {
-                        fn(p.resume_ctx.load(std::memory_order_relaxed), cont);
+                        fn(p.resume_loop.ctx, cont);
                         return RPP_CORO_STD::noop_coroutine();
                     }
                     return cont;
@@ -127,8 +127,7 @@ namespace rpp
         {
             using value_type = T;
             std::atomic<void*> continuation{nullptr};
-            std::atomic<loop_post_fn> resume_via{nullptr};
-            std::atomic<void*> resume_ctx{nullptr};
+            loop_ctx resume_loop{}; // plain payload, published via the continuation release/acquire handshake
             std::variant<std::monostate, T, std::exception_ptr> result{};
 
             Owner get_return_object() noexcept { return Owner{rpp::coro_handle<task_promise>::from_promise(*this)}; }
@@ -143,8 +142,7 @@ namespace rpp
         {
             using value_type = void;
             std::atomic<void*> continuation{nullptr};
-            std::atomic<loop_post_fn> resume_via{nullptr};
-            std::atomic<void*> resume_ctx{nullptr};
+            loop_ctx resume_loop{}; // plain payload, published via the continuation release/acquire handshake
             std::exception_ptr error{};
 
             Owner get_return_object() noexcept { return Owner{rpp::coro_handle<task_promise>::from_promise(*this)}; }
@@ -206,9 +204,9 @@ namespace rpp
             rpp::coro_handle<> await_suspend(rpp::coro_handle<> awaiting) noexcept
             {
                 auto& p = handle.promise();
-                // Capture loop context; release CAS below pairs with task_final_awaiter's acquire.
-                p.resume_via.store(detail::tl_loop.fn, std::memory_order_relaxed);
-                p.resume_ctx.store(detail::tl_loop.ctx, std::memory_order_relaxed);
+                // Capture loop context; the release store/CAS below publishes it, pairing with
+                // task_final_awaiter's acquire-on-failure.
+                p.resume_loop = detail::tl_loop; // plain store, sequenced-before the release store/CAS below
 
                 if constexpr (Lazy)
                 {
diff --git a/tests/test_event_loop.cpp b/tests/test_event_loop.cpp
index d8fdc92..c71f591 100644
--- a/tests/test_event_loop.cpp
+++ b/tests/test_event_loop.cpp
@@ -1335,7 +1335,7 @@ TestImpl(test_event_loop)
         auto pool_completing_task = [&]() -> rpp::task<int>
         {
             // Sleep 1 ms so the future isn't ready before task_base::await_suspend
-            // writes continuation/resume_via, keeping the test race-free under TSAN.
+            // writes continuation/resume_loop, keeping the test race-free under TSAN.
             rpp::cfuture<int> fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; });
             int r = co_await fut; // cfuture resumes this task body on pool thread
             co_return r;          // co_return fires on pool thread -> task_final_awaiter on pool thread

From 76282cf86df9b4defe7644678971d6eaad071548 Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 17:04:00 +0300
Subject: [PATCH 5/6] docs: refresh README line refs after task/event_loop
 changes

Mechanical line-number sync (update_doc_linerefs.py); no display-text or API
changes. Reconciles refs that drifted as this branch reworked task.h /
event_loop handshake (and some pre-existing master drift in sprint.h/tests.h).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md | 126 +++++++++++++++++++++++++++---------------------------
 1 file changed, 63 insertions(+), 63 deletions(-)

diff --git a/README.md b/README.md
index 30a21a9..7d91578 100644
--- a/README.md
+++ b/README.md
@@ -663,17 +663,17 @@ Fast string building and type-safe formatting. `string_buffer` is an always-null
 
 | Method | Description |
 |--------|-------------|
-| [`write(const T& v)`](src/rpp/sprint.h#L124) | Write a value (auto-converts most types) |
-| [`writeln(const Args&... args)`](src/rpp/sprint.h#L339) | Write values followed by newline |
-| [`writef(const char* format, ...)`](src/rpp/sprint.h#L122) | Printf-style formatted write |
-| [`write_hex(const void* data, int numBytes)`](src/rpp/sprint.h#L285) | Write data as hex string |
-| [`write_cont(const Container& c)`](src/rpp/sprint.h#L264) | Write container contents |
-| [`prettyprint(const T& value)`](src/rpp/sprint.h#L347) | Pretty-print a value |
-| [`clear()`](src/rpp/sprint.h#L113) | Clear the buffer |
-| [`reserve(int capacity)`](src/rpp/sprint.h#L114) | Reserve capacity |
-| [`resize(int size)`](src/rpp/sprint.h#L115) | Resize buffer |
-| [`append(const char* data, int len)`](src/rpp/sprint.h#L118) | Append raw data |
-| [`emplace_buffer(int n)`](src/rpp/sprint.h#L121) | Get writable buffer of N bytes |
+| [`write(const T& v)`](src/rpp/sprint.h#L125) | Write a value (auto-converts most types) |
+| [`writeln(const Args&... args)`](src/rpp/sprint.h#L340) | Write values followed by newline |
+| [`writef(const char* format, ...)`](src/rpp/sprint.h#L123) | Printf-style formatted write |
+| [`write_hex(const void* data, int numBytes)`](src/rpp/sprint.h#L286) | Write data as hex string |
+| [`write_cont(const Container& c)`](src/rpp/sprint.h#L265) | Write container contents |
+| [`prettyprint(const T& value)`](src/rpp/sprint.h#L348) | Pretty-print a value |
+| [`clear()`](src/rpp/sprint.h#L114) | Clear the buffer |
+| [`reserve(int capacity)`](src/rpp/sprint.h#L115) | Reserve capacity |
+| [`resize(int size)`](src/rpp/sprint.h#L116) | Resize buffer |
+| [`append(const char* data, int len)`](src/rpp/sprint.h#L119) | Append raw data |
+| [`emplace_buffer(int n)`](src/rpp/sprint.h#L122) | Get writable buffer of N bytes |
 
 ### Free Functions
 
@@ -684,9 +684,9 @@ Fast string building and type-safe formatting. `string_buffer` is an always-null
 | [`to_string(float)`](src/rpp/sprint.h#L51) | Locale-agnostic float to string |
 | [`to_string(double)`](src/rpp/sprint.h#L52) | Locale-agnostic double to string |
 | [`to_string(bool)`](src/rpp/sprint.h#L55) | Bool to `"true"` or `"false"` |
-| [`print(args...)`](src/rpp/sprint.h#L461) | Print to stdout |
-| [`println(args...)`](src/rpp/sprint.h#L481) | Print to stdout with newline |
-| [`to_hex_string(s, opt)`](src/rpp/sprint.h#L407) | Converts string bytes to hexadecimal representation |
+| [`print(args...)`](src/rpp/sprint.h#L462) | Print to stdout |
+| [`println(args...)`](src/rpp/sprint.h#L482) | Print to stdout with newline |
+| [`to_hex_string(s, opt)`](src/rpp/sprint.h#L408) | Converts string bytes to hexadecimal representation |
 
 ### Example: Basic String Building
 
@@ -1417,12 +1417,12 @@ Neither is a future: there is no `get()`/`wait()`/`.then()` — drive by `co_awa
 
 | Type | Description |
 |------|-------------|
-| [`task<T>`](src/rpp/task.h#L62) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop |
-| [`deferred<T>`](src/rpp/task.h#L63) | Lazy coroutine; body runs only when first awaited / `start()`ed. Same API as `task` |
-| [`done()`](src/rpp/task.h#L154) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) |
-| [`deferred<T>::start()`](src/rpp/task.h#L197) | Launch a not-yet-awaited deferred (used by `run_until_done`) |
+| [`task<T>`](src/rpp/task.h#L64) | Eager coroutine; body runs at construction. `co_await` resumes the caller on its loop |
+| [`deferred<T>`](src/rpp/task.h#L66) | Lazy coroutine; body runs only when first awaited / `start()`ed. Same API as `task` |
+| [`done()`](src/rpp/task.h#L199) | True once resolved; lets a driver poll completion without awaiting (no `wait()`) |
+| [`deferred<T>::start()`](src/rpp/task.h#L265) | Launch a not-yet-awaited deferred (used by `run_until_done`) |
 
-Drive a top-level task to completion with [`event_loop::run_until_done(task<T>&)`](src/rpp/event_loop.h#L319) or [`run_until_done(deferred<T>&)`](src/rpp/event_loop.h#L330).
+Drive a top-level task to completion with [`event_loop::run_until_done(task<T>&)`](src/rpp/event_loop.h#L325) or [`run_until_done(deferred<T>&)`](src/rpp/event_loop.h#L354).
 
 Example: [tests/test_task.cpp](tests/test_task.cpp)
 
@@ -1473,34 +1473,34 @@ Single-threaded event loop that serializes coroutine completions. Unlike `thread
 
 | Method | Description |
 |--------|-------------|
-| [`run_loop()`](src/rpp/event_loop.h#L264) | Run the loop until `stop()` is called, then drain remaining work |
-| [`run_once(Duration timeout)`](src/rpp/event_loop.h#L273) | Process at most one pending resume event; `Duration::zero()` for non-blocking poll |
-| [`run_until_idle()`](src/rpp/event_loop.h#L289) | Run until no background tasks and no pending resume events remain |
-| [`run_until_done(event_task& task)`](src/rpp/event_loop.h#L301) | Drive the loop until the given `event_task` completes, then rethrow on failure |
-| [`run_until_done(task<T>& task)`](src/rpp/event_loop.h#L319) | Pump the loop until the eager `rpp::task<T>` completes; returns its value (or rethrows) |
-| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L349) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
-| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L367) | Pump until that future is ready, then return its value; throws on timeout |
-| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L380) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
-| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L685) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
-| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L409) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
-| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L884) | Event-driven join: suspend until all forks complete or timeout expires |
-| [`num_forks()`](src/rpp/event_loop.h#L453) | Number of active forked coroutines |
-| [`drain_forks()`](src/rpp/event_loop.h#L461) | Check completed forks for exceptions and clear them |
-| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L716) | Wait for semaphore signal, resume on loop thread |
-| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L730) | Pop from queue, resume on loop thread |
-| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L744) | Pop from queue returning `optional<T>`, resume on loop thread |
-| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L510) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
-| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L502) | Post a raw coroutine handle resume to the loop thread |
-| [`resume_on_loop()`](src/rpp/event_loop.h#L899) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
-| [`delay(Duration duration)`](src/rpp/event_loop.h#L793) | Sleep on a background thread, resume on the loop thread |
-| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L797) | Sleep until a time point, resume on the loop thread |
-| [`stop()`](src/rpp/event_loop.h#L234) | Signal the loop to stop and finalize pending tasks |
-| [`wait_on_all(Duration timeout)`](src/rpp/event_loop.h#L241) | Block until all pending work drains, with timeout |
-| [`set_except_handler(handler)`](src/rpp/event_loop.h#L247) | Set custom exception handler for unhandled background errors |
-| [`has_pending_work()`](src/rpp/event_loop.h#L226) | True if any background tasks or resume events are pending |
-| [`background_tasks()`](src/rpp/event_loop.h#L217) | Number of tasks currently suspended in background work |
-| [`pending_completions()`](src/rpp/event_loop.h#L223) | Number of pending resume events queued for the loop thread |
-| [`main_thread_id()`](src/rpp/event_loop.h#L229) | Thread ID of the loop's owner thread |
+| [`run_loop()`](src/rpp/event_loop.h#L287) | Run the loop until `stop()` is called, then drain remaining work |
+| [`run_once(Duration timeout)`](src/rpp/event_loop.h#L296) | Process at most one pending resume event; `Duration::zero()` for non-blocking poll |
+| [`run_until_idle()`](src/rpp/event_loop.h#L313) | Run until no background tasks and no pending resume events remain |
+| [`run_until_done(event_task& task)`](src/rpp/event_loop.h#L325) | Drive the loop until the given `event_task` completes, then rethrow on failure |
+| [`run_until_done(task<T>& task)`](src/rpp/event_loop.h#L343) | Pump the loop until the eager `rpp::task<T>` completes; returns its value (or rethrows) |
+| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L373) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
+| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L391) | Pump until that future is ready, then return its value; throws on timeout |
+| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L404) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
+| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L709) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
+| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L433) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
+| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L908) | Event-driven join: suspend until all forks complete or timeout expires |
+| [`num_forks()`](src/rpp/event_loop.h#L477) | Number of active forked coroutines |
+| [`drain_forks()`](src/rpp/event_loop.h#L485) | Check completed forks for exceptions and clear them |
+| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L740) | Wait for semaphore signal, resume on loop thread |
+| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L754) | Pop from queue, resume on loop thread |
+| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L768) | Pop from queue returning `optional<T>`, resume on loop thread |
+| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L534) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
+| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L526) | Post a raw coroutine handle resume to the loop thread |
+| [`resume_on_loop()`](src/rpp/event_loop.h#L923) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
+| [`delay(Duration duration)`](src/rpp/event_loop.h#L817) | Sleep on a background thread, resume on the loop thread |
+| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L821) | Sleep until a time point, resume on the loop thread |
+| [`stop()`](src/rpp/event_loop.h#L239) | Signal the loop to stop and finalize pending tasks |
+| [`wait_on_all(Duration timeout)`](src/rpp/event_loop.h#L246) | Block until all pending work drains, with timeout |
+| [`set_except_handler(handler)`](src/rpp/event_loop.h#L252) | Set custom exception handler for unhandled background errors |
+| [`has_pending_work()`](src/rpp/event_loop.h#L231) | True if any background tasks or resume events are pending |
+| [`background_tasks()`](src/rpp/event_loop.h#L222) | Number of tasks currently suspended in background work |
+| [`pending_completions()`](src/rpp/event_loop.h#L228) | Number of pending resume events queued for the loop thread |
+| [`main_thread_id()`](src/rpp/event_loop.h#L234) | Thread ID of the loop's owner thread |
 
 ### event_task Methods
 
@@ -4343,32 +4343,32 @@ Minimal unit testing framework with test discovery, assertions, and verbose outp
 
 | Item | Description |
 |------|-------------|
-| [`test`](src/rpp/tests.h#L39) | Base test class with lifecycle hooks |
-| [`test_info`](src/rpp/tests.h#L40) | Test registration metadata |
-| [`TestVerbosity`](src/rpp/tests.h#L67) | `None`, `Summary`, `TestLabels`, `AllMessages` |
+| [`test`](src/rpp/tests.h#L29) | Base test class with lifecycle hooks |
+| [`test_info`](src/rpp/tests.h#L35) | Test registration metadata |
+| [`TestVerbosity`](src/rpp/tests.h#L47) | `None`, `Summary`, `TestLabels`, `AllMessages` |
 
 ### Key Macros
 
 | Macro | Description |
 |-------|-------------|
-| [`TestImpl(ClassName)`](src/rpp/tests.h#L713) | Register a test class |
-| [`TestInit(...)`](src/rpp/tests.h#L728) | Test initialization method |
-| [`TestCase(name)`](src/rpp/tests.h#L745) | Define a test case |
-| [`AssertThat(expr, expected)`](src/rpp/tests.h#L585) | Assert equality |
-| [`AssertEqual(a, b)`](src/rpp/tests.h#L601) | Assert exact equality |
-| [`AssertNotEqual(a, b)`](src/rpp/tests.h#L637) | Assert inequality |
-| [`AssertTrue(expr)`](src/rpp/tests.h#L706) | Assert expression is true |
-| [`AssertFalse(expr)`](src/rpp/tests.h#L576) | Assert expression is false |
-| [`AssertThrows(expr)`](src/rpp/tests.h#L610) | Assert expression throws |
+| [`TestImpl(ClassName)`](src/rpp/tests.h#L693) | Register a test class |
+| [`TestInit(...)`](src/rpp/tests.h#L708) | Test initialization method |
+| [`TestCase(name)`](src/rpp/tests.h#L725) | Define a test case |
+| [`AssertThat(expr, expected)`](src/rpp/tests.h#L565) | Assert equality |
+| [`AssertEqual(a, b)`](src/rpp/tests.h#L581) | Assert exact equality |
+| [`AssertNotEqual(a, b)`](src/rpp/tests.h#L617) | Assert inequality |
+| [`AssertTrue(expr)`](src/rpp/tests.h#L686) | Assert expression is true |
+| [`AssertFalse(expr)`](src/rpp/tests.h#L556) | Assert expression is false |
+| [`AssertThrows(expr)`](src/rpp/tests.h#L590) | Assert expression throws |
 
 ### Running Tests
 
 | Method | Description |
 |--------|-------------|
-| [`test::run_tests(patterns)`](src/rpp/tests.h#L283) | Run tests matching patterns |
-| [`test::run_tests(argc, argv)`](src/rpp/tests.h#L294) | Run tests from command line args |
-| [`test::run_tests()`](src/rpp/tests.h#L283) | Run all registered tests |
-| [`register_test(name, factory, autorun)`](src/rpp/tests.h#L65) | Registers a unit test with given name, factory and autorun flag |
+| [`test::run_tests(patterns)`](src/rpp/tests.h#L279) | Run tests matching patterns |
+| [`test::run_tests(argc, argv)`](src/rpp/tests.h#L274) | Run tests from command line args |
+| [`test::run_tests()`](src/rpp/tests.h#L279) | Run all registered tests |
+| [`register_test(name, factory, autorun)`](src/rpp/tests.h#L45) | Registers a unit test with given name, factory and autorun flag |
 
 ### Example: Defining a Test Class with TestCase
 

From 0de5f375e873049749684fe99bc5d20edc0fba6f Mon Sep 17 00:00:00 2001
From: Mark Cowan <mark@battlesnake.co.uk>
Date: Sat, 20 Jun 2026 19:39:16 +0300
Subject: [PATCH 6/6] event_loop: fix run_until_done(task/deferred) completing
 off the loop thread
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Same root cause as this PR's main fix, in a second spot. run_until_done() drove a
top-level task by polling handle.done() in a loop; if the task body finished on a
pool thread (e.g. it awaited a cfuture directly), that poll raced the pool thread
and could read the result before it was ready.

Now run_until_done() awaits the task from a small internal event_task, so
completion is posted back to the loop and the result is read on the loop thread —
reusing the path the event_task driver already uses. The old polling loop is gone.

Tests cover the task, deferred, void and exception cases; all clean under TSAN on
gcc and clang. (README line refs refreshed mechanically.)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md                 | 32 +++++++++---------
 src/rpp/event_loop.h      | 53 +++++++++++++++++++++++-------
 tests/test_event_loop.cpp | 69 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 126 insertions(+), 28 deletions(-)

diff --git a/README.md b/README.md
index 7d91578..09dd0ab 100644
--- a/README.md
+++ b/README.md
@@ -1478,22 +1478,22 @@ Single-threaded event loop that serializes coroutine completions. Unlike `thread
 | [`run_until_idle()`](src/rpp/event_loop.h#L313) | Run until no background tasks and no pending resume events remain |
 | [`run_until_done(event_task& task)`](src/rpp/event_loop.h#L325) | Drive the loop until the given `event_task` completes, then rethrow on failure |
 | [`run_until_done(task<T>& task)`](src/rpp/event_loop.h#L343) | Pump the loop until the eager `rpp::task<T>` completes; returns its value (or rethrows) |
-| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L373) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
-| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L391) | Pump until that future is ready, then return its value; throws on timeout |
-| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L404) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
-| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L709) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
-| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L433) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
-| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L908) | Event-driven join: suspend until all forks complete or timeout expires |
-| [`num_forks()`](src/rpp/event_loop.h#L477) | Number of active forked coroutines |
-| [`drain_forks()`](src/rpp/event_loop.h#L485) | Check completed forks for exceptions and clear them |
-| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L740) | Wait for semaphore signal, resume on loop thread |
-| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L754) | Pop from queue, resume on loop thread |
-| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L768) | Pop from queue returning `optional<T>`, resume on loop thread |
-| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L534) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
-| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L526) | Post a raw coroutine handle resume to the loop thread |
-| [`resume_on_loop()`](src/rpp/event_loop.h#L923) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
-| [`delay(Duration duration)`](src/rpp/event_loop.h#L817) | Sleep on a background thread, resume on the loop thread |
-| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L821) | Sleep until a time point, resume on the loop thread |
+| [`pump_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L372) | Pump on the owner thread until that one future is ready; `bool`, never blocks past timeout |
+| [`run_until_ready(cfuture<T>&, timeout)`](src/rpp/event_loop.h#L390) | Pump until that future is ready, then return its value; throws on timeout |
+| [`ensure_on_owner_thread(source_location)`](src/rpp/event_loop.h#L403) | Debug check: true if on the loop's owner thread, else logs an error at the call site |
+| [`run_async(Func&& fut_or_cb)`](src/rpp/event_loop.h#L708) | Dispatch future or lambda to thread pool, resume coroutine on the loop thread |
+| [`fork(Func&& coro_factory)`](src/rpp/event_loop.h#L432) | Fork a concurrent coroutine path (fire-and-forget, tracked internally) |
+| [`join_forks(Duration timeout)`](src/rpp/event_loop.h#L907) | Event-driven join: suspend until all forks complete or timeout expires |
+| [`num_forks()`](src/rpp/event_loop.h#L476) | Number of active forked coroutines |
+| [`drain_forks()`](src/rpp/event_loop.h#L484) | Check completed forks for exceptions and clear them |
+| [`await(semaphore&, Duration)`](src/rpp/event_loop.h#L739) | Wait for semaphore signal, resume on loop thread |
+| [`await(concurrent_queue<T>&, T&, Duration)`](src/rpp/event_loop.h#L753) | Pop from queue, resume on loop thread |
+| [`await_pop(concurrent_queue<T>&, Duration)`](src/rpp/event_loop.h#L767) | Pop from queue returning `optional<T>`, resume on loop thread |
+| [`post(delegate<void()> callback)`](src/rpp/event_loop.h#L533) | Post a callback to execute on the loop thread (like `run_on_main_thread`) |
+| [`post_resume(coro_handle<> handle)`](src/rpp/event_loop.h#L525) | Post a raw coroutine handle resume to the loop thread |
+| [`resume_on_loop()`](src/rpp/event_loop.h#L922) | `co_await` to unconditionally reschedule the current coroutine onto the loop thread |
+| [`delay(Duration duration)`](src/rpp/event_loop.h#L816) | Sleep on a background thread, resume on the loop thread |
+| [`delay_until(TimePoint until)`](src/rpp/event_loop.h#L820) | Sleep until a time point, resume on the loop thread |
 | [`stop()`](src/rpp/event_loop.h#L239) | Signal the loop to stop and finalize pending tasks |
 | [`wait_on_all(Duration timeout)`](src/rpp/event_loop.h#L246) | Block until all pending work drains, with timeout |
 | [`set_except_handler(handler)`](src/rpp/event_loop.h#L252) | Set custom exception handler for unhandled background errors |
diff --git a/src/rpp/event_loop.h b/src/rpp/event_loop.h
index be77a90..783c4ec 100644
--- a/src/rpp/event_loop.h
+++ b/src/rpp/event_loop.h
@@ -342,19 +342,18 @@ namespace rpp
         template <typename T>
         T run_until_done(rpp::task<T>& task)
         {
-            return drive_to_done(task); // eager: already in flight, just pump
+            return drive_top_level<T>(task); // route through an event_task so completion lands on the loop
         }
 
         /**
          * @brief Drives the loop until the given (lazy) `rpp::deferred<T>` completes, then returns
-         *        its value (or rethrows). Unlike `task`, a deferred has not started yet — this
-         *        starts it, then pumps the loop to completion. @warning Owner-thread only.
+         *        its value (or rethrows). Unlike `task`, a deferred has not started yet — awaiting
+         *        it launches the body, then the loop is pumped to completion. @warning Owner-thread only.
          */
         template <typename T>
         T run_until_done(rpp::deferred<T>& task)
         {
-            task.start(); // lazy: launch the body before pumping
-            return drive_to_done(task);
+            return drive_top_level<T>(task); // co_await drives the lazy launch; completion lands on the loop
         }
 
         /**
@@ -924,15 +923,45 @@ namespace rpp
 
     private:
 
-        // Shared pump used by both run_until_done(task)/(deferred): drive the loop until the
-        // already-started task completes, drain the trailing resumes, then return its result.
+        // Internal event_task wrappers that co_await a top-level task<T>/deferred<T> so its
+        // task_final_awaiter posts the continuation back to this loop (task.h): the result is then
+        // produced — and the user task's frame destroyed — on the loop thread, even when the task
+        // body completed on a pool thread (e.g. it co_awaited a cfuture directly). Captures are
+        // reference PARAMETERS, never a capturing lambda, so the coroutine frame can't dangle.
+        template <class Awaitable, class T>
+        static event_task await_into(Awaitable& task, std::optional<T>& out, std::exception_ptr& err)
+        {
+            try { out.emplace(co_await task); }
+            catch (...) { err = std::current_exception(); }
+        }
         template <class Awaitable>
-        auto drive_to_done(Awaitable& task) -> decltype(task.await_resume())
+        static event_task await_into_void(Awaitable& task, std::exception_ptr& err)
+        {
+            try { co_await task; }
+            catch (...) { err = std::current_exception(); }
+        }
+
+        // Drive a top-level task<T>/deferred<T> to completion through the (already loop-affine)
+        // run_until_done(event_task&) driver, then hand back its value or rethrow. Replaces the old
+        // cross-thread done() poll, which raced a pool thread completing the task frame off-loop.
+        template <typename T, class Awaitable>
+        T drive_top_level(Awaitable& task)
         {
-            while (!task.done())
-                run_once(rpp::millis(5));
-            run_all_ready(); // drain callbacks posted during the final resume (e.g. a trailing post())
-            return task.await_resume(); // done: returns the value or rethrows (non-blocking)
+            std::exception_ptr err;
+            if constexpr (std::is_void_v<T>)
+            {
+                event_task w = await_into_void(task, err);
+                run_until_done(w);
+                if (err) std::rethrow_exception(err);
+            }
+            else
+            {
+                std::optional<T> out;
+                event_task w = await_into(task, out, err);
+                run_until_done(w);
+                if (err) std::rethrow_exception(err);
+                return std::move(*out);
+            }
         }
 
         void start_in_background(task_delegate<void()>&& generic_task) noexcept
diff --git a/tests/test_event_loop.cpp b/tests/test_event_loop.cpp
index c71f591..23d71f7 100644
--- a/tests/test_event_loop.cpp
+++ b/tests/test_event_loop.cpp
@@ -1355,6 +1355,75 @@ TestImpl(test_event_loop)
         AssertThat(continuation_tid.load(), loop_tid);
     }
 
+    // Follow-up: run_until_done(task<T>&)/(deferred<T>&) must also drive a TOP-LEVEL task to
+    // completion on the loop thread when the task body completes on a pool thread (it co_awaited a
+    // cfuture directly). Before the fix the driver polled handle.done() cross-thread (TSAN frame
+    // race); now it routes completion through an internal event_task so the result is read on loop.
+    TestCase(tsan_run_until_done_task_resumes_on_loop_thread)
+    {
+        const rpp::uint64 loop_tid = rpp::get_thread_id();
+        std::atomic<rpp::uint64> body_tid{0};
+
+        auto pool_completing_task = [&]() -> rpp::task<int>
+        {
+            rpp::cfuture<int> fut = rpp::async_task([] { rpp::sleep_ms(1); return 42; });
+            int r = co_await fut;            // resumes this task body on a pool thread
+            body_tid = rpp::get_thread_id();
+            co_return r;                     // task_final_awaiter runs on the pool thread
+        };
+
+        rpp::task<int> t = pool_completing_task(); // eager: already in flight
+        int got = loop->run_until_done(t);         // drives + reads the result on the loop thread
+
+        AssertThat(got, 42);
+        AssertNotEqual(body_tid.load(), loop_tid); // sanity: the body really completed off-loop
+    }
+
+    // deferred<T> variant: the wrapper's co_await drives the lazy launch (no explicit start()).
+    TestCase(tsan_run_until_done_deferred_resumes_on_loop_thread)
+    {
+        auto make = [&]() -> rpp::deferred<int>
+        {
+            rpp::cfuture<int> fut = rpp::async_task([] { rpp::sleep_ms(1); return 7; });
+            int r = co_await fut;
+            co_return r;
+        };
+
+        rpp::deferred<int> d = make(); // lazy: not started until awaited
+        int got = loop->run_until_done(d);
+        AssertThat(got, 7);
+    }
+
+    // void-result variant: exercises the if-constexpr void branch of the wrapper driver.
+    TestCase(tsan_run_until_done_task_void)
+    {
+        std::atomic<bool> ran{false};
+        auto make = [&]() -> rpp::task<void>
+        {
+            co_await rpp::async_task([] { rpp::sleep_ms(1); });
+            ran = true; // completes on a pool thread
+            co_return;
+        };
+
+        rpp::task<void> t = make();
+        loop->run_until_done(t);
+        AssertThat(ran.load(), true);
+    }
+
+    // An exception thrown after an off-loop completion must propagate out of run_until_done
+    // (covers the wrapper's exception_ptr round-trip).
+    TestCase(run_until_done_task_propagates_exception)
+    {
+        auto make = [&]() -> rpp::task<int>
+        {
+            co_await rpp::async_task([] { rpp::sleep_ms(1); });
+            throw std::runtime_error("boom"); // thrown on a pool thread
+        };
+
+        rpp::task<int> t = make();
+        AssertThrows(loop->run_until_done(t), std::runtime_error);
+    }
+
     // NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines)
 
 #endif // RPP_HAS_COROUTINES