feat(runtime): add worker lifetime leases for external owners

CHANGELOG.md

@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Move-only worker lifetime leases for external thread-affine owners. Lease
+  acquisition retains exact-owner continuations and worker drain accounting;
+  release safely wakes retirement, including from a foreign thread. Retiring
+  workers service requeued owner tasks in bounded turns. Foreign wakeups,
+  isolated-scope returns, generator consumers, and blocking/timer/group
+  completion paths honor the retained owner (#1301).
+
 ## [0.6.0] - 2026-10-12
 
 This is the first formal 0.6 release. It includes intentional API and

README.md

@@ -319,7 +319,7 @@ their corresponding feature targets.
 - **Coroutine/runtime**: `hello_world.cpp`, `chained_coroutines.cpp`,
   `exception_handling.cpp`, `join_destroyed_async.cpp`, `join_wait.cpp`, `parallel_tasks.cpp`,
   `dynamic_threads.cpp`,
-  `autoscaler_example.cpp`, `thread_affinity.cpp`, `signal_handling.cpp`,
+  `autoscaler_example.cpp`, `thread_affinity.cpp`, `worker_lease.cpp`, `signal_handling.cpp`,
   `debug_test.cpp`
 - **Benchmarks**: `benchmark.cpp`, `quick_benchmark.cpp`, `microbench.cpp`,
   `scheduler_service_benchmark.cpp`, `io_benchmark.cpp`,
@@ -428,6 +428,11 @@ size_t executed = sched.total_tasks_executed();
 sched.shutdown();
 ```
 
+External thread-affine owners can acquire a
+[worker lifetime lease](wiki/Worker-Leases.md) from an Elio coroutine to retain
+their worker through normal pool shrink and drain. Affinity alone does not
+provide that lifetime guarantee.
+
 Scheduler execution metrics are published from worker-local counters. Optional
 autoscaler `on_block` detection samples those counters at `tick_interval`, so
 ordinary coroutine resumes do not read the clock. Block notifications are

examples/CMakeLists.txt

@@ -73,6 +73,16 @@ add_executable(thread_affinity thread_affinity.cpp)
 target_link_libraries(thread_affinity PRIVATE elio)
 target_link_options(thread_affinity PRIVATE ${STATIC_LINK_FLAGS})
 
+add_executable(worker_lease worker_lease.cpp)
+target_link_libraries(worker_lease PRIVATE elio)
+target_link_options(worker_lease PRIVATE ${STATIC_LINK_FLAGS})
+if(ELIO_BUILD_TESTS)
+    add_test(NAME worker_lease_example COMMAND worker_lease)
+    set_tests_properties(worker_lease_example PROPERTIES TIMEOUT 30
+        PASS_REGULAR_EXPRESSION "Owner cleanup on worker 1")
+    add_dependencies(test_all worker_lease)
+endif()
+
 add_executable(benchmark benchmark.cpp)
 target_link_libraries(benchmark PRIVATE elio)
 target_link_options(benchmark PRIVATE ${STATIC_LINK_FLAGS})

examples/worker_lease.cpp

@@ -0,0 +1,41 @@
+#include <elio/runtime/worker_lease.hpp>
+#include <elio/sync/event.hpp>
+#include <future>
+#include <iostream>
+#include <stdexcept>
+
+int main() {
+    elio::runtime::scheduler sched(2);
+    elio::sync::event retire_owner;
+    std::promise<void> ready;
+    auto ready_result = ready.get_future();
+    std::exception_ptr failure;
+    sched.start();
+    sched.go_to(1, [&]() -> elio::coro::task<void> {
+        bool published = false;
+        try {
+            auto owner = co_await elio::runtime::lease_current_worker();
+            const auto original_thread = std::this_thread::get_id();
+            ready.set_value();
+            published = true;
+            // Represents an external owner whose lifetime spans a non-I/O wait.
+            co_await retire_owner.wait();
+            if (std::this_thread::get_id() != original_thread) {
+                throw std::runtime_error("owner migrated during retirement");
+            }
+            std::cout << "Owner cleanup on worker " << owner.worker_id() << '\n';
+            // Destroy external thread-affine resources before releasing.
+            owner.release();
+        } catch (...) {
+            if (!published) ready.set_exception(std::current_exception());
+            else failure = std::current_exception();
+        }
+    });
+    // Blocking waits are on the controller thread, never a scheduler worker.
+    ready_result.get();
+    sched.set_thread_count(1);
+    retire_owner.set();
+    const bool drained = sched.shutdown();
+    if (failure) std::rethrow_exception(failure);
+    return drained ? 0 : 1;
+}

include/elio/coro/generator.hpp

@@ -27,7 +27,8 @@ namespace detail {
 /// Minimal coroutine type for generator::for_each implementation.
 /// Acts as both the coroutine return type and an awaitable.
 struct for_each_task {
-    struct promise_type {
+    struct promise_type : promise_base {
+        promise_type() : promise_base(defer_nested_task_context) {}
         std::exception_ptr exception_;
         std::coroutine_handle<> consumer_;
 
@@ -44,10 +45,8 @@ struct for_each_task {
                 [[nodiscard]] std::coroutine_handle<> await_suspend(
                     std::coroutine_handle<promise_type> h) noexcept {
                     auto consumer = h.promise().consumer_;
-                    if (consumer) {
-                        return consumer;
-                    }
-                    return std::noop_coroutine();
+                    h.promise().leave_frame_context();
+                    return transfer_to_continuation(consumer, h.promise().parent());
                 }
                 void await_resume() const noexcept {}
             };
@@ -64,7 +63,9 @@ struct for_each_task {
     using handle_type = std::coroutine_handle<promise_type>;
     handle_type handle_;
 
-    for_each_task(handle_type h) noexcept : handle_(h) {}
+    for_each_task(handle_type h) noexcept : handle_(h) {
+        handle_.promise().leave_creation_context();
+    }
 
     for_each_task(for_each_task&& other) noexcept
         : handle_(std::exchange(other.handle_, nullptr)) {}
@@ -82,9 +83,20 @@ struct for_each_task {
         return !handle_ || handle_.done();
     }
 
+    template<typename ConsumerPromise>
     [[nodiscard]] std::coroutine_handle<> await_suspend(
-        std::coroutine_handle<> consumer) noexcept {
-        handle_.promise().consumer_ = consumer;
+        std::coroutine_handle<ConsumerPromise> consumer) {
+        auto& promise = handle_.promise();
+        promise_base* parent = nullptr;
+        if constexpr (std::is_convertible_v<ConsumerPromise*, promise_base*>) {
+            parent = std::addressof(consumer.promise());
+            parent->ensure_independent_execution_context();
+            promise.bind_direct_await_context(parent->execution_context());
+        } else {
+            promise.ensure_independent_execution_context();
+        }
+        promise.enter_frame_context(parent);
+        promise.consumer_ = consumer;
         return handle_;  // symmetric transfer to the loop coroutine
     }
 
@@ -182,10 +194,7 @@ class generator {
                     auto& promise = h.promise();
                     auto consumer = promise.consumer_;
                     promise.leave_frame_context();
-                    if (consumer) {
-                        return consumer;  // symmetric transfer → consumer
-                    }
-                    return std::noop_coroutine();
+                    return detail::transfer_to_continuation(consumer, promise.parent());
                 }
 
                 void await_resume() const noexcept {}
@@ -206,7 +215,7 @@ class generator {
                 [[nodiscard]] std::coroutine_handle<> await_suspend(
                     std::coroutine_handle<promise_type> h) noexcept {
                     h.promise().leave_frame_context();
-                    return consumer;  // symmetric transfer → consumer
+                    return detail::transfer_to_continuation(consumer, h.promise().parent());
                 }
 
                 void await_resume() noexcept {}
@@ -273,8 +282,9 @@ class generator {
                 // transferring control back, including after internal I/O.
                 producer.promise().enter_frame_context(
                     promise_base::current_frame());
-                // Symmetric transfer → producer runs until co_yield / end.
-                return producer;
+                // Producer and consumer have independent ownership domains.
+                return detail::transfer_to_continuation(
+                    producer, std::addressof(producer.promise()));
             }
 
             std::optional<T> await_resume() {

include/elio/coro/promise_base.hpp

@@ -267,6 +267,13 @@ class promise_base {
         return execution_context_;
     }
 
+    /// @internal Lease-aware continuation handoff without materializing policy.
+    [[nodiscard]] std::shared_ptr<detail::continuation_owner>
+    pinned_continuation_owner() const noexcept {
+        return execution_context_
+            ? execution_context_->pinned_continuation_owner() : nullptr;
+    }
+
     /// Bind an unstarted transparent child to the context of the Elio promise
     /// that actually awaits it. A task created outside the awaiter may already
     /// have a provisional independent context; no task body has run yet, so
@@ -473,6 +480,18 @@ static_assert(std::is_standard_layout_v<promise_base>,
 /// resumer.
 namespace detail {
 
+[[nodiscard]] inline std::coroutine_handle<> transfer_to_continuation(
+    std::coroutine_handle<> continuation, promise_base* parent) noexcept {
+    if (continuation && parent) {
+        auto owner = parent->pinned_continuation_owner();
+        if (owner && !owner->is_current()) {
+            owner->schedule(continuation);
+            return std::noop_coroutine();
+        }
+    }
+    return continuation ? continuation : std::noop_coroutine();
+}
+
 class frame_context_scope {
 public:
     explicit frame_context_scope(promise_base* frame) noexcept

include/elio/coro/task.hpp

@@ -74,6 +74,12 @@ struct final_awaiter {
         h.promise().unlink_parent_cancellation();
         h.promise().leave_frame_context();
         if (continuation) {
+            // Cancellation-isolated scopes may execute elsewhere; returning
+            // from one must restore a still-leased parent's exact owner.
+            if (h.promise().direct_await_context_isolated()) {
+                return transfer_to_continuation(
+                    continuation, h.promise().parent());
+            }
             return continuation;
         } else if (h.promise().detached_) {
             // Keep the externally owned join state after freeing the frame so

include/elio/coro/task_execution_context.hpp

@@ -5,6 +5,7 @@
 #include <cassert>
 #include <cstddef>
 #include <cstdint>
+#include <coroutine>
 #include <limits>
 #include <memory>
 #include <mutex>
@@ -21,6 +22,12 @@ class task_execution_context;
 class promise_base;
 
 namespace detail {
+struct continuation_owner {
+    virtual ~continuation_owner() = default;
+    [[nodiscard]] virtual bool is_current() const noexcept = 0;
+    virtual void schedule(std::coroutine_handle<>) noexcept = 0;
+};
+
 struct task_execution_control_block;
 [[nodiscard]] std::shared_ptr<task_execution_context>
 make_task_execution_context();
@@ -106,6 +113,14 @@ class task_execution_context final {
         return active_io_pins_.load(std::memory_order_acquire);
     }
 
+    /// @internal A lease also carries its scheduler domain for foreign wakes.
+    [[nodiscard]] std::shared_ptr<detail::continuation_owner>
+    pinned_continuation_owner() const noexcept {
+        if (continuation_pins_.load(std::memory_order_acquire) == 0) return {};
+        std::lock_guard<std::mutex> lock(io_pin_mutex_);
+        return continuation_owner_;
+    }
+
     [[nodiscard]] bool is_io_pin_owner(
         size_t worker_id, uint64_t context_generation) const noexcept {
         return has_active_io_pin() &&
@@ -151,7 +166,8 @@ class task_execution_context final {
         return cancellation_context_.link_parent(std::move(parent));
     }
 
-    void acquire_io_pin(size_t worker_id, uint64_t context_generation) {
+    void acquire_io_pin(size_t worker_id, uint64_t context_generation,
+                        std::shared_ptr<detail::continuation_owner> owner = {}) {
         std::lock_guard<std::mutex> lock(io_pin_mutex_);
         size_t count = active_io_pins_.load(std::memory_order_relaxed);
         if (count != 0 &&
@@ -169,11 +185,16 @@ class task_execution_context final {
             io_context_generation_.store(
                 context_generation, std::memory_order_relaxed);
         }
+        if (owner) {
+            continuation_owner_ = std::move(owner);
+            continuation_pins_.fetch_add(1, std::memory_order_release);
+        }
         active_io_pins_.store(count + 1, std::memory_order_release);
     }
 
     void release_io_pin(size_t worker_id,
-                        uint64_t context_generation) noexcept {
+                        uint64_t context_generation,
+                        bool routes_continuation = false) noexcept {
         std::lock_guard<std::mutex> lock(io_pin_mutex_);
         size_t count = active_io_pins_.load(std::memory_order_relaxed);
         const bool valid =
@@ -184,6 +205,11 @@ class task_execution_context final {
         assert(valid && "I/O operation guard released a mismatched task pin");
         if (!valid) return;
 
+        if (routes_continuation &&
+            continuation_pins_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
+            continuation_owner_.reset();
+        }
+
         // Owner and generation deliberately remain as the last observed
         // identity when count reaches zero. The count is authoritative, and
         // retaining immutable snapshot fields avoids torn clear/read windows
@@ -198,6 +224,8 @@ class task_execution_context final {
     std::atomic<size_t> io_owner_worker_{NO_AFFINITY};
     std::atomic<uint64_t> io_context_generation_{0};
     std::atomic<size_t> active_io_pins_{0};
+    std::atomic<size_t> continuation_pins_{0};
+    std::shared_ptr<detail::continuation_owner> continuation_owner_;
 };
 
 namespace detail {

include/elio/coro/task_group.hpp

@@ -87,6 +87,12 @@ inline void resume_task_group_join_waiter(
     if (!waiter) {
         return;
     }
+    if (auto* promise = get_promise_base(waiter.address())) {
+        if (auto owner = promise->pinned_continuation_owner()) {
+            owner->schedule(waiter);
+            return;
+        }
+    }
     if (completion.scheduler->try_schedule(waiter)) {
         return;
     }

include/elio/elio.hpp

@@ -34,6 +34,7 @@
 #include "runtime/chase_lev_deque.hpp"
 #include "runtime/async_main.hpp"
 #include "runtime/affinity.hpp"
+#include "runtime/worker_lease.hpp"
 #include "runtime/serve.hpp"
 #include "runtime/spawn.hpp"
 #include "runtime/blocking_pool.hpp"

include/elio/io/io_operation_guard.hpp

@@ -22,7 +22,8 @@ class io_operation_guard final {
 
     io_operation_guard(
         std::shared_ptr<coro::task_execution_context> execution_context,
-        std::shared_ptr<io_context_identity> identity)
+        std::shared_ptr<io_context_identity> identity,
+        std::shared_ptr<coro::detail::continuation_owner> owner = {})
         : execution_context_(std::move(execution_context))
         , identity_(std::move(identity)) {
         if (!identity_) {
@@ -32,8 +33,10 @@ class io_operation_guard final {
 
         if (identity_->is_worker_owned()) {
             if (execution_context_) {
+                routes_continuation_ = bool(owner);
                 execution_context_->acquire_io_pin(
-                    identity_->owner_worker_id, identity_->generation);
+                    identity_->owner_worker_id, identity_->generation,
+                    std::move(owner));
                 task_pinned_ = true;
             }
         }
@@ -69,12 +72,14 @@ class io_operation_guard final {
         // during the release transition.
         if (task_pinned_) {
             execution_context_->release_io_pin(
-                identity_->owner_worker_id, identity_->generation);
+                identity_->owner_worker_id, identity_->generation,
+                routes_continuation_);
         }
         identity_->active_pins.fetch_sub(1, std::memory_order_acq_rel);
 
         active_ = false;
         task_pinned_ = false;
+        routes_continuation_ = false;
         execution_context_.reset();
         identity_.reset();
     }
@@ -89,12 +94,14 @@ class io_operation_guard final {
         identity_ = std::move(other.identity_);
         active_ = std::exchange(other.active_, false);
         task_pinned_ = std::exchange(other.task_pinned_, false);
+        routes_continuation_ = std::exchange(other.routes_continuation_, false);
     }
 
     std::shared_ptr<coro::task_execution_context> execution_context_;
     std::shared_ptr<io_context_identity> identity_;
     bool active_ = false;
     bool task_pinned_ = false;
+    bool routes_continuation_ = false;
 };
 
 } // namespace elio::io::detail