From 8574cef68d75df199639517fa0582e25584132bb Mon Sep 17 00:00:00 2001
From: Darius Clark <darius.clark@proton.me>
Date: Fri, 10 Apr 2026 09:56:55 -0500
Subject: [PATCH 1/5] feat: implemented scope tasks

---
 src/lib.rs    |  40 +++++++
 src/scoped.rs | 281 ++++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 321 insertions(+)
 create mode 100644 src/scoped.rs

diff --git a/src/lib.rs b/src/lib.rs
index b4b517b..e8c2952 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -7,6 +7,7 @@ pub mod tracker;
 #[cfg(feature = "either")]
 pub mod either;
 pub mod rc;
+pub(crate) mod scoped;
 
 use std::fmt::{Debug, Formatter};
 
@@ -18,6 +19,8 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
+pub use crate::scoped::{JoinError as ScopedJoinError, Scope, ScopedJoinHandle};
+
 #[cfg(all(
     not(feature = "threadpool"),
     not(feature = "tokio"),
@@ -467,6 +470,43 @@ pub trait Executor {
             _channel_tx: tx,
         }
     }
+
+    /// Create a structured-concurrency scope in which tasks may be spawned
+    /// that borrow from the enclosing stack frame.
+    ///
+    /// Unlike [`Executor::spawn`], tasks spawned on the [`Scope`] are driven
+    /// cooperatively by the returned future — they make progress whenever
+    /// the scope is polled — so they may borrow any data that outlives the
+    /// `'env` lifetime. Every task is either completed or cancelled before
+    /// `scope` returns, so borrows never outlive the stack frame.
+    ///
+    /// This is the async analogue of [`std::thread::scope`].
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # async fn run() {
+    /// use async_rt::Executor;
+    /// use async_rt::rt::tokio::TokioExecutor;
+    ///
+    /// let executor = TokioExecutor;
+    /// let data = vec![1, 2, 3, 4];
+    /// let sum = executor
+    ///     .scope(async |s| {
+    ///         let a = s.spawn(async { data[0] + data[1] });
+    ///         let b = s.spawn(async { data[2] + data[3] });
+    ///         a.await.unwrap() + b.await.unwrap()
+    ///     })
+    ///     .await;
+    /// assert_eq!(sum, 10);
+    /// # }
+    /// ```
+    fn scope<'env, F, T>(&self, f: F) -> impl Future<Output = T>
+    where
+        F: AsyncFnOnce(&Scope<'env>) -> T,
+    {
+        scoped::scope(f)
+    }
 }
 
 pub trait ExecutorBlocking: Executor {
diff --git a/src/scoped.rs b/src/scoped.rs
new file mode 100644
index 0000000..07f9b47
--- /dev/null
+++ b/src/scoped.rs
@@ -0,0 +1,281 @@
+//! Structured concurrency scopes for async tasks.
+//!
+//! A [`Scope`] lets you spawn futures that borrow from the enclosing stack
+//! frame, analogous to [`std::thread::scope`]. Unlike thread scope, the
+//! tasks are driven cooperatively by the scope's own future — they make
+//! progress whenever the scope is polled — so no runtime-level `spawn` is
+//! involved and no `unsafe` is needed to extend lifetimes.
+//!
+//! The [`scope`] function is the entry point. Inside the async closure,
+//! call [`Scope::spawn`] to enqueue a task; the returned [`ScopedJoinHandle`]
+//! resolves with the task's output. Any tasks still running when the user
+//! closure's future completes are drained before `scope` returns, so every
+//! borrow is released before the stack frame goes away.
+
+use futures::channel::oneshot;
+use futures::future::BoxFuture;
+use futures::stream::FuturesUnordered;
+use futures::{FutureExt, StreamExt};
+use parking_lot::Mutex;
+use std::future::{Future, poll_fn};
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A scope within which tasks can be spawned that borrow from the enclosing
+/// stack frame.
+///
+/// Obtain a `Scope` via the [`scope`] free function. The `'env` lifetime is
+/// the lifetime of data borrowed from outside the scope — spawned futures
+/// may reference any data that outlives `'env`.
+pub struct Scope<'env> {
+    tasks: Mutex<FuturesUnordered<BoxFuture<'env, ()>>>,
+    // Invariant in 'env: prevents the compiler from shrinking or extending
+    // 'env, which would otherwise let callers smuggle references in or out.
+    _env: PhantomData<&'env mut &'env ()>,
+}
+
+impl<'env> Scope<'env> {
+    fn new() -> Self {
+        Self {
+            tasks: Mutex::new(FuturesUnordered::new()),
+            _env: PhantomData,
+        }
+    }
+
+    /// Spawn a task into this scope.
+    ///
+    /// The future may borrow any data that outlives `'env`. The task will
+    /// be polled cooperatively alongside the scope's user closure and any
+    /// other spawned tasks.
+    pub fn spawn<Fut>(&self, fut: Fut) -> ScopedJoinHandle<Fut::Output>
+    where
+        Fut: Future + Send + 'env,
+        Fut::Output: Send + 'env,
+    {
+        let (tx, rx) = oneshot::channel();
+        let wrapped: BoxFuture<'env, ()> = async move {
+            let output = fut.await;
+            // If the receiver was dropped, the caller doesn't care about
+            // the output — just discard the send error.
+            let _ = tx.send(output);
+        }
+        .boxed();
+
+        self.tasks.lock().push(wrapped);
+
+        ScopedJoinHandle { rx }
+    }
+
+    /// Drive the spawned task set, returning `(made_progress, empty)`.
+    ///
+    /// Polls tasks in a loop until no more immediate progress can be made.
+    /// `made_progress` is `true` if at least one task completed this call;
+    /// `empty` is `true` if no tasks remain after polling.
+    fn drive_tasks(&self, cx: &mut Context<'_>) -> (bool, bool) {
+        let mut made_progress = false;
+        loop {
+            let mut tasks = self.tasks.lock();
+            match tasks.poll_next_unpin(cx) {
+                Poll::Ready(Some(())) => {
+                    made_progress = true;
+                    continue;
+                }
+                Poll::Ready(None) => return (made_progress, true),
+                Poll::Pending => return (made_progress, false),
+            }
+        }
+    }
+}
+
+/// A handle to a task spawned on a [`Scope`].
+///
+/// Awaiting the handle yields the task's output. If the scope is dropped
+/// before the task finishes (for example, because the scope future was
+/// cancelled), awaiting yields [`JoinError::Cancelled`].
+///
+/// The handle is `'static` — it carries no borrow of the scope — so it can
+/// be moved into other spawned tasks, channels, or futures without lifetime
+/// gymnastics.
+pub struct ScopedJoinHandle<T> {
+    rx: oneshot::Receiver<T>,
+}
+
+/// Error returned when a scoped task could not produce a value — because
+/// it was cancelled (its scope was dropped before it finished).
+#[derive(Debug)]
+pub enum JoinError {
+    /// The task was cancelled before it produced an output.
+    Cancelled,
+}
+
+impl std::fmt::Display for JoinError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            JoinError::Cancelled => f.write_str("scoped task was cancelled"),
+        }
+    }
+}
+
+impl std::error::Error for JoinError {}
+
+impl<T> Future for ScopedJoinHandle<T> {
+    type Output = Result<T, JoinError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match Pin::new(&mut self.rx).poll(cx) {
+            Poll::Ready(Ok(v)) => Poll::Ready(Ok(v)),
+            Poll::Ready(Err(_)) => Poll::Ready(Err(JoinError::Cancelled)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+/// Create a scope for spawning tasks that borrow from the enclosing stack
+/// frame.
+///
+/// The provided async closure receives a reference to the [`Scope`] and
+/// returns a future. That future is driven alongside any tasks spawned on
+/// the scope; when it completes, any still-running spawned tasks are
+/// drained before `scope` returns its value.
+///
+/// # Example
+///
+/// ```no_run
+/// # async fn run() {
+/// use async_rt::scoped::scope;
+///
+/// let data = vec![1, 2, 3, 4];
+/// let sum = scope(async |s| {
+///     let a = s.spawn(async { data[0] + data[1] });
+///     let b = s.spawn(async { data[2] + data[3] });
+///     a.await.unwrap() + b.await.unwrap()
+/// })
+/// .await;
+/// assert_eq!(sum, 10);
+/// # }
+/// ```
+pub async fn scope<'env, F, T>(f: F) -> T
+where
+    // `AsyncFnOnce` ties the returned future's lifetime to the `&Scope`
+    // argument, so the user future is allowed to borrow `&scope` (e.g. to
+    // call `scope.spawn(...)` repeatedly). An ordinary HRTB over
+    // `FnOnce(&Scope) -> Fut` can't express this because `Fut` is a single
+    // type chosen outside the HRTB.
+    F: AsyncFnOnce(&Scope<'env>) -> T,
+{
+    let scope: Scope<'env> = Scope::new();
+    let result = {
+        let user_fut = f(&scope);
+        let mut user_fut = std::pin::pin!(user_fut);
+
+        // Drive the user future and the spawned task set concurrently.
+        // Order matters: poll `user_fut` first so any `h.await` inside it
+        // can register its waker on the handle's oneshot channel before
+        // we poll the task that will fulfil it. Then drive the tasks; if
+        // a task completes, its `tx.send(...)` fires the handle's waker
+        // and we loop to re-poll `user_fut` in the same call — otherwise
+        // there would be no wakeup source to restart us.
+        poll_fn(|cx| {
+            loop {
+                if let Poll::Ready(r) = user_fut.as_mut().poll(cx) {
+                    return Poll::Ready(r);
+                }
+                let (made_progress, _empty) = scope.drive_tasks(cx);
+                if !made_progress {
+                    return Poll::Pending;
+                }
+            }
+        })
+        .await
+    };
+
+    // Drain any remaining spawned tasks before the scope goes away, so
+    // every borrow of `'env` data is released before this frame unwinds.
+    poll_fn(|cx| {
+        let (_made_progress, empty) = scope.drive_tasks(cx);
+        if empty {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    })
+    .await;
+
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn borrows_stack_data() {
+        let data = vec![1, 2, 3, 4];
+        let data = &data;
+        let sum = scope(async |s: &Scope<'_>| {
+            let a = s.spawn(async move { data[0] + data[1] });
+            let b = s.spawn(async move { data[2] + data[3] });
+            a.await.unwrap() + b.await.unwrap()
+        })
+        .await;
+        assert_eq!(sum, 10);
+    }
+
+    #[tokio::test]
+    async fn drains_unawaited_tasks() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let counter = AtomicUsize::new(0);
+        let counter_ref = &counter;
+        scope(async |s: &Scope<'_>| {
+            for _ in 0..8 {
+                s.spawn(async move {
+                    counter_ref.fetch_add(1, Ordering::SeqCst);
+                });
+            }
+        })
+        .await;
+        assert_eq!(counter.load(Ordering::SeqCst), 8);
+    }
+
+    #[tokio::test]
+    async fn returns_closure_value() {
+        let v: i32 = scope(async |_s: &Scope<'_>| 42).await;
+        assert_eq!(v, 42);
+    }
+
+    #[tokio::test]
+    async fn join_handle_yields_output() {
+        let out = scope(async |s: &Scope<'_>| {
+            let h = s.spawn(async { "hello" });
+            h.await.unwrap()
+        })
+        .await;
+        assert_eq!(out, "hello");
+    }
+
+    #[tokio::test]
+    async fn many_concurrent_tasks_complete() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let counter = AtomicUsize::new(0);
+        let counter_ref = &counter;
+        let total: usize = scope(async |s: &Scope<'_>| {
+            let handles: Vec<_> = (0..32)
+                .map(|i| {
+                    s.spawn(async move {
+                        counter_ref.fetch_add(1, Ordering::SeqCst);
+                        i
+                    })
+                })
+                .collect();
+            let mut sum = 0usize;
+            for h in handles {
+                sum += h.await.unwrap();
+            }
+            sum
+        })
+        .await;
+        assert_eq!(total, (0..32).sum());
+        assert_eq!(counter.load(Ordering::SeqCst), 32);
+    }
+}

From 197ccc811f6a9fee684bff72274634611006cd24 Mon Sep 17 00:00:00 2001
From: Darius Clark <darius.clark@proton.me>
Date: Mon, 13 Apr 2026 06:17:18 -0500
Subject: [PATCH 2/5] chore: add Executor::executor_scope

---
 src/lib.rs    |  67 +++++++-
 src/scoped.rs | 462 ++++++++++++++++++++++++++++++++++++++++++++++----
 2 files changed, 492 insertions(+), 37 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index e8c2952..31b57f1 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -7,7 +7,7 @@ pub mod tracker;
 #[cfg(feature = "either")]
 pub mod either;
 pub mod rc;
-pub(crate) mod scoped;
+pub mod scoped;
 
 use std::fmt::{Debug, Formatter};
 
@@ -19,7 +19,9 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-pub use crate::scoped::{JoinError as ScopedJoinError, Scope, ScopedJoinHandle};
+pub use crate::scoped::{
+    JoinError as ScopedJoinError, Scope, ScopeExecutor, ScopedJoinHandle,
+};
 
 #[cfg(all(
     not(feature = "threadpool"),
@@ -268,6 +270,16 @@ impl<T> Debug for CommunicationTask<T> {
 }
 
 impl<T> CommunicationTask<T> {
+    pub(crate) fn new(
+        task_handle: AbortableJoinHandle<()>,
+        channel_tx: futures::channel::mpsc::Sender<T>,
+    ) -> Self {
+        Self {
+            _task_handle: task_handle,
+            _channel_tx: channel_tx,
+        }
+    }
+
     /// Send a message to the task
     pub async fn send(&mut self, data: T) -> std::io::Result<()> {
         self._channel_tx
@@ -318,6 +330,16 @@ impl<T> Debug for UnboundedCommunicationTask<T> {
 }
 
 impl<T> UnboundedCommunicationTask<T> {
+    pub(crate) fn new(
+        task_handle: AbortableJoinHandle<()>,
+        channel_tx: futures::channel::mpsc::UnboundedSender<T>,
+    ) -> Self {
+        Self {
+            _task_handle: task_handle,
+            _channel_tx: channel_tx,
+        }
+    }
+
     /// Send a message to task
     pub fn send(&mut self, data: T) -> std::io::Result<()> {
         self._channel_tx
@@ -507,6 +529,47 @@ pub trait Executor {
     {
         scoped::scope(f)
     }
+
+    /// Run an async closure with a scoped [`Executor`] wrapper that
+    /// forwards spawns to this executor, waits for all spawned tasks
+    /// to finish when the closure returns, and aborts any outstanding
+    /// tasks if the scope future itself is cancelled.
+    ///
+    /// Unlike [`Executor::scope`], tasks run on the real executor (so
+    /// they get real parallelism) but must be `Send + 'static`.
+    ///
+    /// # Panics in spawned tasks
+    ///
+    /// Panics inside tasks spawned via this scope are **not** propagated
+    /// to the caller. See [`scoped::executor_scope`] for details on
+    /// per-backend behaviour. If you need to react to a task panic,
+    /// await its [`JoinHandle`] directly.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # async fn run() {
+    /// use async_rt::Executor;
+    /// use async_rt::rt::tokio::TokioExecutor;
+    ///
+    /// let executor = TokioExecutor;
+    /// let total = executor
+    ///     .executor_scope(async |s| {
+    ///         let a = s.spawn(async { 1 + 2 });
+    ///         let b = s.spawn(async { 3 + 4 });
+    ///         a.await.unwrap() + b.await.unwrap()
+    ///     })
+    ///     .await;
+    /// assert_eq!(total, 10);
+    /// # }
+    /// ```
+    fn executor_scope<'scope, F, T>(&'scope self, f: F) -> impl Future<Output = T>
+    where
+        Self: Sized,
+        F: AsyncFnOnce(&ScopeExecutor<'scope, Self>) -> T,
+    {
+        scoped::executor_scope(self, f)
+    }
 }
 
 pub trait ExecutorBlocking: Executor {
diff --git a/src/scoped.rs b/src/scoped.rs
index 07f9b47..6e8bc0e 100644
--- a/src/scoped.rs
+++ b/src/scoped.rs
@@ -2,8 +2,8 @@
 //!
 //! A [`Scope`] lets you spawn futures that borrow from the enclosing stack
 //! frame, analogous to [`std::thread::scope`]. Unlike thread scope, the
-//! tasks are driven cooperatively by the scope's own future — they make
-//! progress whenever the scope is polled — so no runtime-level `spawn` is
+//! tasks are driven cooperatively by the scope's own future, they make
+//! progress whenever the scope is polled so no runtime-level `spawn` is
 //! involved and no `unsafe` is needed to extend lifetimes.
 //!
 //! The [`scope`] function is the entry point. Inside the async closure,
@@ -12,26 +12,25 @@
 //! closure's future completes are drained before `scope` returns, so every
 //! borrow is released before the stack frame goes away.
 
+use crate::{
+    AbortableJoinHandle, CommunicationTask, Executor, InnerJoinHandle, JoinHandle,
+    UnboundedCommunicationTask,
+};
+use futures::channel::mpsc::{Receiver, UnboundedReceiver};
 use futures::channel::oneshot;
-use futures::future::BoxFuture;
+use futures::future::{AbortHandle, Abortable, BoxFuture};
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt, StreamExt};
 use parking_lot::Mutex;
-use std::future::{Future, poll_fn};
-use std::marker::PhantomData;
-use std::pin::Pin;
-use std::task::{Context, Poll};
+use core::future::{Future, poll_fn};
+use core::marker::PhantomData;
+use core::pin::Pin;
+use core::task::{Context, Poll};
 
 /// A scope within which tasks can be spawned that borrow from the enclosing
 /// stack frame.
-///
-/// Obtain a `Scope` via the [`scope`] free function. The `'env` lifetime is
-/// the lifetime of data borrowed from outside the scope — spawned futures
-/// may reference any data that outlives `'env`.
 pub struct Scope<'env> {
     tasks: Mutex<FuturesUnordered<BoxFuture<'env, ()>>>,
-    // Invariant in 'env: prevents the compiler from shrinking or extending
-    // 'env, which would otherwise let callers smuggle references in or out.
     _env: PhantomData<&'env mut &'env ()>,
 }
 
@@ -45,7 +44,7 @@ impl<'env> Scope<'env> {
 
     /// Spawn a task into this scope.
     ///
-    /// The future may borrow any data that outlives `'env`. The task will
+    /// The future may borrow any data that outlives its lifetime `'env`. The task will
     /// be polled cooperatively alongside the scope's user closure and any
     /// other spawned tasks.
     pub fn spawn<Fut>(&self, fut: Fut) -> ScopedJoinHandle<Fut::Output>
@@ -57,7 +56,7 @@ impl<'env> Scope<'env> {
         let wrapped: BoxFuture<'env, ()> = async move {
             let output = fut.await;
             // If the receiver was dropped, the caller doesn't care about
-            // the output — just discard the send error.
+            // the output so we will discard it.
             let _ = tx.send(output);
         }
         .boxed();
@@ -67,6 +66,138 @@ impl<'env> Scope<'env> {
         ScopedJoinHandle { rx }
     }
 
+    /// Spawn a task into this scope and return an [`AbortableJoinHandle`].
+    ///
+    /// This mirrors [`Executor::spawn_abortable`] for cooperatively
+    /// scheduled tasks. The returned handle aborts the task when all
+    /// references to it have been dropped.
+    pub fn spawn_abortable<Fut>(&self, fut: Fut) -> AbortableJoinHandle<Fut::Output>
+    where
+        Fut: Future + Send + 'env,
+        Fut::Output: Send + 'env,
+    {
+        let (abort_handle, abort_reg) = AbortHandle::new_pair();
+        let abortable = Abortable::new(fut, abort_reg);
+        let (tx, rx) = oneshot::channel();
+
+        let wrapped: BoxFuture<'env, ()> = async move {
+            let val = abortable.await;
+            let _ = tx.send(val);
+        }
+        .boxed();
+        self.tasks.lock().push(wrapped);
+
+        let join = JoinHandle {
+            inner: InnerJoinHandle::CustomHandle {
+                inner: Some(rx),
+                handle: abort_handle,
+            },
+        };
+        AbortableJoinHandle::from(join)
+    }
+
+    /// Spawn a task into this scope without keeping a handle to it.
+    ///
+    /// Equivalent to [`Executor::dispatch`] for scoped tasks.
+    pub fn dispatch<Fut>(&self, fut: Fut)
+    where
+        Fut: Future + Send + 'env,
+        Fut::Output: Send + 'env,
+    {
+        let _ = self.spawn(fut);
+    }
+
+    /// Spawn a message-driven coroutine into this scope.
+    ///
+    /// Equivalent to [`Executor::spawn_coroutine`] for scoped tasks.
+    pub fn spawn_coroutine<T, F, Fut>(&self, f: F) -> CommunicationTask<T>
+    where
+        F: FnMut(Receiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        self.spawn_coroutine_with_buffer(1, f)
+    }
+
+    /// Like [`Scope::spawn_coroutine`] but with a configurable channel
+    /// buffer.
+    pub fn spawn_coroutine_with_buffer<T, F, Fut>(
+        &self,
+        buffer: usize,
+        mut f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(Receiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        let (tx, rx) = futures::channel::mpsc::channel(buffer);
+        let task_handle = self.spawn_abortable(f(rx));
+        CommunicationTask::new(task_handle, tx)
+    }
+
+    /// Like [`Scope::spawn_coroutine`] but passes a caller-provided
+    /// context into the coroutine alongside the message receiver.
+    pub fn spawn_coroutine_with_context<T, F, C, Fut>(
+        &self,
+        context: C,
+        f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(C, Receiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        self.spawn_coroutine_with_buffer_and_context(context, 1, f)
+    }
+
+    /// Like [`Scope::spawn_coroutine_with_context`] but with a
+    /// configurable channel buffer.
+    pub fn spawn_coroutine_with_buffer_and_context<T, F, C, Fut>(
+        &self,
+        context: C,
+        buffer: usize,
+        mut f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(C, Receiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        let (tx, rx) = futures::channel::mpsc::channel(buffer);
+        let task_handle = self.spawn_abortable(f(context, rx));
+        CommunicationTask::new(task_handle, tx)
+    }
+
+    /// Spawn an unbounded message-driven coroutine into this scope.
+    ///
+    /// Equivalent to [`Executor::spawn_unbounded_coroutine`] for scoped
+    /// tasks.
+    pub fn spawn_unbounded_coroutine<T, F, Fut>(
+        &self,
+        mut f: F,
+    ) -> UnboundedCommunicationTask<T>
+    where
+        F: FnMut(UnboundedReceiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        let (tx, rx) = futures::channel::mpsc::unbounded();
+        let task_handle = self.spawn_abortable(f(rx));
+        UnboundedCommunicationTask::new(task_handle, tx)
+    }
+
+    /// Like [`Scope::spawn_unbounded_coroutine`] but passes a
+    /// caller-provided context into the coroutine.
+    pub fn spawn_unbounded_coroutine_with_context<T, F, C, Fut>(
+        &self,
+        context: C,
+        mut f: F,
+    ) -> UnboundedCommunicationTask<T>
+    where
+        F: FnMut(C, UnboundedReceiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'env,
+    {
+        let (tx, rx) = futures::channel::mpsc::unbounded();
+        let task_handle = self.spawn_abortable(f(context, rx));
+        UnboundedCommunicationTask::new(task_handle, tx)
+    }
+
     /// Drive the spawned task set, returning `(made_progress, empty)`.
     ///
     /// Polls tasks in a loop until no more immediate progress can be made.
@@ -94,22 +225,21 @@ impl<'env> Scope<'env> {
 /// before the task finishes (for example, because the scope future was
 /// cancelled), awaiting yields [`JoinError::Cancelled`].
 ///
-/// The handle is `'static` — it carries no borrow of the scope — so it can
-/// be moved into other spawned tasks, channels, or futures without lifetime
-/// gymnastics.
+/// # Note
+/// The handle is `'static` so it can  be moved into other spawned tasks, channels, or futures
+/// without lifetime gymnastics.
 pub struct ScopedJoinHandle<T> {
     rx: oneshot::Receiver<T>,
 }
 
-/// Error returned when a scoped task could not produce a value — because
-/// it was cancelled (its scope was dropped before it finished).
+/// Error returned when a scoped task was cancelled before it produced an output
 #[derive(Debug)]
 pub enum JoinError {
     /// The task was cancelled before it produced an output.
     Cancelled,
 }
 
-impl std::fmt::Display for JoinError {
+impl core::fmt::Display for JoinError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             JoinError::Cancelled => f.write_str("scoped task was cancelled"),
@@ -117,7 +247,7 @@ impl std::fmt::Display for JoinError {
     }
 }
 
-impl std::error::Error for JoinError {}
+impl core::error::Error for JoinError {}
 
 impl<T> Future for ScopedJoinHandle<T> {
     type Output = Result<T, JoinError>;
@@ -146,6 +276,7 @@ impl<T> Future for ScopedJoinHandle<T> {
 /// use async_rt::scoped::scope;
 ///
 /// let data = vec![1, 2, 3, 4];
+/// let data = &data;
 /// let sum = scope(async |s| {
 ///     let a = s.spawn(async { data[0] + data[1] });
 ///     let b = s.spawn(async { data[2] + data[3] });
@@ -157,11 +288,6 @@ impl<T> Future for ScopedJoinHandle<T> {
 /// ```
 pub async fn scope<'env, F, T>(f: F) -> T
 where
-    // `AsyncFnOnce` ties the returned future's lifetime to the `&Scope`
-    // argument, so the user future is allowed to borrow `&scope` (e.g. to
-    // call `scope.spawn(...)` repeatedly). An ordinary HRTB over
-    // `FnOnce(&Scope) -> Fut` can't express this because `Fut` is a single
-    // type chosen outside the HRTB.
     F: AsyncFnOnce(&Scope<'env>) -> T,
 {
     let scope: Scope<'env> = Scope::new();
@@ -169,13 +295,6 @@ where
         let user_fut = f(&scope);
         let mut user_fut = std::pin::pin!(user_fut);
 
-        // Drive the user future and the spawned task set concurrently.
-        // Order matters: poll `user_fut` first so any `h.await` inside it
-        // can register its waker on the handle's oneshot channel before
-        // we poll the task that will fulfil it. Then drive the tasks; if
-        // a task completes, its `tx.send(...)` fires the handle's waker
-        // and we loop to re-poll `user_fut` in the same call — otherwise
-        // there would be no wakeup source to restart us.
         poll_fn(|cx| {
             loop {
                 if let Poll::Ready(r) = user_fut.as_mut().poll(cx) {
@@ -191,7 +310,7 @@ where
     };
 
     // Drain any remaining spawned tasks before the scope goes away, so
-    // every borrow of `'env` data is released before this frame unwinds.
+    // every borrow of the data is released before this frame unwinds.
     poll_fn(|cx| {
         let (_made_progress, empty) = scope.drive_tasks(cx);
         if empty {
@@ -205,8 +324,134 @@ where
     result
 }
 
+/// An [`Executor`] wrapper that tracks spawned tasks so they can be
+/// cancelled when a scope ends.
+///
+/// `ScopeExecutor` itself implements [`Executor`], so it can be passed to
+/// any code that takes `Executor` every spawn performed through
+/// that code will be tracked by the scope.
+///
+/// Because submitted futures must still be `Send + 'static` (enforced by
+/// [`Executor::spawn`]), this flavour of scope does *not* allow borrowing
+/// from the enclosing stack frame. Use [`scope`] for that.
+pub struct ScopeExecutor<'scope, E> {
+    inner: &'scope E,
+    task_handles: Mutex<Vec<JoinHandle<()>>>,
+    _scope: PhantomData<&'scope mut &'scope ()>,
+}
+
+impl<'scope, E> ScopeExecutor<'scope, E> {
+    fn new(inner: &'scope E) -> Self {
+        Self {
+            inner,
+            task_handles: Mutex::new(Vec::new()),
+            _scope: PhantomData,
+        }
+    }
+
+    /// Abort every task currently tracked by this scope.
+    fn abort_all(&self) {
+        for handle in self.task_handles.lock().iter() {
+            handle.abort();
+        }
+    }
+}
+
+impl<E> Drop for ScopeExecutor<'_, E> {
+    fn drop(&mut self) {
+        self.abort_all();
+    }
+}
+
+impl<'scope, E> Executor for ScopeExecutor<'scope, E>
+where
+    E: Executor,
+{
+    fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
+        let abortable = Abortable::new(future, abort_registration);
+        let (tx, rx) = oneshot::channel();
+        let wrapped = async move {
+            let val = abortable.await;
+            let _ = tx.send(val);
+        };
+
+        let task_handle = self.inner.spawn(wrapped);
+        self.task_handles.lock().push(task_handle);
+
+        JoinHandle {
+            inner: InnerJoinHandle::CustomHandle {
+                inner: Some(rx),
+                handle: abort_handle,
+            },
+        }
+    }
+}
+
+/// Run an async closure with a scoped [`Executor`] wrapper.
+///
+/// If the future containing `executor_scope` itself is dropped
+/// (canceled externally), it will abort every outstanding task.
+///
+/// Unlike [`scope`], futures spawned here must be `Send + 'static`
+/// they're going onto the real executor, so they can't borrow from the
+/// enclosing stack frame.
+///
+/// # Panics in spawned tasks
+///
+/// Panics inside spawned tasks are **not** propagated to the scope.
+/// Unlike [`std::thread::scope`], `executor_scope` does not re-raise
+/// panics collected from its tasks: the panic is caught at the task
+/// boundary by the underlying runtime, and the scope simply treats the
+/// task as finished.
+///
+/// If you need to react to a task panic, poll or `.await` the returned
+/// [`JoinHandle`] yourself and check its result and don't rely on the
+/// scope to surface it.
+///
+/// # Example
+///
+/// ```no_run
+/// # async fn run() {
+/// use async_rt::Executor;
+/// use async_rt::rt::tokio::TokioExecutor;
+///
+/// let executor = TokioExecutor;
+/// let total = executor
+///     .executor_scope(async |s| {
+///         let a = s.spawn(async { 1 + 2 });
+///         let b = s.spawn(async { 3 + 4 });
+///         a.await.unwrap() + b.await.unwrap()
+///     })
+///     .await;
+/// assert_eq!(total, 10);
+/// # }
+/// ```
+pub async fn executor_scope<'scope, E, F, T>(executor: &'scope E, f: F) -> T
+where
+    E: Executor,
+    F: AsyncFnOnce(&ScopeExecutor<'scope, E>) -> T,
+{
+    let scope_exec = ScopeExecutor::new(executor);
+    let result = f(&scope_exec).await;
+
+    // Wait for every spawned task to complete
+    let handles: Vec<_> = scope_exec.task_handles.lock().drain(..).collect();
+    for handle in handles {
+        let _ = handle.await;
+    }
+
+    result
+}
+
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+    use futures_timer::Delay;
     use super::*;
 
     #[tokio::test]
@@ -278,4 +523,151 @@ mod tests {
         assert_eq!(total, (0..32).sum());
         assert_eq!(counter.load(Ordering::SeqCst), 32);
     }
+
+    #[tokio::test]
+    async fn executor_scope_runs_tasks() {
+        use crate::rt::tokio::TokioExecutor;
+        let executor = TokioExecutor;
+        let total = executor
+            .executor_scope(async |s| {
+                let a = s.spawn(async { 1 + 2 });
+                let b = s.spawn(async { 3 + 4 });
+                a.await.unwrap() + b.await.unwrap()
+            })
+            .await;
+        assert_eq!(total, 10);
+    }
+
+    #[tokio::test]
+    async fn scope_spawn_coroutine_receives_messages() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let total = AtomicUsize::new(0);
+        let total_ref = &total;
+
+        scope(async |s: &Scope<'_>| {
+            let mut task = s.spawn_coroutine(|mut rx| async move {
+                while let Some(value) = rx.next().await {
+                    total_ref.fetch_add(value, Ordering::SeqCst);
+                }
+            });
+            for v in [1usize, 2, 3, 4] {
+                task.send(v).await.unwrap();
+            }
+            drop(task); // closes channel → coroutine exits cleanly
+        })
+        .await;
+
+        assert_eq!(total.load(Ordering::SeqCst), 10);
+    }
+
+    #[tokio::test]
+    async fn scope_dispatch_runs_fire_and_forget() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+        let flag = AtomicBool::new(false);
+        let flag_ref = &flag;
+
+        scope(async |s: &Scope<'_>| {
+            s.dispatch(async move {
+                flag_ref.store(true, Ordering::SeqCst);
+            });
+        })
+        .await;
+
+        assert!(flag.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn executor_scope_drains_unawaited_tasks() {
+        use crate::rt::tokio::TokioExecutor;
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        let executor = TokioExecutor;
+        let flag = Arc::new(AtomicBool::new(false));
+        let flag_clone = flag.clone();
+
+        executor
+            .executor_scope(async move |s| {
+                // Spawn a task but do NOT await its handle.
+                // executor_scope should wait for it to complete before
+                // returning.
+                let _h = s.spawn(async move {
+                    Delay::new(Duration::from_millis(50)).await;
+                    flag_clone.store(true, Ordering::SeqCst);
+                });
+            })
+            .await;
+
+        assert!(
+            flag.load(Ordering::SeqCst),
+            "unawaited task should have completed before executor_scope returned"
+        );
+    }
+
+    #[tokio::test]
+    async fn executor_scope_swallows_task_panic() {
+        use crate::rt::tokio::TokioExecutor;
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let executor = TokioExecutor;
+        let sibling_done = Arc::new(AtomicUsize::new(0));
+        let sibling_done_clone = sibling_done.clone();
+
+        let result = executor
+            .executor_scope(async move |s| {
+                // Task A: panics. Its JoinHandle is dropped without
+                // being awaited. the panic goes to the runtime.
+                let _panicker = s.spawn(async {
+                    panic!("deliberate test panic");
+                });
+                // Task B: completes normally. The scope should still
+                // drain it before returning.
+                let _sibling = s.spawn(async move {
+                    sibling_done_clone.fetch_add(1, Ordering::SeqCst);
+                });
+                42usize
+            })
+            .await;
+
+        assert_eq!(result, 42);
+        assert_eq!(sibling_done.load(Ordering::SeqCst), 1);
+    }
+
+    #[tokio::test]
+    async fn executor_scope_aborts_on_external_cancel() {
+        use crate::rt::tokio::TokioExecutor;
+        use futures::future::Either;
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        let executor = TokioExecutor;
+        let flag = Arc::new(AtomicBool::new(false));
+        let flag_clone = flag.clone();
+
+        {
+            let scope_fut = executor.executor_scope(async move |s| {
+                let _h = s.spawn(async move {
+                    Delay::new(Duration::from_millis(200)).await;
+                    flag_clone.store(true, Ordering::SeqCst);
+                });
+                futures::future::pending::<()>().await;
+            });
+
+            let scope_fut = std::pin::pin!(scope_fut);
+            let timer = std::pin::pin!(Delay::new(Duration::from_millis(30)));
+            let result = futures::future::select(scope_fut, timer).await;
+            assert!(
+                matches!(result, Either::Right(_)),
+                "timer should have won the race"
+            );
+        }
+
+        // Give the (supposedly aborted) task plenty of time.
+        Delay::new(Duration::from_millis(300)).await;
+        assert!(
+            !flag.load(Ordering::SeqCst),
+            "task should have been aborted by scope drop"
+        );
+    }
 }

From d6c64b75a90aff18aba061a58e13ed651bf6a072 Mon Sep 17 00:00:00 2001
From: Darius Clark <darius.clark@proton.me>
Date: Sat, 25 Apr 2026 20:57:27 -0500
Subject: [PATCH 3/5] chore: use a vector of futures

---
 src/scoped.rs | 61 ++++++++++++++++++++++++++++++++-------------------
 1 file changed, 38 insertions(+), 23 deletions(-)

diff --git a/src/scoped.rs b/src/scoped.rs
index 6e8bc0e..c1c9f49 100644
--- a/src/scoped.rs
+++ b/src/scoped.rs
@@ -30,14 +30,14 @@ use core::task::{Context, Poll};
 /// A scope within which tasks can be spawned that borrow from the enclosing
 /// stack frame.
 pub struct Scope<'env> {
-    tasks: Mutex<FuturesUnordered<BoxFuture<'env, ()>>>,
+    inbox: Mutex<Vec<BoxFuture<'env, ()>>>,
     _env: PhantomData<&'env mut &'env ()>,
 }
 
 impl<'env> Scope<'env> {
     fn new() -> Self {
         Self {
-            tasks: Mutex::new(FuturesUnordered::new()),
+            inbox: Mutex::new(Vec::new()),
             _env: PhantomData,
         }
     }
@@ -61,7 +61,7 @@ impl<'env> Scope<'env> {
         }
         .boxed();
 
-        self.tasks.lock().push(wrapped);
+        self.inbox.lock().push(wrapped);
 
         ScopedJoinHandle { rx }
     }
@@ -85,7 +85,7 @@ impl<'env> Scope<'env> {
             let _ = tx.send(val);
         }
         .boxed();
-        self.tasks.lock().push(wrapped);
+        self.inbox.lock().push(wrapped);
 
         let join = JoinHandle {
             inner: InnerJoinHandle::CustomHandle {
@@ -198,22 +198,32 @@ impl<'env> Scope<'env> {
         UnboundedCommunicationTask::new(task_handle, tx)
     }
 
-    /// Drive the spawned task set, returning `(made_progress, empty)`.
-    ///
-    /// Polls tasks in a loop until no more immediate progress can be made.
-    /// `made_progress` is `true` if at least one task completed this call;
-    /// `empty` is `true` if no tasks remain after polling.
-    fn drive_tasks(&self, cx: &mut Context<'_>) -> (bool, bool) {
-        let mut made_progress = false;
-        loop {
-            let mut tasks = self.tasks.lock();
-            match tasks.poll_next_unpin(cx) {
-                Poll::Ready(Some(())) => {
-                    made_progress = true;
-                    continue;
+}
+
+/// Drive a scope once: absorb any inbox entries, then poll the active
+/// set until no more immediate progress can be made.
+fn drive_scope<'env>(
+    active: &mut FuturesUnordered<BoxFuture<'env, ()>>,
+    scope: &Scope<'env>,
+    cx: &mut Context<'_>,
+) -> (bool, bool) {
+    let mut made_progress = false;
+    loop {
+        // Absorb any newly-spawned tasks into the active set. Take the
+        // Vec out with mem::take so the lock is held just long enough
+        // to swap, never across a poll.
+        let incoming = std::mem::take(&mut *scope.inbox.lock());
+        active.extend(incoming);
+        match active.poll_next_unpin(cx) {
+            Poll::Ready(Some(())) => made_progress = true,
+            Poll::Ready(None) => return (made_progress, true),
+            Poll::Pending => {
+                // A task may have pushed new futures to the inbox
+                // during its own poll, so it doesn't return Pending until
+                // we've had a chance to poll them.
+                if scope.inbox.lock().is_empty() {
+                    return (made_progress, false);
                 }
-                Poll::Ready(None) => return (made_progress, true),
-                Poll::Pending => return (made_progress, false),
             }
         }
     }
@@ -291,6 +301,11 @@ where
     F: AsyncFnOnce(&Scope<'env>) -> T,
 {
     let scope: Scope<'env> = Scope::new();
+    // The active task set lives on the driver, not inside `Scope`. Only
+    // `drive_scope` touches it, so no lock guards it, and the inbox mutex
+    // is enough for the spawn side.
+    let mut active: FuturesUnordered<BoxFuture<'env, ()>> = FuturesUnordered::new();
+
     let result = {
         let user_fut = f(&scope);
         let mut user_fut = std::pin::pin!(user_fut);
@@ -300,7 +315,7 @@ where
                 if let Poll::Ready(r) = user_fut.as_mut().poll(cx) {
                     return Poll::Ready(r);
                 }
-                let (made_progress, _empty) = scope.drive_tasks(cx);
+                let (made_progress, _empty) = drive_scope(&mut active, &scope, cx);
                 if !made_progress {
                     return Poll::Pending;
                 }
@@ -312,7 +327,7 @@ where
     // Drain any remaining spawned tasks before the scope goes away, so
     // every borrow of the data is released before this frame unwinds.
     poll_fn(|cx| {
-        let (_made_progress, empty) = scope.drive_tasks(cx);
+        let (_made_progress, empty) = drive_scope(&mut active, &scope, cx);
         if empty {
             Poll::Ready(())
         } else {
@@ -325,7 +340,7 @@ where
 }
 
 /// An [`Executor`] wrapper that tracks spawned tasks so they can be
-/// cancelled when a scope ends.
+/// canceled when a scope ends.
 ///
 /// `ScopeExecutor` itself implements [`Executor`], so it can be passed to
 /// any code that takes `Executor` every spawn performed through
@@ -428,7 +443,7 @@ where
 ///         a.await.unwrap() + b.await.unwrap()
 ///     })
 ///     .await;
-/// assert_eq!(total, 10);
+///     assert_eq!(total, 10);
 /// # }
 /// ```
 pub async fn executor_scope<'scope, E, F, T>(executor: &'scope E, f: F) -> T

From c2bfc382d51033362de0d1e0a46bdf90a526570c Mon Sep 17 00:00:00 2001
From: Darius Clark <darius.clark@proton.me>
Date: Wed, 22 Jul 2026 22:27:57 -0500
Subject: [PATCH 4/5] feat: add cancellation during draining, and match
 executor and support nested scoped spawning

---
 src/lib.rs    |   6 +-
 src/scoped.rs | 459 +++++++++++++++++++++++++++++++++++++++++---------
 2 files changed, 377 insertions(+), 88 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 9026f24..fea1122 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -19,9 +19,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-pub use crate::scoped::{
-    JoinError as ScopedJoinError, Scope, ScopeExecutor, ScopedJoinHandle,
-};
+pub use crate::scoped::{JoinError as ScopedJoinError, Scope, ScopeExecutor, ScopedJoinHandle};
 
 #[cfg(all(
     not(feature = "threadpool"),
@@ -662,7 +660,7 @@ pub trait Executor {
     /// ```
     fn scope<'env, F, T>(&self, f: F) -> impl Future<Output = T>
     where
-        F: AsyncFnOnce(&Scope<'env>) -> T,
+        F: for<'scope> AsyncFnOnce(&'scope Scope<'scope, 'env>) -> T,
     {
         scoped::scope(f)
     }
diff --git a/src/scoped.rs b/src/scoped.rs
index c1c9f49..0220f65 100644
--- a/src/scoped.rs
+++ b/src/scoped.rs
@@ -16,44 +16,57 @@ use crate::{
     AbortableJoinHandle, CommunicationTask, Executor, InnerJoinHandle, JoinHandle,
     UnboundedCommunicationTask,
 };
+use core::future::{Future, poll_fn};
+use core::marker::PhantomData;
+use core::pin::Pin;
+use core::task::{Context, Poll};
 use futures::channel::mpsc::{Receiver, UnboundedReceiver};
 use futures::channel::oneshot;
 use futures::future::{AbortHandle, Abortable, BoxFuture};
 use futures::stream::FuturesUnordered;
 use futures::{FutureExt, StreamExt};
 use parking_lot::Mutex;
-use core::future::{Future, poll_fn};
-use core::marker::PhantomData;
-use core::pin::Pin;
-use core::task::{Context, Poll};
+use std::sync::{Arc, Weak};
+
+struct ScopeState<'scope> {
+    inbox: Mutex<Vec<BoxFuture<'scope, ()>>>,
+}
 
 /// A scope within which tasks can be spawned that borrow from the enclosing
 /// stack frame.
-pub struct Scope<'env> {
-    inbox: Mutex<Vec<BoxFuture<'env, ()>>>,
+pub struct Scope<'scope, 'env: 'scope> {
+    state: Weak<ScopeState<'scope>>,
+    _scope: PhantomData<&'scope mut &'scope ()>,
     _env: PhantomData<&'env mut &'env ()>,
 }
 
-impl<'env> Scope<'env> {
+impl<'scope, 'env> Scope<'scope, 'env> {
     fn new() -> Self {
         Self {
-            inbox: Mutex::new(Vec::new()),
+            state: Weak::new(),
+            _scope: PhantomData,
             _env: PhantomData,
         }
     }
 
+    fn push(&self, task: BoxFuture<'scope, ()>) {
+        if let Some(state) = self.state.upgrade() {
+            state.inbox.lock().push(task);
+        }
+    }
+
     /// Spawn a task into this scope.
     ///
     /// The future may borrow any data that outlives its lifetime `'env`. The task will
     /// be polled cooperatively alongside the scope's user closure and any
     /// other spawned tasks.
-    pub fn spawn<Fut>(&self, fut: Fut) -> ScopedJoinHandle<Fut::Output>
+    pub fn spawn<Fut>(&'scope self, fut: Fut) -> ScopedJoinHandle<Fut::Output>
     where
-        Fut: Future + Send + 'env,
-        Fut::Output: Send + 'env,
+        Fut: Future + Send + 'scope,
+        Fut::Output: Send + 'scope,
     {
         let (tx, rx) = oneshot::channel();
-        let wrapped: BoxFuture<'env, ()> = async move {
+        let wrapped: BoxFuture<'scope, ()> = async move {
             let output = fut.await;
             // If the receiver was dropped, the caller doesn't care about
             // the output so we will discard it.
@@ -61,7 +74,7 @@ impl<'env> Scope<'env> {
         }
         .boxed();
 
-        self.inbox.lock().push(wrapped);
+        self.push(wrapped);
 
         ScopedJoinHandle { rx }
     }
@@ -71,21 +84,21 @@ impl<'env> Scope<'env> {
     /// This mirrors [`Executor::spawn_abortable`] for cooperatively
     /// scheduled tasks. The returned handle aborts the task when all
     /// references to it have been dropped.
-    pub fn spawn_abortable<Fut>(&self, fut: Fut) -> AbortableJoinHandle<Fut::Output>
+    pub fn spawn_abortable<Fut>(&'scope self, fut: Fut) -> AbortableJoinHandle<Fut::Output>
     where
-        Fut: Future + Send + 'env,
-        Fut::Output: Send + 'env,
+        Fut: Future + Send + 'scope,
+        Fut::Output: Send + 'scope,
     {
         let (abort_handle, abort_reg) = AbortHandle::new_pair();
         let abortable = Abortable::new(fut, abort_reg);
         let (tx, rx) = oneshot::channel();
 
-        let wrapped: BoxFuture<'env, ()> = async move {
+        let wrapped: BoxFuture<'scope, ()> = async move {
             let val = abortable.await;
             let _ = tx.send(val);
         }
         .boxed();
-        self.inbox.lock().push(wrapped);
+        self.push(wrapped);
 
         let join = JoinHandle {
             inner: InnerJoinHandle::CustomHandle {
@@ -99,21 +112,22 @@ impl<'env> Scope<'env> {
     /// Spawn a task into this scope without keeping a handle to it.
     ///
     /// Equivalent to [`Executor::dispatch`] for scoped tasks.
-    pub fn dispatch<Fut>(&self, fut: Fut)
+    pub fn dispatch<Fut>(&'scope self, fut: Fut)
     where
-        Fut: Future + Send + 'env,
-        Fut::Output: Send + 'env,
+        Fut: Future + Send + 'scope,
+        Fut::Output: Send + 'scope,
     {
-        let _ = self.spawn(fut);
+        drop(self.spawn(fut));
     }
 
     /// Spawn a message-driven coroutine into this scope.
     ///
     /// Equivalent to [`Executor::spawn_coroutine`] for scoped tasks.
-    pub fn spawn_coroutine<T, F, Fut>(&self, f: F) -> CommunicationTask<T>
+    pub fn spawn_coroutine<T, F, Fut>(&'scope self, f: F) -> CommunicationTask<T>
     where
-        F: FnMut(Receiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        F: FnMut(T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        T: Send + 'scope,
     {
         self.spawn_coroutine_with_buffer(1, f)
     }
@@ -121,107 +135,217 @@ impl<'env> Scope<'env> {
     /// Like [`Scope::spawn_coroutine`] but with a configurable channel
     /// buffer.
     pub fn spawn_coroutine_with_buffer<T, F, Fut>(
-        &self,
+        &'scope self,
+        buffer: usize,
+        mut f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        T: Send + 'scope,
+    {
+        let (tx, mut rx) = futures::channel::mpsc::channel(buffer);
+        let task_handle = self.spawn_abortable(async move {
+            while let Some(message) = rx.next().await {
+                f(message).await;
+            }
+        });
+        CommunicationTask::new(task_handle, tx)
+    }
+
+    /// Spawn an unbounded message-driven coroutine into this scope.
+    ///
+    /// Equivalent to [`Executor::spawn_unbounded_coroutine`] for scoped
+    /// tasks.
+    pub fn spawn_unbounded_coroutine<T, F, Fut>(
+        &'scope self,
+        mut f: F,
+    ) -> UnboundedCommunicationTask<T>
+    where
+        F: FnMut(T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        T: Send + 'scope,
+    {
+        let (tx, mut rx) = futures::channel::mpsc::unbounded();
+        let task_handle = self.spawn_abortable(async move {
+            while let Some(message) = rx.next().await {
+                f(message).await;
+            }
+        });
+        UnboundedCommunicationTask::new(task_handle, tx)
+    }
+
+    /// Spawn a message-driven coroutine with caller-provided context.
+    ///
+    /// If the context must be borrowed across awaits, use
+    /// [`Scope::spawn_coroutine_with_receiver_and_context`].
+    pub fn spawn_coroutine_with_context<T, C, F, Fut>(
+        &'scope self,
+        context: C,
+        f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(&mut C, T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        C: Send + 'scope,
+        T: Send + 'scope,
+    {
+        self.spawn_coroutine_with_buffer_and_context(context, 1, f)
+    }
+
+    /// Like [`Scope::spawn_coroutine_with_context`] but with a configurable
+    /// channel buffer.
+    pub fn spawn_coroutine_with_buffer_and_context<T, C, F, Fut>(
+        &'scope self,
+        context: C,
+        buffer: usize,
+        mut f: F,
+    ) -> CommunicationTask<T>
+    where
+        F: FnMut(&mut C, T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        C: Send + 'scope,
+        T: Send + 'scope,
+    {
+        let (tx, mut rx) = futures::channel::mpsc::channel(buffer);
+        let task_handle = self.spawn_abortable(async move {
+            let mut context = context;
+            while let Some(message) = rx.next().await {
+                f(&mut context, message).await;
+            }
+        });
+        CommunicationTask::new(task_handle, tx)
+    }
+
+    /// Spawn an unbounded message-driven coroutine with caller-provided
+    /// context.
+    pub fn spawn_unbounded_coroutine_with_context<T, C, F, Fut>(
+        &'scope self,
+        context: C,
+        mut f: F,
+    ) -> UnboundedCommunicationTask<T>
+    where
+        F: FnMut(&mut C, T) -> Fut + Send + 'scope,
+        Fut: Future<Output = ()> + Send + 'scope,
+        C: Send + 'scope,
+        T: Send + 'scope,
+    {
+        let (tx, mut rx) = futures::channel::mpsc::unbounded();
+        let task_handle = self.spawn_abortable(async move {
+            let mut context = context;
+            while let Some(message) = rx.next().await {
+                f(&mut context, message).await;
+            }
+        });
+        UnboundedCommunicationTask::new(task_handle, tx)
+    }
+
+    /// Spawn a coroutine that receives the bounded channel directly.
+    pub fn spawn_coroutine_with_receiver<T, F, Fut>(&'scope self, f: F) -> CommunicationTask<T>
+    where
+        F: FnMut(Receiver<T>) -> Fut,
+        Fut: Future<Output = ()> + Send + 'scope,
+    {
+        self.spawn_coroutine_with_receiver_and_buffer(1, f)
+    }
+
+    /// Like [`Scope::spawn_coroutine_with_receiver`] but with a configurable
+    /// channel buffer.
+    pub fn spawn_coroutine_with_receiver_and_buffer<T, F, Fut>(
+        &'scope self,
         buffer: usize,
         mut f: F,
     ) -> CommunicationTask<T>
     where
         F: FnMut(Receiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        Fut: Future<Output = ()> + Send + 'scope,
     {
         let (tx, rx) = futures::channel::mpsc::channel(buffer);
         let task_handle = self.spawn_abortable(f(rx));
         CommunicationTask::new(task_handle, tx)
     }
 
-    /// Like [`Scope::spawn_coroutine`] but passes a caller-provided
-    /// context into the coroutine alongside the message receiver.
-    pub fn spawn_coroutine_with_context<T, F, C, Fut>(
-        &self,
+    /// Spawn a coroutine that receives caller-provided context and the
+    /// bounded channel directly.
+    pub fn spawn_coroutine_with_receiver_and_context<T, F, C, Fut>(
+        &'scope self,
         context: C,
         f: F,
     ) -> CommunicationTask<T>
     where
         F: FnMut(C, Receiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        Fut: Future<Output = ()> + Send + 'scope,
     {
-        self.spawn_coroutine_with_buffer_and_context(context, 1, f)
+        self.spawn_coroutine_with_receiver_buffer_and_context(context, 1, f)
     }
 
-    /// Like [`Scope::spawn_coroutine_with_context`] but with a
+    /// Like [`Scope::spawn_coroutine_with_receiver_and_context`] but with a
     /// configurable channel buffer.
-    pub fn spawn_coroutine_with_buffer_and_context<T, F, C, Fut>(
-        &self,
+    pub fn spawn_coroutine_with_receiver_buffer_and_context<T, F, C, Fut>(
+        &'scope self,
         context: C,
         buffer: usize,
         mut f: F,
     ) -> CommunicationTask<T>
     where
         F: FnMut(C, Receiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        Fut: Future<Output = ()> + Send + 'scope,
     {
         let (tx, rx) = futures::channel::mpsc::channel(buffer);
         let task_handle = self.spawn_abortable(f(context, rx));
         CommunicationTask::new(task_handle, tx)
     }
 
-    /// Spawn an unbounded message-driven coroutine into this scope.
-    ///
-    /// Equivalent to [`Executor::spawn_unbounded_coroutine`] for scoped
-    /// tasks.
-    pub fn spawn_unbounded_coroutine<T, F, Fut>(
-        &self,
+    /// Spawn a coroutine that receives the unbounded channel directly.
+    pub fn spawn_unbounded_coroutine_with_receiver<T, F, Fut>(
+        &'scope self,
         mut f: F,
     ) -> UnboundedCommunicationTask<T>
     where
         F: FnMut(UnboundedReceiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        Fut: Future<Output = ()> + Send + 'scope,
     {
         let (tx, rx) = futures::channel::mpsc::unbounded();
         let task_handle = self.spawn_abortable(f(rx));
         UnboundedCommunicationTask::new(task_handle, tx)
     }
 
-    /// Like [`Scope::spawn_unbounded_coroutine`] but passes a
-    /// caller-provided context into the coroutine.
-    pub fn spawn_unbounded_coroutine_with_context<T, F, C, Fut>(
-        &self,
+    /// Spawn a coroutine that receives caller-provided context and the
+    /// unbounded channel directly.
+    pub fn spawn_unbounded_coroutine_with_receiver_and_context<T, F, C, Fut>(
+        &'scope self,
         context: C,
         mut f: F,
     ) -> UnboundedCommunicationTask<T>
     where
         F: FnMut(C, UnboundedReceiver<T>) -> Fut,
-        Fut: Future<Output = ()> + Send + 'env,
+        Fut: Future<Output = ()> + Send + 'scope,
     {
         let (tx, rx) = futures::channel::mpsc::unbounded();
         let task_handle = self.spawn_abortable(f(context, rx));
         UnboundedCommunicationTask::new(task_handle, tx)
     }
-
 }
 
 /// Drive a scope once: absorb any inbox entries, then poll the active
 /// set until no more immediate progress can be made.
-fn drive_scope<'env>(
-    active: &mut FuturesUnordered<BoxFuture<'env, ()>>,
-    scope: &Scope<'env>,
+fn drive_scope<'scope>(
+    active: &mut FuturesUnordered<BoxFuture<'scope, ()>>,
+    state: &ScopeState<'scope>,
     cx: &mut Context<'_>,
 ) -> (bool, bool) {
     let mut made_progress = false;
     loop {
-        // Absorb any newly-spawned tasks into the active set. Take the
-        // Vec out with mem::take so the lock is held just long enough
-        // to swap, never across a poll.
-        let incoming = std::mem::take(&mut *scope.inbox.lock());
+        // Take the queued tasks while holding the lock only long enough to
+        // swap the inbox, never while polling user code.
+        let incoming = std::mem::take(&mut *state.inbox.lock());
         active.extend(incoming);
         match active.poll_next_unpin(cx) {
             Poll::Ready(Some(())) => made_progress = true,
             Poll::Ready(None) => return (made_progress, true),
             Poll::Pending => {
-                // A task may have pushed new futures to the inbox
-                // during its own poll, so it doesn't return Pending until
-                // we've had a chance to poll them.
-                if scope.inbox.lock().is_empty() {
+                // A child may have spawned another task during its poll.
+                if state.inbox.lock().is_empty() {
                     return (made_progress, false);
                 }
             }
@@ -235,9 +359,9 @@ fn drive_scope<'env>(
 /// before the task finishes (for example, because the scope future was
 /// cancelled), awaiting yields [`JoinError::Cancelled`].
 ///
-/// # Note
-/// The handle is `'static` so it can  be moved into other spawned tasks, channels, or futures
-/// without lifetime gymnastics.
+/// The handle does not borrow the spawned future itself. It can therefore
+/// outlive the scope when its output type is also `'static`; borrowed output
+/// types retain their normal lifetime restrictions.
 pub struct ScopedJoinHandle<T> {
     rx: oneshot::Receiver<T>,
 }
@@ -298,13 +422,19 @@ impl<T> Future for ScopedJoinHandle<T> {
 /// ```
 pub async fn scope<'env, F, T>(f: F) -> T
 where
-    F: AsyncFnOnce(&Scope<'env>) -> T,
+    F: for<'scope> AsyncFnOnce(&'scope Scope<'scope, 'env>) -> T,
 {
-    let scope: Scope<'env> = Scope::new();
+    // Declaration order is part of the safety invariant: `active` is dropped
+    // first, then `state` (and its queued tasks), and finally `scope`.
+    let mut scope = Scope::new();
+    let state = Arc::new(ScopeState {
+        inbox: Mutex::new(Vec::new()),
+    });
+    scope.state = Arc::downgrade(&state);
     // The active task set lives on the driver, not inside `Scope`. Only
     // `drive_scope` touches it, so no lock guards it, and the inbox mutex
     // is enough for the spawn side.
-    let mut active: FuturesUnordered<BoxFuture<'env, ()>> = FuturesUnordered::new();
+    let mut active = FuturesUnordered::new();
 
     let result = {
         let user_fut = f(&scope);
@@ -315,7 +445,7 @@ where
                 if let Poll::Ready(r) = user_fut.as_mut().poll(cx) {
                     return Poll::Ready(r);
                 }
-                let (made_progress, _empty) = drive_scope(&mut active, &scope, cx);
+                let (made_progress, _empty) = drive_scope(&mut active, &state, cx);
                 if !made_progress {
                     return Poll::Pending;
                 }
@@ -327,7 +457,7 @@ where
     // Drain any remaining spawned tasks before the scope goes away, so
     // every borrow of the data is released before this frame unwinds.
     poll_fn(|cx| {
-        let (_made_progress, empty) = drive_scope(&mut active, &scope, cx);
+        let (_made_progress, empty) = drive_scope(&mut active, &state, cx);
         if empty {
             Poll::Ready(())
         } else {
@@ -351,7 +481,7 @@ where
 /// from the enclosing stack frame. Use [`scope`] for that.
 pub struct ScopeExecutor<'scope, E> {
     inner: &'scope E,
-    task_handles: Mutex<Vec<JoinHandle<()>>>,
+    task_handles: Mutex<Vec<AbortableJoinHandle<()>>>,
     _scope: PhantomData<&'scope mut &'scope ()>,
 }
 
@@ -395,7 +525,9 @@ where
             let _ = tx.send(val);
         };
 
-        let task_handle = self.inner.spawn(wrapped);
+        // Track an abort-on-drop handle so cancellation remains effective even
+        // after the handles are drained from `ScopeExecutor` for joining.
+        let task_handle: AbortableJoinHandle<()> = self.inner.spawn(wrapped).into();
         self.task_handles.lock().push(task_handle);
 
         JoinHandle {
@@ -465,15 +597,17 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::time::Duration;
-    use futures_timer::Delay;
     use super::*;
+    #[cfg(feature = "tokio")]
+    use futures_timer::Delay;
+    #[cfg(feature = "tokio")]
+    use std::time::Duration;
 
     #[tokio::test]
     async fn borrows_stack_data() {
         let data = vec![1, 2, 3, 4];
         let data = &data;
-        let sum = scope(async |s: &Scope<'_>| {
+        let sum = scope(async |s: &Scope<'_, '_>| {
             let a = s.spawn(async move { data[0] + data[1] });
             let b = s.spawn(async move { data[2] + data[3] });
             a.await.unwrap() + b.await.unwrap()
@@ -487,7 +621,7 @@ mod tests {
         use std::sync::atomic::{AtomicUsize, Ordering};
         let counter = AtomicUsize::new(0);
         let counter_ref = &counter;
-        scope(async |s: &Scope<'_>| {
+        scope(async |s: &Scope<'_, '_>| {
             for _ in 0..8 {
                 s.spawn(async move {
                     counter_ref.fetch_add(1, Ordering::SeqCst);
@@ -500,13 +634,13 @@ mod tests {
 
     #[tokio::test]
     async fn returns_closure_value() {
-        let v: i32 = scope(async |_s: &Scope<'_>| 42).await;
+        let v: i32 = scope(async |_s: &Scope<'_, '_>| 42).await;
         assert_eq!(v, 42);
     }
 
     #[tokio::test]
     async fn join_handle_yields_output() {
-        let out = scope(async |s: &Scope<'_>| {
+        let out = scope(async |s: &Scope<'_, '_>| {
             let h = s.spawn(async { "hello" });
             h.await.unwrap()
         })
@@ -519,7 +653,7 @@ mod tests {
         use std::sync::atomic::{AtomicUsize, Ordering};
         let counter = AtomicUsize::new(0);
         let counter_ref = &counter;
-        let total: usize = scope(async |s: &Scope<'_>| {
+        let total: usize = scope(async |s: &Scope<'_, '_>| {
             let handles: Vec<_> = (0..32)
                 .map(|i| {
                     s.spawn(async move {
@@ -539,6 +673,40 @@ mod tests {
         assert_eq!(counter.load(Ordering::SeqCst), 32);
     }
 
+    #[tokio::test]
+    async fn child_task_can_spawn_nested_task() {
+        let result = scope(async |s: &Scope<'_, '_>| {
+            let outer = s.spawn(async move {
+                let inner = s.spawn(async { 41usize });
+                inner.await.unwrap() + 1
+            });
+            outer.await.unwrap()
+        })
+        .await;
+
+        assert_eq!(result, 42);
+    }
+
+    #[tokio::test]
+    async fn drains_unawaited_nested_task() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        let nested_ran = AtomicBool::new(false);
+        let nested_ran_ref = &nested_ran;
+
+        scope(async |s: &Scope<'_, '_>| {
+            s.dispatch(async move {
+                s.dispatch(async move {
+                    nested_ran_ref.store(true, Ordering::SeqCst);
+                });
+            });
+        })
+        .await;
+
+        assert!(nested_ran.load(Ordering::SeqCst));
+    }
+
+    #[cfg(feature = "tokio")]
     #[tokio::test]
     async fn executor_scope_runs_tasks() {
         use crate::rt::tokio::TokioExecutor;
@@ -559,11 +727,9 @@ mod tests {
         let total = AtomicUsize::new(0);
         let total_ref = &total;
 
-        scope(async |s: &Scope<'_>| {
-            let mut task = s.spawn_coroutine(|mut rx| async move {
-                while let Some(value) = rx.next().await {
-                    total_ref.fetch_add(value, Ordering::SeqCst);
-                }
+        scope(async |s: &Scope<'_, '_>| {
+            let mut task = s.spawn_coroutine(|value| async move {
+                total_ref.fetch_add(value, Ordering::SeqCst);
             });
             for v in [1usize, 2, 3, 4] {
                 task.send(v).await.unwrap();
@@ -575,13 +741,102 @@ mod tests {
         assert_eq!(total.load(Ordering::SeqCst), 10);
     }
 
+    #[tokio::test]
+    async fn scope_receiver_coroutine_receives_messages() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let total = AtomicUsize::new(0);
+        let total_ref = &total;
+
+        scope(async |s: &Scope<'_, '_>| {
+            let mut task = s.spawn_coroutine_with_receiver(|mut rx| async move {
+                while let Some(value) = rx.next().await {
+                    total_ref.fetch_add(value, Ordering::SeqCst);
+                }
+            });
+            for value in [1usize, 2, 3, 4] {
+                task.send(value).await.unwrap();
+            }
+            drop(task);
+        })
+        .await;
+
+        assert_eq!(total.load(Ordering::SeqCst), 10);
+    }
+
+    #[tokio::test]
+    async fn scope_coroutine_api_matches_executor() {
+        use futures::future::ready;
+
+        scope(async |s: &Scope<'_, '_>| {
+            let task = s.spawn_coroutine_with_buffer(2, |_value: usize| ready(()));
+            drop(task);
+
+            let task = s.spawn_unbounded_coroutine(|_value: usize| ready(()));
+            drop(task);
+
+            let task =
+                s.spawn_coroutine_with_context(0usize, |context: &mut usize, value: usize| {
+                    *context += value;
+                    ready(())
+                });
+            drop(task);
+
+            let task = s.spawn_coroutine_with_buffer_and_context(
+                0usize,
+                2,
+                |context: &mut usize, value: usize| {
+                    *context += value;
+                    ready(())
+                },
+            );
+            drop(task);
+
+            let task = s.spawn_unbounded_coroutine_with_context(
+                0usize,
+                |context: &mut usize, value: usize| {
+                    *context += value;
+                    ready(())
+                },
+            );
+            drop(task);
+
+            let task =
+                s.spawn_coroutine_with_receiver_and_buffer(2, |_rx: Receiver<usize>| async {});
+            drop(task);
+
+            let task = s.spawn_coroutine_with_receiver_and_context(
+                0usize,
+                |_context, _rx: Receiver<usize>| async {},
+            );
+            drop(task);
+
+            let task = s.spawn_coroutine_with_receiver_buffer_and_context(
+                0usize,
+                2,
+                |_context, _rx: Receiver<usize>| async {},
+            );
+            drop(task);
+
+            let task =
+                s.spawn_unbounded_coroutine_with_receiver(|_rx: UnboundedReceiver<usize>| async {});
+            drop(task);
+
+            let task = s.spawn_unbounded_coroutine_with_receiver_and_context(
+                0usize,
+                |_context, _rx: UnboundedReceiver<usize>| async {},
+            );
+            drop(task);
+        })
+        .await;
+    }
+
     #[tokio::test]
     async fn scope_dispatch_runs_fire_and_forget() {
         use std::sync::atomic::{AtomicBool, Ordering};
         let flag = AtomicBool::new(false);
         let flag_ref = &flag;
 
-        scope(async |s: &Scope<'_>| {
+        scope(async |s: &Scope<'_, '_>| {
             s.dispatch(async move {
                 flag_ref.store(true, Ordering::SeqCst);
             });
@@ -591,6 +846,7 @@ mod tests {
         assert!(flag.load(Ordering::SeqCst));
     }
 
+    #[cfg(feature = "tokio")]
     #[tokio::test]
     async fn executor_scope_drains_unawaited_tasks() {
         use crate::rt::tokio::TokioExecutor;
@@ -619,6 +875,7 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "tokio")]
     #[tokio::test]
     async fn executor_scope_swallows_task_panic() {
         use crate::rt::tokio::TokioExecutor;
@@ -649,6 +906,7 @@ mod tests {
         assert_eq!(sibling_done.load(Ordering::SeqCst), 1);
     }
 
+    #[cfg(feature = "tokio")]
     #[tokio::test]
     async fn executor_scope_aborts_on_external_cancel() {
         use crate::rt::tokio::TokioExecutor;
@@ -685,4 +943,37 @@ mod tests {
             "task should have been aborted by scope drop"
         );
     }
+
+    #[cfg(feature = "tokio")]
+    #[tokio::test]
+    async fn executor_scope_aborts_when_cancelled_during_drain() {
+        use crate::rt::tokio::TokioExecutor;
+        use futures::future::{Either, pending, select};
+
+        let executor = TokioExecutor;
+        let (started_tx, started_rx) = oneshot::channel();
+        let (held_tx, held_rx) = oneshot::channel::<()>();
+
+        let scope_fut = Box::pin(executor.executor_scope(async move |s| {
+            let _handle = s.spawn(async move {
+                let _held_until_task_drop = held_tx;
+                let _ = started_tx.send(());
+                pending::<()>().await;
+            });
+        }));
+
+        let scope_fut = match select(scope_fut, started_rx).await {
+            Either::Right((Ok(()), scope_fut)) => scope_fut,
+            Either::Left(_) => panic!("scope unexpectedly completed"),
+            Either::Right((Err(_), _)) => panic!("child task never started"),
+        };
+
+        drop(scope_fut);
+
+        match select(held_rx, Delay::new(Duration::from_secs(1))).await {
+            Either::Left((Err(_), _)) => {}
+            Either::Left((Ok(_), _)) => unreachable!("child never sends a value"),
+            Either::Right(_) => panic!("child remained detached after scope cancellation"),
+        }
+    }
 }

From 487d3b4a12d1746110a2398ea2e88799270ee511 Mon Sep 17 00:00:00 2001
From: Darius Clark <darius.clark@proton.me>
Date: Wed, 22 Jul 2026 23:01:08 -0500
Subject: [PATCH 5/5] fix: add completion guards and waker

---
 src/lib.rs    |   5 +-
 src/scoped.rs | 148 +++++++++++++++++++++++++++++++++++++++++++++++---
 2 files changed, 143 insertions(+), 10 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index bba5113..41146ac 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -652,9 +652,8 @@ pub trait Executor {
     /// that borrow from the enclosing stack frame.
     ///
     /// Unlike [`Executor::spawn`], tasks spawned on the [`Scope`] are driven
-    /// cooperatively by the returned future — they make progress whenever
-    /// the scope is polled — so they may borrow any data that outlives the
-    /// `'env` lifetime. Every task is either completed or cancelled before
+    /// cooperatively by the returned future so they may borrow any data that outlives
+    /// the `'env` lifetime. Every task is either completed or canceled before
     /// `scope` returns, so borrows never outlive the stack frame.
     ///
     /// This is the async analogue of [`std::thread::scope`].
diff --git a/src/scoped.rs b/src/scoped.rs
index ea41449..65d3935 100644
--- a/src/scoped.rs
+++ b/src/scoped.rs
@@ -13,7 +13,7 @@
 //! borrow is released before the stack frame goes away.
 
 use crate::{
-    AbortableJoinHandle, CommunicationTask, Executor, InnerJoinHandle, JoinHandle,
+    AbortableJoinHandle, CommunicationTask, CompletionGuard, Executor, InnerJoinHandle, JoinHandle,
     UnboundedCommunicationTask,
 };
 use core::future::{Future, poll_fn};
@@ -24,14 +24,16 @@ use futures::channel::mpsc::{Receiver, UnboundedReceiver};
 use futures::channel::oneshot;
 use futures::future::{AbortHandle, Abortable, BoxFuture};
 use futures::stream::FuturesUnordered;
+use futures::task::AtomicWaker;
 use futures::{FutureExt, StreamExt};
 use parking_lot::Mutex;
-use std::sync::{Arc, Weak};
-use std::sync::atomic::AtomicBool;
 use pollable_map::optional::Optional;
+use std::sync::atomic::AtomicBool;
+use std::sync::{Arc, Weak};
 
 struct ScopeState<'scope> {
     inbox: Mutex<Vec<BoxFuture<'scope, ()>>>,
+    waker: AtomicWaker,
 }
 
 /// A scope within which tasks can be spawned that borrow from the enclosing
@@ -54,6 +56,7 @@ impl<'scope, 'env> Scope<'scope, 'env> {
     fn push(&self, task: BoxFuture<'scope, ()>) {
         if let Some(state) = self.state.upgrade() {
             state.inbox.lock().push(task);
+            state.waker.wake();
         }
     }
 
@@ -94,8 +97,11 @@ impl<'scope, 'env> Scope<'scope, 'env> {
         let (abort_handle, abort_reg) = AbortHandle::new_pair();
         let abortable = Abortable::new(fut, abort_reg);
         let (tx, rx) = oneshot::channel();
+        let finished = Arc::new(AtomicBool::new(false));
+        let completion = CompletionGuard::new(finished.clone());
 
         let wrapped: BoxFuture<'scope, ()> = async move {
+            let _completion = completion;
             let val = abortable.await;
             let _ = tx.send(val);
         }
@@ -106,7 +112,7 @@ impl<'scope, 'env> Scope<'scope, 'env> {
             inner: InnerJoinHandle::CustomHandle {
                 inner: Optional::new(rx),
                 handle: abort_handle,
-                finished: Arc::new(AtomicBool::new(false)),
+                finished,
             },
         };
         AbortableJoinHandle::from(join)
@@ -339,15 +345,22 @@ fn drive_scope<'scope>(
 ) -> (bool, bool) {
     let mut made_progress = false;
     loop {
+        state.waker.register(cx.waker());
+
         // Take the queued tasks while holding the lock only long enough to
         // swap the inbox, never while polling user code.
         let incoming = std::mem::take(&mut *state.inbox.lock());
         active.extend(incoming);
         match active.poll_next_unpin(cx) {
             Poll::Ready(Some(())) => made_progress = true,
-            Poll::Ready(None) => return (made_progress, true),
+            Poll::Ready(None) => {
+                state.waker.register(cx.waker());
+                if state.inbox.lock().is_empty() {
+                    return (made_progress, true);
+                }
+            }
             Poll::Pending => {
-                // A child may have spawned another task during its poll.
+                state.waker.register(cx.waker());
                 if state.inbox.lock().is_empty() {
                     return (made_progress, false);
                 }
@@ -432,6 +445,7 @@ where
     let mut scope = Scope::new();
     let state = Arc::new(ScopeState {
         inbox: Mutex::new(Vec::new()),
+        waker: AtomicWaker::new(),
     });
     scope.state = Arc::downgrade(&state);
     // The active task set lives on the driver, not inside `Scope`. Only
@@ -523,7 +537,10 @@ where
         let (abort_handle, abort_registration) = AbortHandle::new_pair();
         let abortable = Abortable::new(future, abort_registration);
         let (tx, rx) = oneshot::channel();
+        let finished = Arc::new(AtomicBool::new(false));
+        let completion = CompletionGuard::new(finished.clone());
         let wrapped = async move {
+            let _completion = completion;
             let val = abortable.await;
             let _ = tx.send(val);
         };
@@ -537,7 +554,7 @@ where
             inner: InnerJoinHandle::CustomHandle {
                 inner: Optional::new(rx),
                 handle: abort_handle,
-                finished: Arc::new(AtomicBool::new(false)),
+                finished,
             },
         }
     }
@@ -652,6 +669,101 @@ mod tests {
         assert_eq!(out, "hello");
     }
 
+    #[tokio::test]
+    async fn abortable_handle_reports_completion_without_being_polled() {
+        scope(async |s: &Scope<'_, '_>| {
+            let handle = s.spawn_abortable(async {});
+
+            // Yield the user future so the cooperative driver can complete
+            // the child without polling its join handle.
+            crate::task::yield_now().await;
+
+            assert!(handle.is_finished());
+        })
+        .await;
+    }
+
+    #[test]
+    fn pushing_task_wakes_scope_driver() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use std::task::{Wake, Waker};
+
+        struct WakeCounter(AtomicUsize);
+
+        impl Wake for WakeCounter {
+            fn wake(self: Arc<Self>) {
+                self.0.fetch_add(1, Ordering::SeqCst);
+            }
+
+            fn wake_by_ref(self: &Arc<Self>) {
+                self.0.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+
+        let mut scope = Scope::new();
+        let state = Arc::new(ScopeState {
+            inbox: Mutex::new(Vec::new()),
+            waker: AtomicWaker::new(),
+        });
+        scope.state = Arc::downgrade(&state);
+
+        let wake_counter = Arc::new(WakeCounter(AtomicUsize::new(0)));
+        let waker = Waker::from(wake_counter.clone());
+        state.waker.register(&waker);
+
+        let scope_ref = &scope;
+        scope.push(
+            async move {
+                // Consume the driver's registered waker while tasks are
+                // being polled, forcing drive_scope to register it again.
+                scope_ref.push(async {}.boxed());
+            }
+            .boxed(),
+        );
+
+        assert_eq!(state.inbox.lock().len(), 1);
+        assert_eq!(wake_counter.0.load(Ordering::SeqCst), 1);
+
+        let mut active = FuturesUnordered::new();
+        let mut context = Context::from_waker(&waker);
+        let (_, empty) = drive_scope(&mut active, &state, &mut context);
+        assert!(empty);
+        assert_eq!(wake_counter.0.load(Ordering::SeqCst), 2);
+
+        // The nested push consumed the previous registration. This push
+        // only wakes the driver if drive_scope registered again afterward.
+        scope.push(async {}.boxed());
+        assert_eq!(wake_counter.0.load(Ordering::SeqCst), 3);
+
+        let (_, empty) = drive_scope(&mut active, &state, &mut context);
+        assert!(empty);
+
+        let state_ref = Arc::downgrade(&state);
+        scope.push(
+            poll_fn(move |_cx| {
+                // Model a push that occurs after registration but before
+                // the inbox is drained: its wake is consumed, then the
+                // newly active task returns Pending.
+                state_ref.upgrade().unwrap().waker.wake();
+                Poll::<()>::Pending
+            })
+            .boxed(),
+        );
+        assert_eq!(wake_counter.0.load(Ordering::SeqCst), 4);
+
+        let (_, empty) = drive_scope(&mut active, &state, &mut context);
+        assert!(!empty);
+        let wakes_after_pending = wake_counter.0.load(Ordering::SeqCst);
+        assert!(wakes_after_pending > 4);
+
+        // The Pending path must have registered once more before returning.
+        scope.push(async {}.boxed());
+        assert_eq!(
+            wake_counter.0.load(Ordering::SeqCst),
+            wakes_after_pending + 1
+        );
+    }
+
     #[tokio::test]
     async fn many_concurrent_tasks_complete() {
         use std::sync::atomic::{AtomicUsize, Ordering};
@@ -725,6 +837,28 @@ mod tests {
         assert_eq!(total, 10);
     }
 
+    #[cfg(feature = "tokio")]
+    #[tokio::test(flavor = "current_thread")]
+    async fn executor_scope_handle_reports_completion_without_being_polled() {
+        use crate::rt::tokio::TokioExecutor;
+
+        let executor = TokioExecutor;
+        executor
+            .executor_scope(async |s| {
+                let (completed_tx, completed_rx) = oneshot::channel();
+                let handle = s.spawn(async move {
+                    let _ = completed_tx.send(());
+                });
+
+                completed_rx.await.unwrap();
+
+                // On the current-thread runtime, the spawned wrapper finishes
+                // before the task it woke can be polled again.
+                assert!(handle.is_finished());
+            })
+            .await;
+    }
+
     #[tokio::test]
     async fn scope_spawn_coroutine_receives_messages() {
         use std::sync::atomic::{AtomicUsize, Ordering};