Skip to main content

asupersync/cx/
scope.rs

1//! Scope API for spawning work within a region.
2//!
3//! A `Scope` provides the API for spawning tasks, creating child regions,
4//! and registering finalizers.
5//!
6//! # Execution Tiers and Soundness Rules
7//!
8//! Asupersync defines two execution tiers with different constraints:
9//!
10//! ## Fiber Tier (Phase 0)
11//!
12//! - Single-thread, borrow-friendly execution
13//! - Can capture borrowed references (`&T`) since no migration
14//! - Implemented via `spawn_local` (currently requires Send bounds; relaxed in Phase 1+)
15//!
16//! ## Task Tier (Phase 1+)
17//!
18//! - Multi-threaded, `Send` tasks that may migrate across workers
19//! - **Must capture only `Send + 'static` data** by construction
20//! - Can reference region-owned data via [`RRef<T>`](crate::types::rref::RRef)
21//!
22//! # Soundness Rules for Send Tasks
23//!
24//! The [`spawn`](Scope::spawn) method enforces the following bounds:
25//!
26//! | Component | Bound | Rationale |
27//! |-----------|-------|-----------|
28//! | Factory | `F: Send + 'static` | Factory may be called on any worker |
29//! | Future | `Fut: Send + 'static` | Task may migrate between polls |
30//! | Output | `Fut::Output: Send + 'static` | Result sent to potentially different thread |
31//!
32//! ## What Can Be Captured
33//!
34//! **Allowed captures in Send tasks:**
35//! - Owned `'static` data that is `Send` (e.g., `String`, `Vec<T>`, `Arc<T>`)
36//! - [`RRef<T>`](crate::types::rref::RRef) handles to region-heap-allocated data
37//! - Atomic types (`AtomicU64`, etc.)
38//! - Clone'd `Cx` (the capability context)
39//!
40//! **Disallowed captures:**
41//! - Borrowed references (`&T`, `&mut T`) - not `'static`
42//! - `Rc<T>`, `RefCell<T>` - not `Send`
43//! - Raw pointers (unless wrapped in a `Send` type)
44//! - References to stack-local data
45//!
46//! ## RRef for Region-Owned Data
47//!
48//! When tasks need to share data within a region without cloning, use the region
49//! heap and [`RRef<T>`](crate::types::rref::RRef):
50//!
51//! ```ignore
52//! // Allocate in region heap
53//! let index = region.heap_alloc(expensive_data);
54//! let rref = RRef::<ExpensiveData>::new(region_id, index);
55//!
56//! // Pass RRef to task - it's Copy + Send
57//! scope.spawn(state, &cx, move |cx| async move {
58//!     // Access via region record (requires runtime lookup)
59//!     let data = rref.get_via_region(&region_record)?;
60//!     process(data).await
61//! });
62//! ```
63//!
64//! # Compile-Time Enforcement
65//!
66//! The bounds are enforced at compile time. Attempting to capture non-Send
67//! or non-static data will result in a compilation error:
68//!
69//! ```compile_fail
70//! use std::rc::Rc;
71//! use asupersync::cx::Scope;
72//!
73//! fn try_capture_rc(scope: &Scope, state: &mut RuntimeState, cx: &Cx) {
74//!     let rc = Rc::new(42); // Rc is !Send
75//!     scope.spawn(state, cx, move |_| async move {
76//!         println!("{}", rc); // ERROR: Rc is not Send
77//!     });
78//! }
79//! ```
80//!
81//! ```compile_fail
82//! use asupersync::cx::Scope;
83//!
84//! fn try_capture_borrow(scope: &Scope, state: &mut RuntimeState, cx: &Cx) {
85//!     let local = 42;
86//!     let reference = &local; // Borrowed, not 'static
87//!     scope.spawn(state, cx, move |_| async move {
88//!         println!("{}", reference); // ERROR: borrowed data not 'static
89//!     });
90//! }
91//! ```
92//!
93//! # Lab Runtime Compatibility
94//!
95//! The Send bounds do not affect lab runtime determinism. The lab runtime
96//! simulates multi-worker scheduling deterministically (same seed = same
97//! execution), regardless of whether tasks are actually migrated.
98
99use crate::channel::oneshot;
100use crate::combinator::{Either, Select};
101use crate::cx::{Cx, cap};
102use crate::record::AdmissionError;
103use crate::record::task::TaskState;
104use crate::runtime::resource_monitor::RegionPriority;
105use crate::runtime::task_handle::{JoinError, TaskHandle};
106use crate::runtime::{RegionCreateError, RuntimeState, SpawnError, StoredTask};
107use crate::tracing_compat::{debug, debug_span};
108use crate::types::{
109    Budget, CancelReason, CapabilityBudget, CapabilityBudgetRequirements, Outcome, PanicPayload,
110    Policy, RegionId, TaskId,
111};
112use std::future::Future;
113use std::marker::PhantomData;
114use std::pin::Pin;
115use std::sync::Arc;
116use std::task::{Context, Poll};
117
118/// A scope for spawning work within a region.
119///
120/// The scope provides methods for:
121/// - Spawning tasks
122/// - Creating child regions
123/// - Registering finalizers
124/// - Cancelling all children
125pub struct Scope<'r, P: Policy = crate::types::policy::FailFast> {
126    /// The region this scope belongs to.
127    pub(crate) region: RegionId,
128    /// The budget for this scope.
129    pub(crate) budget: Budget,
130    /// The capability/resource budget for this scope.
131    pub(crate) capability_budget: CapabilityBudget,
132    /// Phantom data for the policy type.
133    pub(crate) _policy: PhantomData<&'r P>,
134}
135
136#[derive(Clone, Copy)]
137struct ChildRegionAdmission {
138    budget: Budget,
139    capability_budget: CapabilityBudget,
140    requirements: CapabilityBudgetRequirements,
141    priority: RegionPriority,
142}
143
144#[pin_project::pin_project]
145pub(crate) struct CatchUnwind<F> {
146    #[pin]
147    pub(crate) inner: F,
148}
149
150impl<F: Future> Future for CatchUnwind<F> {
151    type Output = std::thread::Result<F::Output>;
152
153    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
154        let mut this = self.project();
155        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
156            this.inner.as_mut().poll(cx)
157        }));
158        match result {
159            Ok(Poll::Pending) => Poll::Pending,
160            Ok(Poll::Ready(v)) => Poll::Ready(Ok(v)),
161            Err(payload) => Poll::Ready(Err(payload)),
162        }
163    }
164}
165
166pub(crate) fn payload_to_string(payload: &Box<dyn std::any::Any + Send>) -> String {
167    payload
168        .downcast_ref::<&str>()
169        .map(ToString::to_string)
170        .or_else(|| payload.downcast_ref::<String>().cloned())
171        .unwrap_or_else(|| "unknown panic".to_string())
172}
173
174struct RegionRunner<'a, Fut> {
175    fut: Pin<&'a mut CatchUnwind<Fut>>,
176    state: Option<&'a mut RuntimeState>,
177    child_region: RegionId,
178}
179
180impl<'a, Fut: Future> Future for RegionRunner<'a, Fut> {
181    type Output = (std::thread::Result<Fut::Output>, &'a mut RuntimeState);
182
183    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
184        let this = self.get_mut();
185        match this.fut.as_mut().poll(cx) {
186            Poll::Ready(res) => {
187                let state = this.state.take().expect("polled after ready");
188                Poll::Ready((res, state))
189            }
190            Poll::Pending => Poll::Pending,
191        }
192    }
193}
194
195impl<Fut> Drop for RegionRunner<'_, Fut> {
196    fn drop(&mut self) {
197        if let Some(state) = self.state.take() {
198            let reason = CancelReason::fail_fast().with_region(self.child_region);
199            let _ = state.cancel_request(self.child_region, &reason, None);
200            if let Some(region) = state.region(self.child_region) {
201                region.begin_close(None);
202            }
203            state.advance_region_state(self.child_region);
204        }
205    }
206}
207
208struct RegionCloseFuture {
209    state: Arc<parking_lot::Mutex<crate::record::region::RegionCloseState>>,
210}
211
212impl Future for RegionCloseFuture {
213    type Output = ();
214
215    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
216        let mut state = self.state.lock();
217        if state.closed {
218            Poll::Ready(())
219        } else {
220            if !state.waiters.iter().any(|w| w.will_wake(cx.waker())) {
221                state.waiters.push(cx.waker().clone());
222            }
223            Poll::Pending
224        }
225    }
226}
227
228impl<P: Policy> Scope<'_, P> {
229    /// Creates a new scope (internal use).
230    #[must_use]
231    #[allow(dead_code)]
232    #[cfg_attr(feature = "test-internals", visibility::make(pub))]
233    pub(crate) fn new(region: RegionId, budget: Budget) -> Self {
234        Self::new_with_capability_budget(region, budget, CapabilityBudget::UNSPECIFIED)
235    }
236
237    /// Creates a new scope with an explicit capability budget (internal use).
238    #[must_use]
239    #[allow(dead_code)]
240    #[cfg_attr(feature = "test-internals", visibility::make(pub))]
241    pub(crate) fn new_with_capability_budget(
242        region: RegionId,
243        budget: Budget,
244        capability_budget: CapabilityBudget,
245    ) -> Self {
246        Self {
247            region,
248            budget,
249            capability_budget,
250            _policy: PhantomData,
251        }
252    }
253
254    /// Returns the region ID for this scope.
255    #[must_use]
256    pub fn region_id(&self) -> RegionId {
257        self.region
258    }
259
260    /// Returns the budget for this scope.
261    #[must_use]
262    pub fn budget(&self) -> Budget {
263        self.budget
264    }
265
266    /// Returns the capability/resource budget for this scope.
267    #[must_use]
268    pub fn capability_budget(&self) -> CapabilityBudget {
269        self.capability_budget
270    }
271
272    // =========================================================================
273    // Task Spawning
274    // =========================================================================
275
276    /// Spawns a new task within this scope's region.
277    ///
278    /// This is the **Task Tier** spawn method for parallel execution. The task
279    /// may migrate between worker threads, so all captured data must be thread-safe.
280    ///
281    /// The task will be owned by the region and will be cancelled if the
282    /// region is cancelled. The returned `TaskHandle` can be used to await
283    /// the task's result.
284    ///
285    /// # Arguments
286    ///
287    /// * `state` - The runtime state
288    /// * `cx` - The capability context (used for tracing/authorization)
289    /// * `f` - A closure that produces the future, receiving the new task's `Cx`
290    ///
291    /// # Returns
292    ///
293    /// A `TaskHandle<T>` that can be used to await the task's result.
294    ///
295    /// # Soundness Rules (Type Bounds)
296    ///
297    /// The following bounds encode the soundness rules for Send tasks:
298    ///
299    /// * `F: FnOnce(Cx) -> Fut + Send + 'static` - Factory called on any worker
300    /// * `Fut: Future + Send + 'static` - Task may migrate between polls
301    /// * `Fut::Output: Send + 'static` - Result crosses thread boundary
302    ///
303    /// These bounds ensure captured data can safely cross thread boundaries.
304    /// Use [`RRef<T>`](crate::types::rref::RRef) for region-heap-allocated data.
305    ///
306    /// # Allowed Captures
307    ///
308    /// | Type | Allowed | Reason |
309    /// |------|---------|--------|
310    /// | `String`, `Vec<T>`, owned data | ✅ | Send + 'static by ownership |
311    /// | `Arc<T>` where T: Send + Sync | ✅ | Thread-safe shared ownership |
312    /// | `RRef<T>` | ✅ | Region-heap reference, Copy + Send |
313    /// | `Cx` (cloned) | ✅ | Capability context is Send + Sync |
314    /// | `Rc<T>`, `RefCell<T>` | ❌ | Not Send |
315    /// | `&T`, `&mut T` | ❌ | Not 'static |
316    ///
317    /// # Example
318    ///
319    /// ```ignore
320    /// let handle = scope.spawn(&mut state, &cx, |cx| async move {
321    ///     cx.trace("Child task running");
322    ///     compute_value().await
323    /// });
324    ///
325    /// let result = handle.join(&cx).await?;
326    /// ```
327    ///
328    /// # Example with RRef
329    ///
330    /// ```ignore
331    /// // Allocate expensive data in region heap
332    /// let index = region_record.heap_alloc(vec![1, 2, 3, 4, 5]);
333    /// let rref = RRef::<Vec<i32>>::new(region_id, index);
334    ///
335    /// // RRef is Copy + Send, can be captured by multiple tasks
336    /// scope.spawn(&mut state, &cx, move |cx| async move {
337    ///     // Would access via runtime state in real code
338    ///     process_data(rref).await
339    /// });
340    /// ```
341    ///
342    /// # Compile-Time Errors
343    ///
344    /// Attempting to capture `!Send` types fails at compile time:
345    ///
346    /// ```compile_fail,E0277
347    /// # // This test demonstrates that Rc cannot be captured
348    /// use std::rc::Rc;
349    /// fn require_send<T: Send>(_: &T) {}
350    /// fn test_rc_rejected<'r, P: asupersync::types::Policy>(
351    ///     scope: &asupersync::cx::Scope<'r, P>,
352    ///     state: &mut asupersync::runtime::RuntimeState,
353    ///     cx: &asupersync::cx::Cx,
354    /// ) {
355    ///     let rc = Rc::new(42);
356    ///     require_send(&rc);
357    ///     let _ = scope.spawn(state, cx, move |_| async move {
358    ///         let _ = rc;  // Rc<i32> is not Send
359    ///     });
360    /// }
361    /// ```
362    ///
363    /// Attempting to capture non-`'static` references fails:
364    ///
365    /// ```compile_fail,E0597
366    /// # // This test demonstrates that borrowed data cannot be captured
367    /// fn require_static<T: 'static>(_: T) {}
368    /// fn test_borrow_rejected<'r, P: asupersync::types::Policy>(
369    ///     scope: &asupersync::cx::Scope<'r, P>,
370    ///     state: &mut asupersync::runtime::RuntimeState,
371    ///     cx: &asupersync::cx::Cx,
372    /// ) {
373    ///     let local = 42;
374    ///     let borrow = &local;
375    ///     require_static(borrow);
376    ///     let _ = scope.spawn(state, cx, move |_| async move {
377    ///         let _ = borrow;  // &i32 is not 'static
378    ///     });
379    /// }
380    /// ```
381    pub fn spawn<F, Fut, Caps>(
382        &self,
383        state: &mut RuntimeState,
384        cx: &Cx<Caps>,
385        f: F,
386    ) -> Result<(TaskHandle<Fut::Output>, StoredTask), SpawnError>
387    where
388        Caps: cap::HasSpawn + Send + Sync + 'static,
389        F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
390        Fut: Future + Send + 'static,
391        Fut::Output: Send + 'static,
392    {
393        // Create oneshot channel for result delivery
394        let (tx, rx) = oneshot::channel::<Result<Fut::Output, JoinError>>();
395
396        // Create task record
397        let task_id = self.create_task_record(state)?;
398
399        // Trace task spawn event
400        let _span = debug_span!(
401            "task_spawn",
402            task_id = ?task_id,
403            region_id = ?self.region,
404            initial_state = "Created",
405            budget_deadline = ?self.budget.deadline,
406            budget_poll_quota = self.budget.poll_quota,
407            budget_cost_quota = ?self.budget.cost_quota,
408            budget_priority = self.budget.priority,
409            budget_source = "scope"
410        )
411        .entered();
412        debug!(
413            task_id = ?task_id,
414            region_id = ?self.region,
415            initial_state = "Created",
416            budget_deadline = ?self.budget.deadline,
417            budget_poll_quota = self.budget.poll_quota,
418            budget_cost_quota = ?self.budget.cost_quota,
419            budget_priority = self.budget.priority,
420            budget_source = "scope",
421            "task spawned"
422        );
423
424        let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
425
426        // Create the TaskHandle
427        let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
428
429        // Set the shared inner state in the TaskRecord
430        // This links the user-facing Cx to the runtime's TaskRecord
431        if let Some(record) = state.task_mut(task_id) {
432            record.set_cx_inner(child_cx.inner.clone());
433            record.set_cx(child_cx_full.clone());
434        }
435
436        // br-asupersync-qg5th0: result delivery through `tx.send_blocking`
437        // (no Cx) instead of `tx.send(&cx_for_send, ...)`. The wrapped
438        // future runs under the task's own Cx, which gets cancelled on
439        // `handle.abort()` or region cancel. Routing the deliver-result
440        // step through that same Cx caused `tx.send` to fail with
441        // `SendError::Cancelled` exactly when the task explicitly
442        // observed cancellation and tried to return its
443        // cancellation-aware payload (e.g. `"cancelled"`). Without the
444        // Cx-cancel check the post-completion delivery is unconditional,
445        // which is what consumers of `TaskHandle::join` expect.
446
447        // Instantiate the future with the child context.
448        // We use a guard to rollback task creation if the factory panics.
449        // This prevents zombie tasks (recorded but never started) which would
450        // cause the region to never close (deadlock).
451        let future = {
452            struct TaskCreationGuard<'a> {
453                state: &'a mut RuntimeState,
454                task_id: TaskId,
455                region_id: RegionId,
456                committed: bool,
457            }
458
459            impl Drop for TaskCreationGuard<'_> {
460                fn drop(&mut self) {
461                    if !self.committed {
462                        // Rollback task creation
463                        if let Some(region) = self.state.region_mut(self.region_id) {
464                            region.remove_task(self.task_id);
465                        }
466                        self.state.recycle_task(self.task_id);
467                    }
468                }
469            }
470
471            let mut guard = TaskCreationGuard {
472                state,
473                task_id,
474                region_id: self.region,
475                committed: false,
476            };
477
478            let fut = f(child_cx);
479            guard.committed = true;
480            fut
481        };
482
483        // Wrap the future to send its result through the channel
484        // We use CatchUnwind to ensure panics are propagated as JoinError::Panicked
485        // rather than silent channel closure (which looks like cancellation).
486        let wrapped = async move {
487            let result_result = CatchUnwind { inner: future }.await;
488            match result_result {
489                Ok(result) => {
490                    let _ = tx.send_blocking(Ok(result));
491                    crate::types::Outcome::Ok(())
492                }
493                Err(payload) => {
494                    let msg = payload_to_string(&payload);
495                    let panic_payload = PanicPayload::new(msg);
496                    let _ = tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
497                    crate::types::Outcome::Panicked(panic_payload)
498                }
499            }
500        };
501
502        // Create stored task with task_id for poll tracing
503        let stored = StoredTask::new_with_id(wrapped, task_id);
504
505        Ok((handle, stored))
506    }
507
508    /// Spawns a Send task (explicit Task Tier API).
509    ///
510    /// This is an explicit alias for [`spawn`](Self::spawn) that makes the
511    /// execution tier clear in the API. Use this when you want to emphasize
512    /// that the task may migrate between workers.
513    ///
514    /// # Type Bounds (Soundness Rules)
515    ///
516    /// Same as [`spawn`](Self::spawn):
517    /// - `F: FnOnce(Cx) -> Fut + Send + 'static`
518    /// - `Fut: Future + Send + 'static`
519    /// - `Fut::Output: Send + 'static`
520    ///
521    /// # Example
522    ///
523    /// ```ignore
524    /// // Explicit task tier spawn
525    /// let (handle, stored) = scope.spawn_task(&mut state, &cx, |cx| async move {
526    ///     // This task may run on any worker
527    ///     compute_parallel().await
528    /// })?;
529    /// ```
530    #[inline]
531    pub fn spawn_task<F, Fut, Caps>(
532        &self,
533        state: &mut RuntimeState,
534        cx: &Cx<Caps>,
535        f: F,
536    ) -> Result<(TaskHandle<Fut::Output>, StoredTask), SpawnError>
537    where
538        Caps: cap::HasSpawn + Send + Sync + 'static,
539        F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
540        Fut: Future + Send + 'static,
541        Fut::Output: Send + 'static,
542    {
543        self.spawn(state, cx, f)
544    }
545
546    /// Spawns a task and registers it with the runtime state.
547    ///
548    /// This is a convenience method that combines `spawn()` with
549    /// `RuntimeState::store_spawned_task()`. It's the primary method
550    /// used by the `spawn!` macro.
551    ///
552    /// # Arguments
553    ///
554    /// * `state` - The runtime state (for storing the task)
555    /// * `cx` - The capability context (for creating child context)
556    /// * `f` - A closure that produces the future, receiving the new task's `Cx`
557    ///
558    /// # Returns
559    ///
560    /// A `TaskHandle<T>` for awaiting the task's result.
561    ///
562    /// # Example
563    ///
564    /// ```ignore
565    /// let handle = scope.spawn_registered(&mut state, &cx, |cx| async move {
566    ///     cx.trace("Child task running");
567    ///     compute_value().await
568    /// })?;
569    ///
570    /// let result = handle.join(&cx).await?;
571    /// ```
572    pub fn spawn_registered<F, Fut, Caps>(
573        &self,
574        state: &mut RuntimeState,
575        cx: &Cx<Caps>,
576        f: F,
577    ) -> Result<TaskHandle<Fut::Output>, SpawnError>
578    where
579        Caps: cap::HasSpawn + Send + Sync + 'static,
580        F: FnOnce(Cx<Caps>) -> Fut + Send + 'static,
581        Fut: Future + Send + 'static,
582        Fut::Output: Send + 'static,
583    {
584        let (handle, stored) = self.spawn(state, cx, f)?;
585        state.store_spawned_task(handle.task_id(), stored);
586        Ok(handle)
587    }
588
589    /// Spawns a local (non-Send) task within this scope's region (**Fiber Tier**).
590    ///
591    /// This is the **Fiber Tier** spawn method. Local tasks are pinned to the
592    /// current worker thread and cannot be stolen by other workers. This enables
593    /// borrow-friendly execution with `!Send` types like `Rc` or `RefCell`.
594    ///
595    /// # Execution Tier: Fiber
596    ///
597    /// | Property | Value |
598    /// |----------|-------|
599    /// | Migration | Never (thread-pinned) |
600    /// | Send bound | Not required |
601    /// | Borrowing | Requires `'static` (no local `&T`) |
602    /// | Use case | `!Send` types, borrowed data |
603    ///
604    /// # Arguments
605    ///
606    /// * `state` - The runtime state
607    /// * `cx` - The capability context
608    /// * `f` - A closure that produces the future, receiving the new task's `Cx`
609    ///
610    /// # Panics
611    ///
612    /// Panics if called from a blocking thread (spawn_blocking context).
613    ///
614    /// # Example
615    ///
616    /// ```ignore
617    /// use std::rc::Rc;
618    /// use std::cell::RefCell;
619    ///
620    /// let counter = Rc::new(RefCell::new(0));
621    /// let counter_clone = counter.clone();
622    ///
623    /// let handle = scope.spawn_local(&mut state, &cx, |cx| async move {
624    ///     // Rc<RefCell<_>> is !Send but allowed in local tasks
625    ///     *counter_clone.borrow_mut() += 1;
626    /// });
627    /// ```
628    #[allow(clippy::too_many_lines)]
629    pub fn spawn_local<F, Fut, Caps>(
630        &self,
631        state: &mut RuntimeState,
632        cx: &Cx<Caps>,
633        f: F,
634    ) -> Result<TaskHandle<Fut::Output>, SpawnError>
635    where
636        Caps: cap::HasSpawn + Send + Sync + 'static,
637        F: FnOnce(Cx<Caps>) -> Fut + 'static,
638        Fut: Future + 'static,
639        Fut::Output: Send + 'static,
640    {
641        use crate::runtime::stored_task::LocalStoredTask;
642        use crate::runtime::task_handle::JoinError;
643
644        // Create oneshot channel for result delivery
645        let (result_tx, rx) = oneshot::channel::<Result<Fut::Output, JoinError>>();
646
647        // Create task record
648        let task_id = self.create_task_record(state)?;
649
650        // Trace task spawn event
651        let _span = debug_span!(
652            "task_spawn",
653            task_id = ?task_id,
654            region_id = ?self.region,
655            initial_state = "Created",
656            budget_deadline = ?self.budget.deadline,
657            budget_poll_quota = self.budget.poll_quota,
658            budget_cost_quota = ?self.budget.cost_quota,
659            budget_priority = self.budget.priority,
660            budget_source = "scope_local"
661        )
662        .entered();
663        debug!(
664            task_id = ?task_id,
665            region_id = ?self.region,
666            initial_state = "Created",
667            budget_deadline = ?self.budget.deadline,
668            budget_poll_quota = self.budget.poll_quota,
669            budget_cost_quota = ?self.budget.cost_quota,
670            budget_priority = self.budget.priority,
671            budget_source = "scope_local",
672            "local task spawned"
673        );
674
675        let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
676
677        // Create the TaskHandle
678        let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
679
680        // Set the shared inner state in the TaskRecord
681        if let Some(record) = state.task_mut(task_id) {
682            record.set_cx_inner(child_cx.inner.clone());
683            record.set_cx(child_cx_full.clone());
684        }
685
686        // br-asupersync-qg5th0: see comment on the `spawn` sibling above —
687        // result delivery uses `result_tx.send_blocking` so a cancelled
688        // task can still publish its cancellation-aware payload.
689
690        // Instantiate the future with the child context.
691        // We use a guard to rollback task creation if the factory panics.
692        let future = {
693            struct TaskCreationGuard<'a> {
694                state: &'a mut RuntimeState,
695                task_id: TaskId,
696                region_id: RegionId,
697                committed: bool,
698            }
699
700            impl Drop for TaskCreationGuard<'_> {
701                fn drop(&mut self) {
702                    if !self.committed {
703                        // Rollback task creation
704                        if let Some(region) = self.state.region_mut(self.region_id) {
705                            region.remove_task(self.task_id);
706                        }
707                        self.state.recycle_task(self.task_id);
708                    }
709                }
710            }
711
712            let mut guard = TaskCreationGuard {
713                state,
714                task_id,
715                region_id: self.region,
716                committed: false,
717            };
718
719            let fut = f(child_cx);
720            guard.committed = true;
721            fut
722        };
723
724        // Wrap the future to send its result through the channel
725        let wrapped = async move {
726            let result_result = CatchUnwind { inner: future }.await;
727            match result_result {
728                Ok(result) => {
729                    let _ = result_tx.send_blocking(Ok(result));
730                    crate::types::Outcome::Ok(())
731                }
732                Err(payload) => {
733                    let msg = payload_to_string(&payload);
734                    let panic_payload = PanicPayload::new(msg);
735                    let _ =
736                        result_tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
737                    crate::types::Outcome::Panicked(panic_payload)
738                }
739            }
740        };
741
742        // Create local stored task
743        let stored = LocalStoredTask::new_with_id(wrapped, task_id);
744
745        // Store in thread-local storage
746        crate::runtime::local::store_local_task(task_id, stored);
747
748        // Mark the task record as local so that safety guards in the scheduler
749        // (inject_ready panic, try_steal debug_assert) can detect accidental
750        // cross-thread migration of !Send futures.
751        if let Some(record) = state.task_mut(task_id) {
752            if let Some(worker_id) = crate::runtime::scheduler::three_lane::current_worker_id() {
753                record.pin_to_worker(worker_id);
754            } else {
755                record.mark_local();
756            }
757            record.wake_state.notify();
758        }
759
760        // Schedule the task on the current worker's NON-STEALABLE local scheduler.
761        // spawn_local tasks MUST NOT be stealable.
762        let scheduled = crate::runtime::scheduler::three_lane::schedule_local_task(task_id);
763
764        if scheduled {
765            return Ok(handle);
766        }
767
768        // No local scheduler available: rollback to avoid a permanently parked task.
769        let _ = crate::runtime::local::remove_local_task(task_id);
770        if let Some(region) = state.region(self.region) {
771            region.remove_task(task_id);
772        }
773        state.recycle_task(task_id);
774        Err(SpawnError::LocalSchedulerUnavailable)
775    }
776
777    /// Spawns a blocking operation on a dedicated thread pool.
778    ///
779    /// This is used for CPU-bound or legacy synchronous operations that
780    /// should not block async workers. The closure runs on a separate
781    /// thread pool designed for blocking work.
782    ///
783    /// # Arguments
784    ///
785    /// * `state` - The runtime state
786    /// * `cx` - The capability context
787    /// * `f` - The blocking closure to run, receiving a context
788    ///
789    /// # Type Bounds
790    ///
791    /// * `F: FnOnce(Cx) -> R + Send + 'static` - The closure must be Send
792    /// * `R: Send + 'static` - The result must be Send
793    ///
794    /// # Example
795    ///
796    /// ```ignore
797    /// let (handle, stored) = scope.spawn_blocking(&mut state, &cx, |cx| {
798    ///     cx.trace("Starting blocking work");
799    ///     // CPU-intensive work
800    ///     expensive_computation()
801    /// });
802    ///
803    /// let result = handle.join(&cx).await?;
804    /// ```
805    ///
806    /// # Note
807    ///
808    /// In Phase 0 (single-threaded), blocking operations run inline.
809    /// A proper blocking pool is implemented in Phase 1+.
810    pub fn spawn_blocking<F, R, Caps>(
811        &self,
812        state: &mut RuntimeState,
813        cx: &Cx<Caps>, // Parent Cx
814        f: F,
815    ) -> Result<(TaskHandle<R>, StoredTask), SpawnError>
816    where
817        Caps: cap::HasSpawn + Send + Sync + 'static,
818        F: FnOnce(Cx<Caps>) -> R + Send + 'static,
819        R: Send + 'static,
820    {
821        // Create oneshot channel for result delivery
822        let (tx, rx) = oneshot::channel::<Result<R, JoinError>>();
823
824        // Create task record
825        let task_id = self.create_task_record(state)?;
826
827        // Trace task spawn event
828        debug!(
829            task_id = ?task_id,
830            region_id = ?self.region,
831            initial_state = "Created",
832            poll_quota = self.budget.poll_quota,
833            spawn_kind = "blocking",
834            "blocking task spawned"
835        );
836
837        let (child_cx, child_cx_full) = self.build_child_task_cx(state, cx, task_id);
838
839        // Create the TaskHandle
840        let handle = TaskHandle::new(task_id, rx, Arc::downgrade(&child_cx.inner));
841
842        // Set the shared inner state in the TaskRecord
843        if let Some(record) = state.task_mut(task_id) {
844            record.set_cx_inner(child_cx.inner.clone());
845            record.set_cx(child_cx_full.clone());
846        }
847
848        // br-asupersync-qg5th0: see comment on the `spawn` sibling above —
849        // result delivery uses `tx.send_blocking` so a cancelled blocking
850        // task can still publish its cancellation-aware payload.
851
852        // For Phase 0, we run blocking code as an async task
853        // In Phase 1+, this would spawn on a blocking thread pool
854        let wrapped = async move {
855            // Execute the blocking closure with child context
856            // Catch panics to report them correctly
857            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(child_cx)));
858            match result {
859                Ok(res) => {
860                    let _ = tx.send_blocking(Ok(res));
861                    crate::types::Outcome::Ok(())
862                }
863                Err(payload) => {
864                    let msg = payload_to_string(&payload);
865                    let panic_payload = PanicPayload::new(msg);
866                    let _ = tx.send_blocking(Err(JoinError::Panicked(panic_payload.clone())));
867                    crate::types::Outcome::Panicked(panic_payload)
868                }
869            }
870        };
871
872        let stored = StoredTask::new_with_id(wrapped, task_id);
873
874        Ok((handle, stored))
875    }
876
877    // =========================================================================
878    // Child Regions
879    // =========================================================================
880
881    /// Creates a child region and runs the provided future within a child scope.
882    ///
883    /// The child region inherits the parent's budget by default. Use
884    /// [`Scope::region_with_budget`] to tighten constraints for the child.
885    ///
886    /// The returned outcome is the result of the body future. After the body
887    /// completes, the child region begins its close sequence and advances until
888    /// it can close (assuming all child tasks have completed and obligations are resolved).
889    ///
890    /// # Errors
891    ///
892    /// Returns [`RegionCreateError`] if the parent is closed, missing, or at capacity.
893    pub async fn region<P2, F, Fut, T, Caps>(
894        &self,
895        state: &mut RuntimeState,
896        cx: &Cx<Caps>,
897        policy: P2,
898        f: F,
899    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
900    where
901        P2: Policy,
902        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
903        Fut: Future<Output = Outcome<T, P2::Error>>,
904    {
905        self.region_with_budget(state, cx, self.budget, policy, f)
906            .await
907    }
908
909    /// Creates a child region with an explicit budget (met with the parent budget).
910    ///
911    /// The effective budget is `parent.meet(child)` to ensure nested scopes can
912    /// never relax constraints.
913    pub async fn region_with_budget<P2, F, Fut, T, Caps>(
914        &self,
915        state: &mut RuntimeState,
916        _cx: &Cx<Caps>,
917        budget: Budget,
918        _policy: P2,
919        f: F,
920    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
921    where
922        P2: Policy,
923        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
924        Fut: Future<Output = Outcome<T, P2::Error>>,
925    {
926        self.region_with_budget_and_priority(state, _cx, budget, RegionPriority::Normal, _policy, f)
927            .await
928    }
929
930    /// Creates a child region with an explicit resource-pressure priority.
931    ///
932    /// The child inherits this scope's scheduler and capability budgets while
933    /// classifying the admission request before pressure checks run.
934    pub async fn region_with_priority<P2, F, Fut, T, Caps>(
935        &self,
936        state: &mut RuntimeState,
937        _cx: &Cx<Caps>,
938        priority: RegionPriority,
939        _policy: P2,
940        f: F,
941    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
942    where
943        P2: Policy,
944        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
945        Fut: Future<Output = Outcome<T, P2::Error>>,
946    {
947        self.region_with_budget_and_priority(state, _cx, self.budget, priority, _policy, f)
948            .await
949    }
950
951    /// Creates a child region with explicit scheduler budget and pressure priority.
952    pub async fn region_with_budget_and_priority<P2, F, Fut, T, Caps>(
953        &self,
954        state: &mut RuntimeState,
955        _cx: &Cx<Caps>,
956        budget: Budget,
957        priority: RegionPriority,
958        _policy: P2,
959        f: F,
960    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
961    where
962        P2: Policy,
963        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
964        Fut: Future<Output = Outcome<T, P2::Error>>,
965    {
966        self.region_with_child_admission(
967            state,
968            _cx,
969            ChildRegionAdmission {
970                budget,
971                capability_budget: CapabilityBudget::UNSPECIFIED,
972                requirements: CapabilityBudgetRequirements::NONE,
973                priority,
974            },
975            _policy,
976            f,
977        )
978        .await
979    }
980
981    /// Creates a child region with explicit scheduler and capability budgets.
982    ///
983    /// Both budgets inherit from the parent scope and can only be tightened by
984    /// the child-supplied envelopes. Required capability dimensions fail closed
985    /// if neither the parent nor child supplies a non-exhausted envelope.
986    pub async fn region_with_budget_and_capability_budget<P2, F, Fut, T, Caps>(
987        &self,
988        state: &mut RuntimeState,
989        _cx: &Cx<Caps>,
990        budget: Budget,
991        capability_budget: CapabilityBudget,
992        requirements: CapabilityBudgetRequirements,
993        _policy: P2,
994        f: F,
995    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
996    where
997        P2: Policy,
998        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
999        Fut: Future<Output = Outcome<T, P2::Error>>,
1000    {
1001        self.region_with_child_admission(
1002            state,
1003            _cx,
1004            ChildRegionAdmission {
1005                budget,
1006                capability_budget,
1007                requirements,
1008                priority: RegionPriority::Normal,
1009            },
1010            _policy,
1011            f,
1012        )
1013        .await
1014    }
1015
1016    async fn region_with_child_admission<P2, F, Fut, T, Caps>(
1017        &self,
1018        state: &mut RuntimeState,
1019        _cx: &Cx<Caps>,
1020        admission: ChildRegionAdmission,
1021        _policy: P2,
1022        f: F,
1023    ) -> Result<Outcome<T, P2::Error>, RegionCreateError>
1024    where
1025        P2: Policy,
1026        F: FnOnce(Scope<'_, P2>, &mut RuntimeState) -> Fut,
1027        Fut: Future<Output = Outcome<T, P2::Error>>,
1028    {
1029        let child_region = state.create_child_region_with_capability_budget_and_priority(
1030            self.region,
1031            admission.budget,
1032            admission.capability_budget,
1033            admission.requirements,
1034            admission.priority,
1035        )?;
1036        let child_budget = state
1037            .region(child_region)
1038            .map_or(self.budget, crate::record::RegionRecord::budget);
1039        let child_capability_budget = state.region(child_region).map_or(
1040            self.capability_budget,
1041            crate::record::RegionRecord::capability_budget,
1042        );
1043        let child_scope = Scope::<P2>::new_with_capability_budget(
1044            child_region,
1045            child_budget,
1046            child_capability_budget,
1047        );
1048
1049        let fut_result =
1050            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(child_scope, &mut *state)));
1051
1052        let fut = match fut_result {
1053            Ok(fut) => fut,
1054            Err(payload) => {
1055                let reason = CancelReason::fail_fast().with_region(child_region);
1056                let _ = state.cancel_request(child_region, &reason, None);
1057
1058                // Factory panicked after `scope.spawn(...)` returned the
1059                // `(TaskHandle, StoredTask)` pair but before the StoredTask
1060                // was scheduled. The matching TaskRecord is orphaned in
1061                // `TaskState::Created` — no executor will ever poll it, so
1062                // `region.task_count()` stays > 0 and `advance_region_state`
1063                // loops forever in Closing. Rip out any such never-polled
1064                // records bound to this child region so the region can reach
1065                // Finalizing and the surrounding close future can complete.
1066                //
1067                // br-asupersync-qg5th0: the `cancel_request` call above
1068                // transitions every region task from `Created` into
1069                // `CancelRequested { .. }` before this filter runs, so
1070                // matching on `Created` alone misses orphans whose state
1071                // was just bumped by the cancel pass. An unpolled task can
1072                // only ever reach `Created` or `CancelRequested` — anything
1073                // beyond requires the task to have been polled at least
1074                // once (which means `cancel_request` reached the executor,
1075                // which means the StoredTask was scheduled, which is
1076                // exactly what an orphan is not). Match both variants so
1077                // the orphan reap is robust to the cancel pass that
1078                // immediately precedes it.
1079                let orphan_task_ids: Vec<TaskId> = state
1080                    .tasks_iter()
1081                    .filter_map(|(_, t)| {
1082                        if t.owner == child_region
1083                            && matches!(
1084                                t.state,
1085                                TaskState::Created | TaskState::CancelRequested { .. }
1086                            )
1087                        {
1088                            Some(t.id)
1089                        } else {
1090                            None
1091                        }
1092                    })
1093                    .collect();
1094                for task_id in orphan_task_ids {
1095                    if let Some(region) = state.region(child_region) {
1096                        region.remove_task(task_id);
1097                    }
1098                    state.recycle_task(task_id);
1099                }
1100
1101                if let Some(region) = state.region(child_region) {
1102                    region.begin_close(None);
1103                }
1104                state.advance_region_state(child_region);
1105                std::panic::resume_unwind(payload);
1106            }
1107        };
1108
1109        let pinned_fut = std::pin::pin!(CatchUnwind { inner: fut });
1110
1111        let runner = RegionRunner {
1112            fut: pinned_fut,
1113            state: Some(state),
1114            child_region,
1115        };
1116
1117        let (result, state) = runner.await;
1118        let outcome = match result {
1119            Ok(outcome) => outcome,
1120            Err(payload) => {
1121                let msg = payload_to_string(&payload);
1122                Outcome::Panicked(PanicPayload::new(msg))
1123            }
1124        };
1125
1126        match &outcome {
1127            Outcome::Ok(_) => {
1128                if let Some(region) = state.region(child_region) {
1129                    region.begin_close(None);
1130                }
1131            }
1132            Outcome::Cancelled(reason) => {
1133                let _ = state.cancel_request(child_region, reason, None);
1134                if let Some(region) = state.region(child_region) {
1135                    region.begin_close(None);
1136                }
1137            }
1138            Outcome::Err(_) | Outcome::Panicked(_) => {
1139                let reason = CancelReason::fail_fast().with_region(child_region);
1140                let _ = state.cancel_request(child_region, &reason, None);
1141                if let Some(region) = state.region(child_region) {
1142                    region.begin_close(None);
1143                }
1144            }
1145        }
1146
1147        let close_notify = state.region(child_region).map(|r| r.close_notify.clone());
1148        state.advance_region_state(child_region);
1149
1150        if let Some(notify) = close_notify {
1151            RegionCloseFuture { state: notify }.await;
1152        }
1153
1154        Ok(outcome)
1155    }
1156
1157    // =========================================================================
1158    // Combinators
1159    // =========================================================================
1160
1161    /// Joins two tasks, waiting for both to complete.
1162    ///
1163    /// This method waits for both tasks to complete, regardless of their outcome.
1164    /// It returns a tuple of results.
1165    ///
1166    /// # Example
1167    /// ```ignore
1168    /// let (h1, _) = scope.spawn(...);
1169    /// let (h2, _) = scope.spawn(...);
1170    /// let (r1, r2) = scope.join(cx, h1, h2).await;
1171    /// ```
1172    pub async fn join<T1, T2>(
1173        &self,
1174        cx: &Cx,
1175        mut h1: TaskHandle<T1>,
1176        mut h2: TaskHandle<T2>,
1177    ) -> (Result<T1, JoinError>, Result<T2, JoinError>) {
1178        let mut f1 = h1.join(cx);
1179        let mut f2 = h2.join(cx);
1180        let r1 = std::pin::Pin::new(&mut f1).await;
1181        let r2 = std::pin::Pin::new(&mut f2).await;
1182        (r1, r2)
1183    }
1184
1185    /// Races two tasks, waiting for the first to complete.
1186    ///
1187    /// The loser is cancelled and drained (awaited until it completes cancellation).
1188    ///
1189    /// # Example
1190    /// ```ignore
1191    /// let (h1, _) = scope.spawn(...);
1192    /// let (h2, _) = scope.spawn(...);
1193    /// match scope.race(cx, h1, h2).await {
1194    ///     Ok(val) => println!("Winner result: {val}"),
1195    ///     Err(e) => println!("Race failed: {e}"),
1196    /// }
1197    /// ```
1198    fn record_loser_drain_start(&self, cx: &Cx, participants: Vec<TaskId>) -> Option<u64> {
1199        let time = cx.now_for_observability();
1200        cx.loser_drain_history_handle()
1201            .map(|history| history.record_race_start(self.region, participants, time))
1202    }
1203
1204    fn record_loser_drain_task_complete(cx: &Cx, task: TaskId) {
1205        if let Some(history) = cx.loser_drain_history_handle() {
1206            history.record_task_complete(task, cx.now_for_observability());
1207        }
1208    }
1209
1210    fn record_loser_drain_complete(cx: &Cx, race_id: Option<u64>, winner: TaskId) {
1211        if let (Some(race_id), Some(history)) = (race_id, cx.loser_drain_history_handle()) {
1212            history.record_race_complete(race_id, winner, cx.now_for_observability());
1213        }
1214    }
1215
1216    fn best_effort_poll_loser_join<T>(cx: &Cx, handle: &mut TaskHandle<T>) -> bool {
1217        let mut drain = std::pin::pin!(handle.join(cx));
1218        let waker = std::task::Waker::noop();
1219        let mut poll_cx = std::task::Context::from_waker(waker);
1220        matches!(drain.as_mut().poll(&mut poll_cx), Poll::Ready(_))
1221    }
1222
1223    /// Races two task handles and returns the winner while draining the loser.
1224    pub async fn race<T>(
1225        &self,
1226        cx: &Cx,
1227        mut h1: TaskHandle<T>,
1228        mut h2: TaskHandle<T>,
1229    ) -> Result<T, JoinError> {
1230        let race_id = self.record_loser_drain_start(cx, vec![h1.task_id(), h2.task_id()]);
1231        let winner = {
1232            let f1 = h1.join_with_drop_reason(cx, CancelReason::race_loser());
1233            let mut f1 = std::pin::pin!(f1);
1234            let f2 = h2.join_with_drop_reason(cx, CancelReason::race_loser());
1235            let mut f2 = std::pin::pin!(f2);
1236            Select::new(f1.as_mut(), f2.as_mut())
1237                .await
1238                .map_err(|_| JoinError::PolledAfterCompletion)?
1239        };
1240
1241        match winner {
1242            Either::Left(res) => {
1243                Self::record_loser_drain_task_complete(cx, h1.task_id());
1244                if matches!(&res, Err(JoinError::Panicked(_)))
1245                    && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1246                {
1247                    // In direct block_on tests there is no scheduler driving the
1248                    // loser task after the winner panic surfaces. Best-effort poll
1249                    // once so a cooperative loser can observe cancellation, then
1250                    // preserve the winner panic without deadlocking the test.
1251                    if Self::best_effort_poll_loser_join(cx, &mut h2) {
1252                        Self::record_loser_drain_task_complete(cx, h2.task_id());
1253                    }
1254                    Self::record_loser_drain_complete(cx, race_id, h1.task_id());
1255                    return res;
1256                }
1257                let loser_res = h2.join(cx).await;
1258                Self::record_loser_drain_task_complete(cx, h2.task_id());
1259                Self::record_loser_drain_complete(cx, race_id, h1.task_id());
1260                if let Err(JoinError::Panicked(p)) = res {
1261                    Err(JoinError::Panicked(p))
1262                } else if let Err(JoinError::Panicked(p)) = loser_res {
1263                    Err(JoinError::Panicked(p))
1264                } else {
1265                    res
1266                }
1267            }
1268            Either::Right(res) => {
1269                Self::record_loser_drain_task_complete(cx, h2.task_id());
1270                if matches!(&res, Err(JoinError::Panicked(_)))
1271                    && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1272                {
1273                    // See the left-branch comment above.
1274                    if Self::best_effort_poll_loser_join(cx, &mut h1) {
1275                        Self::record_loser_drain_task_complete(cx, h1.task_id());
1276                    }
1277                    Self::record_loser_drain_complete(cx, race_id, h2.task_id());
1278                    return res;
1279                }
1280                let loser_res = h1.join(cx).await;
1281                Self::record_loser_drain_task_complete(cx, h1.task_id());
1282                Self::record_loser_drain_complete(cx, race_id, h2.task_id());
1283                if let Err(JoinError::Panicked(p)) = res {
1284                    Err(JoinError::Panicked(p))
1285                } else if let Err(JoinError::Panicked(p)) = loser_res {
1286                    Err(JoinError::Panicked(p))
1287                } else {
1288                    res
1289                }
1290            }
1291        }
1292    }
1293
1294    /// Hedges a primary operation with a backup operation.
1295    ///
1296    /// 1. Spawns the primary task immediately.
1297    /// 2. Waits for the delay.
1298    /// 3. If primary finishes before delay: returns primary result.
1299    /// 4. If delay fires: spawns backup task and races them.
1300    ///
1301    /// The loser is cancelled and drained.
1302    ///
1303    /// # Arguments
1304    /// * `state` - The runtime state
1305    /// * `cx` - The capability context
1306    /// * `delay` - The hedge delay
1307    /// * `primary` - The primary future factory
1308    /// * `backup` - The backup future factory
1309    ///
1310    /// # Returns
1311    /// `Ok(T)` if successful, `Err(JoinError)` if failed/cancelled.
1312    pub async fn hedge<F1, Fut1, F2, Fut2, T>(
1313        &self,
1314        state: &mut RuntimeState,
1315        cx: &Cx,
1316        delay: std::time::Duration,
1317        primary: F1,
1318        backup: F2,
1319    ) -> Result<T, JoinError>
1320    where
1321        F1: FnOnce(Cx) -> Fut1 + Send + 'static,
1322        Fut1: Future<Output = T> + Send + 'static,
1323        F2: FnOnce(Cx) -> Fut2 + Send + 'static,
1324        Fut2: Future<Output = T> + Send + 'static,
1325        T: Send + 'static,
1326    {
1327        use crate::combinator::Either;
1328        use crate::combinator::select::Select;
1329        // 1. Spawn primary
1330        let mut h1 = self
1331            .spawn_registered(state, cx, primary)
1332            .map_err(|_| JoinError::Cancelled(CancelReason::resource_unavailable()))?;
1333
1334        // 2. Race primary vs delay.
1335        // Scope the pinned join future so we can safely reuse h1 afterwards.
1336        let primary_or_delay = {
1337            let f1_primary = h1.join(cx);
1338            let mut f1_primary = std::pin::pin!(f1_primary);
1339
1340            let now = cx
1341                .timer_driver()
1342                .map_or_else(crate::time::wall_now, |d| d.now());
1343            let sleep_fut = crate::time::sleep(now, delay);
1344            let mut sleep_pinned = std::pin::pin!(sleep_fut);
1345
1346            let res = Select::new(f1_primary.as_mut(), sleep_pinned.as_mut())
1347                .await
1348                .map_err(|_| JoinError::PolledAfterCompletion)?;
1349            if matches!(res, Either::Right(())) {
1350                f1_primary.defuse_drop_abort();
1351            }
1352            res
1353        };
1354
1355        match primary_or_delay {
1356            Either::Left(res) => {
1357                // Primary finished first
1358                res
1359            }
1360            Either::Right(()) => {
1361                // Timeout fired. Spawn backup.
1362                let Ok(mut h2) = self.spawn_registered(state, cx, backup) else {
1363                    // Backup admission failed after primary already started.
1364                    // Request cancellation on primary to avoid orphaned work.
1365                    h1.abort_with_reason(CancelReason::resource_unavailable());
1366
1367                    if crate::runtime::scheduler::three_lane::current_worker_id().is_some() {
1368                        // In scheduler-backed runtime execution, fully drain the
1369                        // cancelled primary before returning.
1370                        match h1.join(cx).await {
1371                            Ok(res) => return Ok(res),
1372                            Err(JoinError::Panicked(p)) => return Err(JoinError::Panicked(p)),
1373                            Err(JoinError::Cancelled(_) | JoinError::PolledAfterCompletion) => {}
1374                        }
1375                    } else {
1376                        // In no-scheduler contexts (e.g. direct unit-test block_on),
1377                        // full join can deadlock because nothing drives stored tasks.
1378                        // Keep this as best-effort and return promptly.
1379                        let mut drain = std::pin::pin!(h1.join(cx));
1380                        let waker = std::task::Waker::noop();
1381                        let mut poll_cx = Context::from_waker(waker);
1382                        match drain.as_mut().poll(&mut poll_cx) {
1383                            std::task::Poll::Ready(Ok(res)) => return Ok(res),
1384                            std::task::Poll::Ready(Err(JoinError::Panicked(p))) => {
1385                                return Err(JoinError::Panicked(p));
1386                            }
1387                            _ => {}
1388                        }
1389                    }
1390
1391                    return Err(JoinError::Cancelled(CancelReason::resource_unavailable()));
1392                };
1393
1394                // Now race h1 and h2 with bounded future borrows.
1395                let race_outcome = {
1396                    let f1_race = h1.join_with_drop_reason(cx, CancelReason::race_loser());
1397                    let mut f1_race = std::pin::pin!(f1_race);
1398                    let f2_race = h2.join_with_drop_reason(cx, CancelReason::race_loser());
1399                    let mut f2_race = std::pin::pin!(f2_race);
1400                    Select::new(f1_race.as_mut(), f2_race.as_mut())
1401                        .await
1402                        .map_err(|_| JoinError::PolledAfterCompletion)?
1403                };
1404
1405                match race_outcome {
1406                    Either::Left(res) => {
1407                        if matches!(&res, Err(JoinError::Panicked(_)))
1408                            && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1409                        {
1410                            Self::best_effort_poll_loser_join(cx, &mut h2);
1411                            return res;
1412                        }
1413                        let loser_res = h2.join(cx).await;
1414                        if let Err(JoinError::Panicked(p)) = res {
1415                            Err(JoinError::Panicked(p))
1416                        } else if let Err(JoinError::Panicked(p)) = loser_res {
1417                            Err(JoinError::Panicked(p))
1418                        } else {
1419                            res
1420                        }
1421                    }
1422                    Either::Right(res) => {
1423                        if matches!(&res, Err(JoinError::Panicked(_)))
1424                            && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1425                        {
1426                            Self::best_effort_poll_loser_join(cx, &mut h1);
1427                            return res;
1428                        }
1429                        let loser_res = h1.join(cx).await;
1430                        if let Err(JoinError::Panicked(p)) = res {
1431                            Err(JoinError::Panicked(p))
1432                        } else if let Err(JoinError::Panicked(p)) = loser_res {
1433                            Err(JoinError::Panicked(p))
1434                        } else {
1435                            res
1436                        }
1437                    }
1438                }
1439            }
1440        }
1441    }
1442
1443    /// Races multiple tasks, waiting for the first to complete.
1444    ///
1445    /// The winner's result is returned. Losers are cancelled and drained.
1446    ///
1447    /// # Arguments
1448    /// * `cx` - The capability context
1449    /// * `handles` - Vector of task handles to race
1450    ///
1451    /// # Returns
1452    /// `Ok((value, index))` if the winner succeeded.
1453    /// `Err(e)` if the winner failed (error/cancel/panic).
1454    pub async fn race_all<T>(
1455        &self,
1456        cx: &Cx,
1457        handles: Vec<TaskHandle<T>>,
1458    ) -> Result<(T, usize), JoinError> {
1459        let mut handles = handles;
1460        if handles.is_empty() {
1461            return std::future::poll_fn(|_poll_cx| {
1462                if cx.checkpoint().is_err() {
1463                    let reason = cx
1464                        .cancel_reason()
1465                        .unwrap_or_else(|| CancelReason::user("race_all cancelled"));
1466                    std::task::Poll::Ready(Err(JoinError::Cancelled(reason)))
1467                } else {
1468                    std::task::Poll::Pending
1469                }
1470            })
1471            .await;
1472        }
1473        let participant_tasks: Vec<_> = handles.iter().map(TaskHandle::task_id).collect();
1474        let race_id = self.record_loser_drain_start(cx, participant_tasks.clone());
1475
1476        let mut futures: Vec<_> = handles
1477            .iter_mut()
1478            .map(|h| h.join_with_drop_reason(cx, CancelReason::race_loser()))
1479            .collect();
1480        let mut ready_results: Vec<Option<Result<T, JoinError>>> = std::iter::repeat_with(|| None)
1481            .take(futures.len())
1482            .collect();
1483
1484        // Poll every candidate in each round and keep all same-round ready
1485        // outcomes. This prevents losing loser panic outcomes when multiple
1486        // tasks become ready in the same poll.
1487        let winner_idx = std::future::poll_fn(|poll_cx| {
1488            let mut newly_ready = Vec::new();
1489
1490            for (i, future) in futures.iter_mut().enumerate() {
1491                if ready_results[i].is_some() {
1492                    continue;
1493                }
1494                if let std::task::Poll::Ready(res) = std::pin::Pin::new(future).poll(poll_cx) {
1495                    ready_results[i] = Some(res);
1496                    newly_ready.push(i);
1497                }
1498            }
1499
1500            if newly_ready.is_empty() {
1501                std::task::Poll::Pending
1502            } else {
1503                // Fairly select a winner among all that became ready in this round
1504                let chosen = newly_ready[cx.random_usize(newly_ready.len())];
1505                std::task::Poll::Ready(chosen)
1506            }
1507        })
1508        .await;
1509
1510        let winner_result = ready_results[winner_idx]
1511            .take()
1512            .expect("winner index must have a ready result");
1513        let winner_task = participant_tasks[winner_idx];
1514        Self::record_loser_drain_task_complete(cx, winner_task);
1515
1516        // Release mutable borrows of handles held by JoinFuture values before
1517        // explicit loser cancellation/join.
1518        drop(futures);
1519
1520        // Drain completed losers first so terminal panic outcomes are not
1521        // obscured by strengthening cancellation reasons on already-finished tasks.
1522        let mut loser_panic = None;
1523        let mut pending_loser_indices = Vec::new();
1524        for (i, handle) in handles.iter_mut().enumerate() {
1525            if i == winner_idx {
1526                continue;
1527            }
1528            if let Some(res) = ready_results[i].take() {
1529                Self::record_loser_drain_task_complete(cx, handle.task_id());
1530                if let Err(JoinError::Panicked(p)) = res {
1531                    if loser_panic.is_none() {
1532                        loser_panic = Some(p);
1533                    }
1534                }
1535            } else if handle.is_finished() {
1536                let res = handle.join(cx).await;
1537                Self::record_loser_drain_task_complete(cx, handle.task_id());
1538                if let Err(JoinError::Panicked(p)) = res {
1539                    if loser_panic.is_none() {
1540                        loser_panic = Some(p);
1541                    }
1542                }
1543            } else {
1544                pending_loser_indices.push(i);
1545            }
1546        }
1547
1548        // Cancel and drain unfinished losers.
1549        // Note: Losers may also already have a race-loser reason from dropped
1550        // join futures; strengthening keeps attribution deterministic.
1551        for &idx in &pending_loser_indices {
1552            handles[idx].abort_with_reason(CancelReason::race_loser());
1553        }
1554        if matches!(&winner_result, Err(JoinError::Panicked(_)))
1555            && crate::runtime::scheduler::three_lane::current_worker_id().is_none()
1556        {
1557            // In direct block_on tests there is no scheduler driving pending
1558            // losers after the winner panic surfaces. Best-effort poll each
1559            // loser once so cooperative tasks can observe cancellation, then
1560            // preserve the winner panic without deadlocking the test.
1561            for idx in pending_loser_indices {
1562                if Self::best_effort_poll_loser_join(cx, &mut handles[idx]) {
1563                    Self::record_loser_drain_task_complete(cx, handles[idx].task_id());
1564                }
1565            }
1566            Self::record_loser_drain_complete(cx, race_id, winner_task);
1567            return winner_result.map(|val| (val, winner_idx));
1568        }
1569        for idx in pending_loser_indices {
1570            let res = handles[idx].join(cx).await;
1571            Self::record_loser_drain_task_complete(cx, handles[idx].task_id());
1572            if let Err(JoinError::Panicked(p)) = res {
1573                if loser_panic.is_none() {
1574                    loser_panic = Some(p);
1575                }
1576            }
1577        }
1578
1579        let winner_result = winner_result.map(|val| (val, winner_idx));
1580        Self::record_loser_drain_complete(cx, race_id, winner_task);
1581        if matches!(&winner_result, Err(JoinError::Panicked(_))) {
1582            return winner_result;
1583        }
1584
1585        loser_panic.map_or(winner_result, |panic_payload| {
1586            Err(JoinError::Panicked(panic_payload))
1587        })
1588    }
1589
1590    /// Joins multiple tasks, waiting for all to complete.
1591    ///
1592    /// Returns a vector of results in the same order as the input handles.
1593    pub async fn join_all<T>(
1594        &self,
1595        cx: &Cx,
1596        mut handles: Vec<TaskHandle<T>>,
1597    ) -> Vec<Result<T, JoinError>> {
1598        let mut futures: Vec<_> = handles.iter_mut().map(|h| h.join(cx)).collect();
1599        let mut results = Vec::with_capacity(futures.len());
1600        for fut in &mut futures {
1601            results.push(std::pin::Pin::new(fut).await);
1602        }
1603        results
1604    }
1605
1606    pub(crate) fn build_child_task_cx<Caps>(
1607        &self,
1608        state: &RuntimeState,
1609        parent_cx: &Cx<Caps>,
1610        task_id: TaskId,
1611    ) -> (Cx<Caps>, Cx<cap::All>) {
1612        // br-asupersync-3b5fz6: Optimize Arc handle usage to reduce 45.1% CPU hotspot
1613        // Cache observability and entropy to avoid recomputation
1614        let child_observability = parent_cx.child_observability(self.region, task_id);
1615        let child_entropy = parent_cx.child_entropy(task_id);
1616
1617        // Get driver handles once - minimize state.handle() calls
1618        let io_driver = state.io_driver_handle();
1619        let timer_driver = state.timer_driver_handle();
1620        // Note: logical_clock requires owned timer_driver, but we still save one clone
1621        let logical_clock = state
1622            .logical_clock_mode()
1623            .build_handle(timer_driver.clone());
1624
1625        // Batch parent handle retrieval to reduce method call overhead
1626        let io_cap = parent_cx.io_cap_handle();
1627        let registry = parent_cx.registry_handle();
1628        let remote_cap = parent_cx.remote_cap_handle();
1629        let blocking_pool = parent_cx.blocking_pool_handle();
1630        let evidence_sink = parent_cx.evidence_sink_handle();
1631        let macaroon = parent_cx.macaroon_handle();
1632        let pressure_opt = parent_cx.pressure_handle();
1633
1634        // Build context with cached handles - minimizes Arc clone operations
1635        let mut child_cx = Cx::<Caps>::new_with_drivers(
1636            self.region,
1637            task_id,
1638            self.budget,
1639            Some(child_observability),
1640            io_driver,
1641            io_cap,
1642            timer_driver,
1643            Some(child_entropy),
1644        )
1645        .with_logical_clock(logical_clock)
1646        .with_registry_handle(registry)
1647        .with_remote_cap_handle(remote_cap)
1648        .with_blocking_pool_handle(blocking_pool)
1649        .with_evidence_sink(evidence_sink)
1650        .with_macaroon_handle(macaroon);
1651        let _ = child_cx.apply_child_capability_budget(
1652            self.capability_budget,
1653            CapabilityBudgetRequirements::NONE,
1654        );
1655
1656        // Apply pressure handle if present
1657        if let Some(pressure) = pressure_opt {
1658            child_cx = child_cx.with_pressure(pressure);
1659        }
1660
1661        // Set state handles with minimal additional Arc clones
1662        child_cx.set_trace_buffer(state.trace_handle());
1663        child_cx.set_loser_drain_history_handle(state.loser_drain_history_handle());
1664        let child_cx_full = child_cx.retype::<cap::All>();
1665
1666        (child_cx, child_cx_full)
1667    }
1668
1669    /// Creates a task record in the runtime state.
1670    ///
1671    /// This is a helper method used by all spawn variants.
1672    pub(crate) fn create_task_record(
1673        &self,
1674        state: &mut RuntimeState,
1675    ) -> Result<TaskId, SpawnError> {
1676        let now = state
1677            .timer_driver()
1678            .map_or(state.now, crate::time::TimerDriverHandle::now);
1679
1680        let region = self.region;
1681        let budget = self.budget;
1682        let idx = state.insert_pooled_task_with(|idx, record| {
1683            // br-asupersync-j1e7zy: mutate the recycled record in place.
1684            // A previous version did `*record = TaskRecord::new_with_time(...)`,
1685            // which dropped the freshly-recycled `wake_state` Arc + `waiters`
1686            // SmallVec only to allocate replacements — making pool hits more
1687            // expensive than pool misses. `insert_pooled_task_with` (and
1688            // `Recyclable::reset` for `TaskRecord`) leave every other field at
1689            // its default, so we only need to overwrite the per-task identity
1690            // and budget fields.
1691            record.id = TaskId::from_arena(idx);
1692            record.owner = region;
1693            record.created_at = now;
1694            record.deadline = budget.deadline;
1695            record.polls_remaining = budget.poll_quota;
1696            #[cfg(feature = "tracing-integration")]
1697            {
1698                record.created_instant = crate::time::wall_now();
1699            }
1700        });
1701        let task_id = TaskId::from_arena(idx);
1702
1703        // Add task to the owning region
1704        if let Some(region) = state.region(self.region) {
1705            if let Err(err) = region.add_task(task_id) {
1706                // Rollback task creation
1707                state.recycle_task(task_id);
1708                return Err(match err {
1709                    AdmissionError::Closed => SpawnError::RegionClosed(self.region),
1710                    AdmissionError::LimitReached { limit, live, .. } => {
1711                        SpawnError::RegionAtCapacity {
1712                            region: self.region,
1713                            limit,
1714                            live,
1715                        }
1716                    }
1717                });
1718            }
1719        } else {
1720            // Rollback task creation
1721            state.recycle_task(task_id);
1722            return Err(SpawnError::RegionNotFound(self.region));
1723        }
1724
1725        state.record_task_spawn(task_id, self.region);
1726
1727        Ok(task_id)
1728    }
1729
1730    // =========================================================================
1731    // Finalizer Registration
1732    // =========================================================================
1733
1734    /// Registers a synchronous finalizer to run when the region closes.
1735    ///
1736    /// Finalizers are stored in LIFO order and executed during the Finalizing
1737    /// phase, after all children have completed. Use this for lightweight
1738    /// cleanup that doesn't need to await.
1739    ///
1740    /// # Arguments
1741    /// * `state` - The runtime state
1742    /// * `f` - The synchronous cleanup function
1743    ///
1744    /// # Returns
1745    /// `true` if the finalizer was registered successfully.
1746    ///
1747    /// # Example
1748    /// ```ignore
1749    /// scope.defer_sync(&mut state, || {
1750    ///     println!("Cleaning up!");
1751    /// });
1752    /// ```
1753    pub fn defer_sync<F>(&self, state: &mut RuntimeState, f: F) -> bool
1754    where
1755        F: FnOnce() + Send + 'static,
1756    {
1757        state.register_sync_finalizer(self.region, f)
1758    }
1759
1760    /// Registers an asynchronous finalizer to run when the region closes.
1761    ///
1762    /// Async finalizers run under a cancel mask to prevent interruption.
1763    /// They are driven to completion with a bounded budget. Use this for
1764    /// cleanup that needs to perform async operations (e.g., closing
1765    /// connections, flushing buffers).
1766    ///
1767    /// # Arguments
1768    /// * `state` - The runtime state
1769    /// * `future` - The async cleanup future
1770    ///
1771    /// # Returns
1772    /// `true` if the finalizer was registered successfully.
1773    ///
1774    /// # Example
1775    /// ```ignore
1776    /// scope.defer_async(&mut state, async {
1777    ///     close_connection().await;
1778    /// });
1779    /// ```
1780    pub fn defer_async<F>(&self, state: &mut RuntimeState, future: F) -> bool
1781    where
1782        F: Future<Output = ()> + Send + 'static,
1783    {
1784        state.register_async_finalizer(self.region, future)
1785    }
1786}
1787
1788impl<P: Policy> std::fmt::Debug for Scope<'_, P> {
1789    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1790        f.debug_struct("Scope")
1791            .field("region", &self.region)
1792            .field("budget", &self.budget)
1793            .field("capability_budget", &self.capability_budget)
1794            .finish()
1795    }
1796}
1797
1798#[cfg(test)]
1799mod tests {
1800    #![allow(
1801        clippy::pedantic,
1802        clippy::nursery,
1803        clippy::expect_fun_call,
1804        clippy::map_unwrap_or,
1805        clippy::cast_possible_wrap,
1806        clippy::future_not_send
1807    )]
1808    use super::*;
1809    use crate::record::RegionLimits;
1810    use crate::runtime::RuntimeState;
1811    use crate::types::{
1812        CancelKind, CapabilityBudgetDimension, CapabilityBudgetRefusal, Outcome, Time,
1813    };
1814    use futures_lite::future::block_on;
1815    use std::sync::Arc;
1816
1817    fn test_cx() -> Cx<cap::All> {
1818        Cx::for_testing()
1819    }
1820
1821    fn test_scope(region: RegionId, budget: Budget) -> Scope<'static> {
1822        Scope::new(region, budget)
1823    }
1824
1825    fn test_scope_with_capability_budget(
1826        region: RegionId,
1827        budget: Budget,
1828        capability_budget: CapabilityBudget,
1829    ) -> Scope<'static> {
1830        Scope::new_with_capability_budget(region, budget, capability_budget)
1831    }
1832
1833    #[test]
1834    fn spawn_creates_task_record() {
1835        let mut state = RuntimeState::new();
1836        let cx = test_cx();
1837        let region = state.create_root_region(Budget::INFINITE);
1838        let scope = test_scope(region, Budget::INFINITE);
1839
1840        let (handle, _stored) = scope
1841            .spawn(&mut state, &cx, |_| async { 42_i32 })
1842            .expect("should spawn async task");
1843
1844        // Task should exist in state
1845        let task = state.task(handle.task_id());
1846        assert!(task.is_some());
1847
1848        // Task should be owned by the region
1849        let task = task.expect("task should exist in state");
1850        assert_eq!(task.owner, region);
1851    }
1852
1853    #[test]
1854    fn spawn_inherits_registry_and_remote_capabilities() {
1855        use crate::cx::registry::RegistryHandle;
1856        use crate::remote::{NodeId, RemoteCap};
1857        use std::task::Context;
1858
1859        let mut state = RuntimeState::new();
1860
1861        let registry = crate::cx::NameRegistry::new();
1862        let registry_handle = RegistryHandle::new(Arc::new(registry));
1863        let parent_registry_arc = registry_handle.as_arc();
1864
1865        let cx = test_cx()
1866            .with_registry_handle(Some(registry_handle))
1867            .with_remote_cap(RemoteCap::new().with_local_node(NodeId::new("origin-test")));
1868
1869        let region = state.create_root_region(Budget::INFINITE);
1870        let scope = test_scope(region, Budget::INFINITE);
1871
1872        let mut handle = scope
1873            .spawn_registered(&mut state, &cx, move |cx| async move {
1874                let child_registry = cx.registry_handle().expect("child must inherit registry");
1875                let child_registry_arc = child_registry.as_arc();
1876                let same_registry = Arc::ptr_eq(&child_registry_arc, &parent_registry_arc);
1877
1878                let child_remote = cx.remote().expect("child must inherit remote cap");
1879                let origin = child_remote.local_node().as_str().to_owned();
1880
1881                (same_registry, origin)
1882            })
1883            .expect("should spawn registered task for capability inheritance test");
1884
1885        let waker = std::task::Waker::noop().clone();
1886        let mut poll_cx = Context::from_waker(&waker);
1887
1888        let stored = state
1889            .get_stored_future(handle.task_id())
1890            .expect("spawn_registered must store the task");
1891        assert!(stored.poll(&mut poll_cx).is_ready());
1892
1893        let mut join_fut = std::pin::pin!(handle.join(&cx));
1894        match join_fut.as_mut().poll(&mut poll_cx) {
1895            Poll::Ready(Ok((same_registry, origin))) => {
1896                assert!(
1897                    same_registry,
1898                    "child should observe the same RegistryCap instance"
1899                );
1900                assert_eq!(origin, "origin-test");
1901            }
1902            other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1903        }
1904    }
1905
1906    #[test]
1907    fn spawn_inherits_runtime_timer_driver() {
1908        use std::task::Context;
1909
1910        let mut state = RuntimeState::new();
1911        let clock = Arc::new(crate::time::VirtualClock::new());
1912        state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(clock));
1913
1914        let cx = test_cx();
1915        let region = state.create_root_region(Budget::INFINITE);
1916        let scope = test_scope(region, Budget::INFINITE);
1917
1918        let (mut handle, mut stored) = scope
1919            .spawn(&mut state, &cx, |cx| async move { cx.has_timer() })
1920            .expect("spawn should succeed");
1921
1922        let waker = std::task::Waker::noop().clone();
1923        let mut poll_cx = Context::from_waker(&waker);
1924        assert!(stored.poll(&mut poll_cx).is_ready());
1925
1926        let mut join_fut = std::pin::pin!(handle.join(&cx));
1927        match join_fut.as_mut().poll(&mut poll_cx) {
1928            Poll::Ready(Ok(has_timer)) => assert!(has_timer),
1929            other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1930        }
1931    }
1932
1933    #[test]
1934    fn create_task_record_uses_runtime_timer_driver_time() {
1935        let mut state = RuntimeState::new();
1936        let clock = Arc::new(crate::time::VirtualClock::starting_at(Time::from_millis(
1937            11,
1938        )));
1939        state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(
1940            clock.clone(),
1941        ));
1942
1943        let region = state.create_root_region(Budget::INFINITE);
1944        let scope = test_scope(region, Budget::INFINITE);
1945
1946        clock.advance(Time::from_millis(7).as_nanos());
1947        let task_id = scope
1948            .create_task_record(&mut state)
1949            .expect("task record should be created");
1950
1951        let task = state.task(task_id).expect("task record");
1952        assert_eq!(task.created_at, Time::from_millis(18));
1953        assert_eq!(task.id, task_id);
1954    }
1955
1956    #[test]
1957    fn spawn_blocking_inherits_runtime_timer_driver() {
1958        use std::task::Context;
1959
1960        let mut state = RuntimeState::new();
1961        let clock = Arc::new(crate::time::VirtualClock::new());
1962        state.set_timer_driver(crate::time::TimerDriverHandle::with_virtual_clock(clock));
1963
1964        let cx = test_cx();
1965        let region = state.create_root_region(Budget::INFINITE);
1966        let scope = test_scope(region, Budget::INFINITE);
1967
1968        let (mut handle, mut stored) = scope
1969            .spawn_blocking(&mut state, &cx, |cx| cx.has_timer())
1970            .expect("spawn_blocking should succeed");
1971
1972        let waker = std::task::Waker::noop().clone();
1973        let mut poll_cx = Context::from_waker(&waker);
1974        assert!(stored.poll(&mut poll_cx).is_ready());
1975
1976        let mut join_fut = std::pin::pin!(handle.join(&cx));
1977        match join_fut.as_mut().poll(&mut poll_cx) {
1978            Poll::Ready(Ok(has_timer)) => assert!(has_timer),
1979            other => unreachable!("Expected Ready(Ok(_)), got {other:?}"),
1980        }
1981    }
1982
1983    #[test]
1984    fn spawn_registered_stores_task() {
1985        let mut state = RuntimeState::new();
1986        let cx = test_cx();
1987        let region = state.create_root_region(Budget::INFINITE);
1988        let scope = test_scope(region, Budget::INFINITE);
1989
1990        // spawn_registered should both create and store the task
1991        let handle = scope
1992            .spawn_registered(&mut state, &cx, |_| async { 42_i32 })
1993            .expect("should spawn and register task");
1994
1995        // Task record should exist
1996        let task = state.task(handle.task_id());
1997        assert!(task.is_some());
1998        assert_eq!(task.expect("task should exist in state").owner, region);
1999
2000        // StoredTask should be registered (can be retrieved for polling)
2001        let stored = state.get_stored_future(handle.task_id());
2002        assert!(stored.is_some(), "spawn_registered should store the task");
2003    }
2004
2005    #[test]
2006    fn spawn_registered_task_can_be_polled() {
2007        use std::task::Context;
2008
2009        let mut state = RuntimeState::new();
2010        let cx = test_cx();
2011        let region = state.create_root_region(Budget::INFINITE);
2012        let scope = test_scope(region, Budget::INFINITE);
2013
2014        let mut handle = scope
2015            .spawn_registered(&mut state, &cx, |_| async { 42_i32 })
2016            .expect("should spawn registered task for polling test");
2017
2018        // Get the stored future and poll it
2019        let waker = std::task::Waker::noop().clone();
2020        let mut poll_cx = Context::from_waker(&waker);
2021
2022        let stored = state
2023            .get_stored_future(handle.task_id())
2024            .expect("should retrieve stored future for polling");
2025        let poll_result = stored.poll(&mut poll_cx);
2026        assert!(
2027            poll_result.is_ready(),
2028            "Simple async should complete in one poll"
2029        );
2030
2031        // Join should now have the result
2032        let mut join_fut = std::pin::pin!(handle.join(&cx));
2033        match join_fut.as_mut().poll(&mut poll_cx) {
2034            Poll::Ready(Ok(val)) => assert_eq!(val, 42),
2035            other => unreachable!("Expected Ready(Ok(42)), got {other:?}"),
2036        }
2037    }
2038
2039    #[test]
2040    fn spawn_blocking_creates_task_record() {
2041        let mut state = RuntimeState::new();
2042        let cx = test_cx();
2043        let region = state.create_root_region(Budget::INFINITE);
2044        let scope = test_scope(region, Budget::INFINITE);
2045
2046        let (handle, _stored) = scope
2047            .spawn_blocking(&mut state, &cx, |_| 42_i32)
2048            .expect("should spawn blocking task");
2049
2050        // Task should exist
2051        let task = state.task(handle.task_id());
2052        assert!(task.is_some());
2053        assert_eq!(
2054            task.expect("blocking task should exist in state").owner,
2055            region
2056        );
2057    }
2058
2059    #[test]
2060    fn spawn_local_creates_task_record() {
2061        let mut state = RuntimeState::new();
2062        let cx = test_cx();
2063        let region = state.create_root_region(Budget::INFINITE);
2064        let scope = test_scope(region, Budget::INFINITE);
2065
2066        let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
2067        let _local_ready_guard =
2068            crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
2069        let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
2070
2071        // In Phase 0, spawn_local requires Send bounds
2072        // In Phase 1+, this will work with !Send futures
2073        let handle = scope
2074            .spawn_local(&mut state, &cx, |_| async move { 42_i32 })
2075            .unwrap();
2076
2077        // Task should exist
2078        let task = state.task(handle.task_id());
2079        assert!(task.is_some());
2080        assert_eq!(task.unwrap().owner, region);
2081    }
2082
2083    #[test]
2084    fn spawn_local_without_scheduler_fails_and_rolls_back() {
2085        let mut state = RuntimeState::new();
2086        let cx = test_cx();
2087        let region = state.create_root_region(Budget::INFINITE);
2088        let scope = test_scope(region, Budget::INFINITE);
2089
2090        let result = scope.spawn_local(&mut state, &cx, |_| async move { 5_i32 });
2091        assert!(matches!(result, Err(SpawnError::LocalSchedulerUnavailable)));
2092
2093        // Task should not exist
2094        assert!(state.tasks_is_empty());
2095        let region_record = state.region(region).unwrap();
2096        assert!(region_record.task_ids().is_empty());
2097    }
2098
2099    #[test]
2100    fn spawn_local_makes_progress_via_local_ready() {
2101        use std::sync::Arc;
2102        use std::task::Context;
2103
2104        let mut state = RuntimeState::new();
2105        let cx = test_cx();
2106        let region = state.create_root_region(Budget::INFINITE);
2107        let scope = test_scope(region, Budget::INFINITE);
2108
2109        let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
2110        let _local_ready_guard =
2111            crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
2112        let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
2113
2114        let mut handle = scope
2115            .spawn_local(&mut state, &cx, |_| async move { 7_i32 })
2116            .unwrap();
2117
2118        let queued = {
2119            let queue = local_ready.lock();
2120            queue.contains(&handle.task_id())
2121        };
2122        assert!(queued, "spawn_local should enqueue into local_ready");
2123
2124        let task_id = {
2125            let mut queue = local_ready.lock();
2126            queue
2127                .pop_front()
2128                .expect("local_ready should contain spawned task")
2129        };
2130
2131        let mut join_fut = std::pin::pin!(handle.join(&cx));
2132        let waker = std::task::Waker::noop().clone();
2133        let mut ctx = Context::from_waker(&waker);
2134
2135        assert!(join_fut.as_mut().poll(&mut ctx).is_pending());
2136
2137        let mut local_task =
2138            crate::runtime::local::remove_local_task(task_id).expect("local task missing");
2139        assert!(local_task.poll(&mut ctx).is_ready());
2140
2141        match join_fut.as_mut().poll(&mut ctx) {
2142            Poll::Ready(Ok(val)) => assert_eq!(val, 7),
2143            res => unreachable!("Expected Ready(Ok(7)), got {res:?}"),
2144        }
2145    }
2146
2147    #[test]
2148    fn task_added_to_region() {
2149        let mut state = RuntimeState::new();
2150        let cx = test_cx();
2151        let region = state.create_root_region(Budget::INFINITE);
2152        let scope = test_scope(region, Budget::INFINITE);
2153
2154        let (handle, _stored) = scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2155
2156        // Check region has the task
2157        let region_record = state.region(region).unwrap();
2158        assert!(region_record.task_ids().contains(&handle.task_id()));
2159    }
2160
2161    #[test]
2162    fn multiple_spawns_create_distinct_tasks() {
2163        let mut state = RuntimeState::new();
2164        let cx = test_cx();
2165        let region = state.create_root_region(Budget::INFINITE);
2166        let scope = test_scope(region, Budget::INFINITE);
2167
2168        let (handle1, _) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2169        let (handle2, _) = scope.spawn(&mut state, &cx, |_| async { 2_i32 }).unwrap();
2170        let (handle3, _) = scope.spawn(&mut state, &cx, |_| async { 3_i32 }).unwrap();
2171
2172        // All task IDs should be different
2173        assert_ne!(handle1.task_id(), handle2.task_id());
2174        assert_ne!(handle2.task_id(), handle3.task_id());
2175        assert_ne!(handle1.task_id(), handle3.task_id());
2176
2177        // All tasks should be in the region
2178        let region_record = state.region(region).unwrap();
2179        assert!(region_record.task_ids().contains(&handle1.task_id()));
2180        assert!(region_record.task_ids().contains(&handle2.task_id()));
2181        assert!(region_record.task_ids().contains(&handle3.task_id()));
2182    }
2183
2184    #[test]
2185    fn spawn_into_closing_region_should_fail() {
2186        let mut state = RuntimeState::new();
2187        let cx = test_cx();
2188        let region = state.create_root_region(Budget::INFINITE);
2189        let scope = test_scope(region, Budget::INFINITE);
2190
2191        // Transition region to Closing
2192        let region_record = state.region_mut(region).expect("region");
2193        region_record.begin_close(None);
2194
2195        // Attempt to spawn should fail
2196        let result = scope.spawn(&mut state, &cx, |_| async { 42 });
2197        assert!(matches!(result, Err(SpawnError::RegionClosed(_))));
2198    }
2199
2200    #[test]
2201    fn test_join_manual_poll() {
2202        use std::task::Context;
2203
2204        let mut state = RuntimeState::new();
2205        let cx = test_cx();
2206        let region = state.create_root_region(Budget::INFINITE);
2207        let scope = test_scope(region, Budget::INFINITE);
2208
2209        // Spawn a task
2210        let (mut handle, mut stored_task) =
2211            scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2212        // The stored task is returned directly, not put in state by scope.spawn
2213
2214        // Create join future
2215        let mut join_fut = std::pin::pin!(handle.join(&cx));
2216
2217        // Create waker context
2218        let waker = std::task::Waker::noop().clone();
2219        let mut ctx = Context::from_waker(&waker);
2220
2221        // Poll join - should be pending
2222        assert!(join_fut.as_mut().poll(&mut ctx).is_pending());
2223
2224        // Poll stored task - should complete and send result
2225        assert!(stored_task.poll(&mut ctx).is_ready());
2226
2227        // Poll join - should be ready now
2228        match join_fut.as_mut().poll(&mut ctx) {
2229            Poll::Ready(Ok(val)) => assert_eq!(val, 42),
2230            other => unreachable!("Expected Ready(Ok(42)), got {other:?}"),
2231        }
2232    }
2233
2234    #[test]
2235    fn spawn_abort_cancels_task() {
2236        use std::task::{Context, Poll};
2237
2238        let mut state = RuntimeState::new();
2239        let cx = test_cx();
2240        let region = state.create_root_region(Budget::INFINITE);
2241        let scope = test_scope(region, Budget::INFINITE);
2242
2243        // Spawn a task that checks for cancellation
2244        let (mut handle, mut stored_task) = scope
2245            .spawn(&mut state, &cx, |cx| async move {
2246                // We expect to be cancelled immediately because abort() is called before we run
2247                if cx.checkpoint().is_err() {
2248                    return "cancelled";
2249                }
2250                "finished"
2251            })
2252            .unwrap();
2253
2254        // Abort the task via handle
2255        handle.abort();
2256
2257        // Drive the task
2258        let waker = std::task::Waker::noop().clone();
2259        let mut ctx = Context::from_waker(&waker);
2260
2261        // Task should run, see cancellation, and return "cancelled"
2262        match stored_task.poll(&mut ctx) {
2263            Poll::Ready(crate::types::Outcome::Ok(())) => {}
2264            res => unreachable!("Task should have completed with Ok(()), got {res:?}"),
2265        }
2266
2267        // Check result via handle
2268        let mut join_fut = std::pin::pin!(handle.join(&cx));
2269        match join_fut.as_mut().poll(&mut ctx) {
2270            Poll::Ready(Ok(val)) => assert_eq!(val, "cancelled"),
2271            Poll::Ready(Err(e)) => unreachable!("Task failed unexpectedly: {e}"),
2272            Poll::Pending => unreachable!("Join should be ready"),
2273        }
2274    }
2275
2276    #[test]
2277    fn hedge_backup_spawn_failure_aborts_primary() {
2278        let mut state = RuntimeState::new();
2279        let cx = test_cx();
2280        let region = state.create_root_region(Budget::INFINITE);
2281        let scope = test_scope(region, Budget::INFINITE);
2282
2283        let limits = RegionLimits {
2284            max_tasks: Some(1),
2285            ..RegionLimits::unlimited()
2286        };
2287        assert!(state.set_region_limits(region, limits));
2288
2289        let result = block_on(scope.hedge(
2290            &mut state,
2291            &cx,
2292            std::time::Duration::ZERO,
2293            |_| async { 1_u8 },
2294            |_| async { 2_u8 },
2295        ));
2296
2297        assert!(matches!(
2298            result,
2299            Err(JoinError::Cancelled(reason))
2300                if reason.kind == CancelKind::ResourceUnavailable
2301        ));
2302
2303        let task_id = *state
2304            .region(region)
2305            .expect("region missing")
2306            .task_ids()
2307            .first()
2308            .expect("primary task should remain tracked");
2309
2310        let task = state.task(task_id).expect("primary task record missing");
2311        let (cancel_requested, cancel_reason_kind) = {
2312            let inner = task
2313                .cx_inner
2314                .as_ref()
2315                .expect("primary task must have shared Cx inner")
2316                .read();
2317            (
2318                inner.cancel_requested,
2319                inner.cancel_reason.as_ref().map(|r| r.kind),
2320            )
2321        };
2322
2323        assert!(
2324            cancel_requested,
2325            "primary task must be cancellation-requested when backup spawn fails"
2326        );
2327        assert_eq!(cancel_reason_kind, Some(CancelKind::ResourceUnavailable));
2328    }
2329
2330    #[test]
2331    fn region_closes_empty_child() {
2332        let mut state = RuntimeState::new();
2333        let cx = test_cx();
2334        let parent = state.create_root_region(Budget::INFINITE);
2335        let scope = test_scope(parent, Budget::INFINITE);
2336
2337        let outcome = block_on(scope.region(
2338            &mut state,
2339            &cx,
2340            crate::types::policy::FailFast,
2341            |child, _state| {
2342                let child_id = child.region_id();
2343                async move { Outcome::Ok(child_id) }
2344            },
2345        ))
2346        .expect("child region created");
2347
2348        let child_id = match outcome {
2349            Outcome::Ok(id) => id,
2350            other => unreachable!("expected Outcome::Ok(child_id), got {other:?}"),
2351        };
2352
2353        assert!(
2354            state.region(child_id).is_none(),
2355            "closed child region should be reclaimed from arena"
2356        );
2357
2358        let parent_record = state.region(parent).expect("parent record missing");
2359        assert!(
2360            !parent_record.child_ids().contains(&child_id),
2361            "closed child should be removed from parent"
2362        );
2363    }
2364
2365    #[test]
2366    fn region_budget_is_met_with_parent() {
2367        let mut state = RuntimeState::new();
2368        let cx = test_cx();
2369        let parent = state.create_root_region(Budget::with_deadline_secs(10));
2370        let scope = test_scope(parent, Budget::with_deadline_secs(10));
2371
2372        let outcome = block_on(scope.region_with_budget(
2373            &mut state,
2374            &cx,
2375            Budget::with_deadline_secs(30),
2376            crate::types::policy::FailFast,
2377            |child, _state| {
2378                let child_id = child.region_id();
2379                let child_budget = child.budget();
2380                async move { Outcome::Ok((child_id, child_budget)) }
2381            },
2382        ))
2383        .expect("child region created");
2384
2385        let (child_id, child_budget) = match outcome {
2386            Outcome::Ok(tuple) => tuple,
2387            other => unreachable!("expected Outcome::Ok(child_id), got {other:?}"),
2388        };
2389
2390        assert_eq!(
2391            child_budget.deadline,
2392            Some(crate::types::Time::from_secs(10))
2393        );
2394        assert!(
2395            state.region(child_id).is_none(),
2396            "closed child region should be reclaimed from arena"
2397        );
2398    }
2399
2400    #[test]
2401    fn region_with_priority_uses_requested_admission_priority() {
2402        let mut state = RuntimeState::new();
2403        let cx = test_cx();
2404        let parent = state.create_root_region(Budget::INFINITE);
2405        let scope = test_scope(parent, Budget::INFINITE);
2406
2407        state
2408            .resource_monitor()
2409            .pressure()
2410            .update_degradation_level(
2411                crate::runtime::resource_monitor::ResourceType::Memory,
2412                crate::runtime::resource_monitor::DegradationLevel::Moderate,
2413            );
2414
2415        block_on(scope.region(
2416            &mut state,
2417            &cx,
2418            crate::types::policy::FailFast,
2419            |child, _state| {
2420                let child_id = child.region_id();
2421                async move { Outcome::Ok(child_id) }
2422            },
2423        ))
2424        .expect("normal child scope should be admitted at moderate pressure");
2425
2426        let err = block_on(scope.region_with_priority(
2427            &mut state,
2428            &cx,
2429            RegionPriority::Low,
2430            crate::types::policy::FailFast,
2431            |_child, _state| async move { Outcome::Ok(()) },
2432        ))
2433        .expect_err("low-priority child scope should be rejected at moderate pressure");
2434
2435        assert!(matches!(
2436            err,
2437            RegionCreateError::ResourcePressure {
2438                requested_priority: RegionPriority::Low,
2439                ..
2440            }
2441        ));
2442    }
2443
2444    #[test]
2445    fn region_with_priority_registers_child_shedding_priority() {
2446        let mut state = RuntimeState::new();
2447        let cx = test_cx();
2448        let parent = state.create_root_region(Budget::INFINITE);
2449        let scope = test_scope(parent, Budget::INFINITE);
2450
2451        let outcome = block_on(scope.region_with_priority(
2452            &mut state,
2453            &cx,
2454            RegionPriority::Low,
2455            crate::types::policy::FailFast,
2456            |child, state| {
2457                let child_id = child.region_id();
2458                state
2459                    .resource_monitor()
2460                    .pressure()
2461                    .update_degradation_level(
2462                        crate::runtime::resource_monitor::ResourceType::Memory,
2463                        crate::runtime::resource_monitor::DegradationLevel::Heavy,
2464                    );
2465                let decision = state
2466                    .resource_monitor()
2467                    .engine()
2468                    .should_shed_region(child_id);
2469                async move { Outcome::Ok(decision) }
2470            },
2471        ))
2472        .expect("low-priority child scope should be admitted before pressure rises");
2473
2474        assert!(matches!(
2475            outcome,
2476            Outcome::Ok(crate::runtime::resource_monitor::SheddingDecision::Pause)
2477        ));
2478    }
2479
2480    #[test]
2481    fn region_capability_budget_is_met_with_parent() {
2482        let mut state = RuntimeState::new();
2483        let cx = test_cx();
2484        let parent_budget = CapabilityBudget::new()
2485            .with_memory_bytes(1_024)
2486            .with_io_bytes(8_192);
2487        let parent =
2488            state.create_root_region_with_capability_budget(Budget::INFINITE, parent_budget);
2489        let scope = test_scope_with_capability_budget(parent, Budget::INFINITE, parent_budget);
2490
2491        let outcome = block_on(
2492            scope.region_with_budget_and_capability_budget(
2493                &mut state,
2494                &cx,
2495                Budget::INFINITE,
2496                CapabilityBudget::new()
2497                    .with_memory_bytes(4_096)
2498                    .with_io_bytes(512),
2499                CapabilityBudgetRequirements::new()
2500                    .require_memory_bytes()
2501                    .require_io_bytes(),
2502                crate::types::policy::FailFast,
2503                |child, _state| {
2504                    let child_budget = child.capability_budget();
2505                    async move { Outcome::Ok(child_budget) }
2506                },
2507            ),
2508        )
2509        .expect("child region created");
2510
2511        let child_budget = match outcome {
2512            Outcome::Ok(budget) => budget,
2513            other => unreachable!("expected Outcome::Ok(capability_budget), got {other:?}"),
2514        };
2515
2516        assert_eq!(child_budget.memory_bytes, Some(1_024));
2517        assert_eq!(child_budget.io_bytes, Some(512));
2518    }
2519
2520    #[test]
2521    fn region_capability_budget_required_absent_dimension_rejects() {
2522        let mut state = RuntimeState::new();
2523        let cx = test_cx();
2524        let parent = state.create_root_region(Budget::INFINITE);
2525        let scope = test_scope(parent, Budget::INFINITE);
2526
2527        let err = block_on(scope.region_with_budget_and_capability_budget(
2528            &mut state,
2529            &cx,
2530            Budget::INFINITE,
2531            CapabilityBudget::new(),
2532            CapabilityBudgetRequirements::new().require_artifact_bytes(),
2533            crate::types::policy::FailFast,
2534            |_child, _state| async move { Outcome::Ok(()) },
2535        ))
2536        .expect_err("missing required artifact budget must fail closed");
2537
2538        assert!(matches!(
2539            err,
2540            RegionCreateError::CapabilityBudgetRefused {
2541                parent: refused_parent,
2542                reason: CapabilityBudgetRefusal::MissingRequired(
2543                    CapabilityBudgetDimension::ArtifactBytes
2544                ),
2545            } if refused_parent == parent
2546        ));
2547    }
2548
2549    #[test]
2550    fn spawned_task_cx_inherits_scope_capability_budget() {
2551        let mut state = RuntimeState::new();
2552        let cx = test_cx();
2553        let capability_budget = CapabilityBudget::new()
2554            .with_memory_bytes(4_096)
2555            .with_cpu_units(16);
2556        let region =
2557            state.create_root_region_with_capability_budget(Budget::INFINITE, capability_budget);
2558        let scope = test_scope_with_capability_budget(region, Budget::INFINITE, capability_budget);
2559
2560        let (handle, _stored) = scope
2561            .spawn(&mut state, &cx, |child_cx| async move {
2562                child_cx.capability_budget()
2563            })
2564            .expect("spawn should admit task");
2565
2566        let task = state.task(handle.task_id()).expect("task record missing");
2567        let actual = task
2568            .cx_inner
2569            .as_ref()
2570            .expect("task cx inner missing")
2571            .read()
2572            .capability_budget;
2573
2574        assert_eq!(actual, capability_budget);
2575    }
2576
2577    #[test]
2578    fn region_spawns_tasks_in_child() {
2579        use std::task::{Context, Poll};
2580
2581        let mut state = RuntimeState::new();
2582        let cx = test_cx();
2583        let parent = state.create_root_region(Budget::INFINITE);
2584        let scope = test_scope(parent, Budget::INFINITE);
2585
2586        let outcome = block_on(scope.region(
2587            &mut state,
2588            &cx,
2589            crate::types::policy::FailFast,
2590            |child, state| {
2591                let child_id = child.region_id();
2592                let (handle, mut stored) = child
2593                    .spawn(state, &cx, |_| async { 7_i32 })
2594                    .expect("spawn in child");
2595
2596                let parent_has = state
2597                    .region(parent)
2598                    .expect("parent record missing")
2599                    .task_ids()
2600                    .contains(&handle.task_id());
2601                let child_has = state
2602                    .region(child_id)
2603                    .expect("child record missing")
2604                    .task_ids()
2605                    .contains(&handle.task_id());
2606
2607                let waker = std::task::Waker::noop().clone();
2608                let mut poll_cx = Context::from_waker(&waker);
2609                let poll_result = stored.poll(&mut poll_cx);
2610                if let Poll::Ready(outcome) = poll_result {
2611                    let task_outcome = match outcome {
2612                        Outcome::Ok(()) => Outcome::Ok(()),
2613                        Outcome::Panicked(payload) => Outcome::Panicked(payload),
2614                        other => unreachable!("unexpected task outcome: {other:?}"),
2615                    };
2616                    if let Some(task_record) = state.task_mut(handle.task_id()) {
2617                        task_record.complete(task_outcome);
2618                    }
2619                    let _ = state.task_completed(handle.task_id());
2620                }
2621
2622                std::future::ready(Outcome::Ok((child_id, parent_has, child_has)))
2623            },
2624        ))
2625        .expect("child region created");
2626
2627        let (child_id, parent_has, child_has) = match outcome {
2628            Outcome::Ok(tuple) => tuple,
2629            other => unreachable!("expected Outcome::Ok(tuple), got {other:?}"),
2630        };
2631
2632        assert!(!parent_has, "task should not be owned by parent region");
2633        assert!(child_has, "task should be owned by child region");
2634
2635        let parent_record = state.region(parent).expect("parent record missing");
2636        assert!(
2637            !parent_record.child_ids().contains(&child_id),
2638            "closed child should be removed from parent"
2639        );
2640    }
2641
2642    #[test]
2643    fn spawn_panic_propagates_as_panicked_error() {
2644        use std::task::{Context, Poll};
2645
2646        let mut state = RuntimeState::new();
2647        let cx = test_cx();
2648        let region = state.create_root_region(Budget::INFINITE);
2649        let scope = test_scope(region, Budget::INFINITE);
2650
2651        let (mut handle, mut stored_task) = scope
2652            .spawn(&mut state, &cx, |_| async {
2653                std::panic::panic_any("oops");
2654            })
2655            .unwrap();
2656
2657        // Drive the task
2658        let waker = std::task::Waker::noop().clone();
2659        let mut ctx = Context::from_waker(&waker);
2660
2661        // Polling stored task should return Ready(Panicked) even if it panics (caught inside)
2662        match stored_task.poll(&mut ctx) {
2663            Poll::Ready(crate::types::Outcome::Panicked(_)) => {}
2664            res => unreachable!("Task should have completed with Panicked, got {res:?}"),
2665        }
2666
2667        // Check result via handle
2668        let mut join_fut = std::pin::pin!(handle.join(&cx));
2669        match join_fut.as_mut().poll(&mut ctx) {
2670            Poll::Ready(Err(JoinError::Panicked(p))) => {
2671                assert_eq!(p.message(), "oops");
2672            }
2673            res => unreachable!("Expected Panicked, got {res:?}"),
2674        }
2675    }
2676
2677    #[test]
2678    fn join_all_success() {
2679        use std::task::{Context, Poll};
2680
2681        let mut state = RuntimeState::new();
2682        let cx = test_cx();
2683        let region = state.create_root_region(Budget::INFINITE);
2684        let scope = test_scope(region, Budget::INFINITE);
2685
2686        let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1 }).unwrap();
2687        let (h2, mut t2) = scope.spawn(&mut state, &cx, |_| async { 2 }).unwrap();
2688
2689        // Drive tasks to completion
2690        let waker = std::task::Waker::noop().clone();
2691        let mut ctx = Context::from_waker(&waker);
2692        assert!(t1.poll(&mut ctx).is_ready());
2693        assert!(t2.poll(&mut ctx).is_ready());
2694
2695        let handles = vec![h1, h2];
2696        let mut fut = Box::pin(scope.join_all(&cx, handles));
2697
2698        match fut.as_mut().poll(&mut ctx) {
2699            Poll::Ready(results) => {
2700                assert_eq!(results.len(), 2);
2701                assert_eq!(results[0].as_ref().unwrap(), &1);
2702                assert_eq!(results[1].as_ref().unwrap(), &2);
2703            }
2704            Poll::Pending => unreachable!("join_all should be ready"),
2705        }
2706    }
2707
2708    #[test]
2709    fn race_all_aborted_task_is_drained() {
2710        use std::task::{Context, Poll};
2711
2712        let mut state = RuntimeState::new();
2713        let cx = test_cx();
2714        let region = state.create_root_region(Budget::INFINITE);
2715        let scope = test_scope(region, Budget::INFINITE);
2716
2717        // Task 1: completes immediately
2718        let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1 }).unwrap();
2719
2720        // Task 2: yields once, checking for cancellation
2721        let (h2, mut t2) = scope
2722            .spawn(&mut state, &cx, |cx| async move {
2723                // Yield once to simulate running
2724                struct YieldOnce(bool);
2725                impl std::future::Future for YieldOnce {
2726                    type Output = ();
2727                    fn poll(
2728                        mut self: std::pin::Pin<&mut Self>,
2729                        cx: &mut std::task::Context<'_>,
2730                    ) -> std::task::Poll<()> {
2731                        if self.0 {
2732                            std::task::Poll::Ready(())
2733                        } else {
2734                            self.0 = true;
2735                            cx.waker().wake_by_ref();
2736                            std::task::Poll::Pending
2737                        }
2738                    }
2739                }
2740                YieldOnce(false).await;
2741
2742                // Check cancellation
2743                if cx.checkpoint().is_err() {
2744                    return 0; // Cancelled
2745                }
2746                2
2747            })
2748            .unwrap();
2749
2750        let waker = std::task::Waker::noop().clone();
2751        let mut ctx = Context::from_waker(&waker);
2752
2753        // Drive t1 to completion (winner)
2754        assert!(t1.poll(&mut ctx).is_ready());
2755
2756        // Initialize race_all
2757        let handles = vec![h1, h2];
2758        let mut race_fut = Box::pin(scope.race_all(&cx, handles));
2759
2760        // Poll race_all.
2761        // It sees h1 ready. Winner=0.
2762        // It aborts h2.
2763        // It awaits h2 drain.
2764        // h2 is still pending (hasn't run), so h2.join() returns Pending.
2765        // race_fut returns Pending.
2766        assert!(race_fut.as_mut().poll(&mut ctx).is_pending());
2767
2768        // Now drive t2. It was aborted, so it should see cancellation if checked?
2769        // Wait, handle.abort() sets inner.cancel_requested.
2770        // But my t2 closure yields first.
2771        // So first poll of t2 -> YieldOnce returns Pending.
2772        assert!(t2.poll(&mut ctx).is_pending());
2773
2774        // Poll race_fut again. Still waiting for h2 drain.
2775        assert!(race_fut.as_mut().poll(&mut ctx).is_pending());
2776
2777        // Poll t2 again. YieldOnce finishes.
2778        // Then it hits checkpoint(). cancel_requested is true.
2779        // It returns 0 (simulated cancellation return).
2780        // Actually, normally tasks return Result or are wrapped.
2781        // Here spawn returns Result<i32>.
2782        // My closure returns i32.
2783        // So h2.join() will return Ok(0).
2784        // This counts as "drained".
2785        assert!(t2.poll(&mut ctx).is_ready());
2786
2787        // Now poll race_fut. h2 drain complete.
2788        // Should return (1, 0).
2789        match race_fut.as_mut().poll(&mut ctx) {
2790            Poll::Ready(Ok((val, idx))) => {
2791                assert_eq!(val, 1);
2792                assert_eq!(idx, 0);
2793            }
2794            res => unreachable!("Expected Ready(Ok((1, 0))), got {res:?}"),
2795        }
2796    }
2797
2798    #[test]
2799    fn race_surfaces_loser_panic_even_if_winner_succeeds() {
2800        use std::task::Context;
2801
2802        let mut state = RuntimeState::new();
2803        let cx = test_cx();
2804        let region = state.create_root_region(Budget::INFINITE);
2805        let scope = test_scope(region, Budget::INFINITE);
2806
2807        let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2808        let (h2, mut t2) = scope
2809            .spawn(&mut state, &cx, |_| async {
2810                std::panic::panic_any("loser panic");
2811            })
2812            .unwrap();
2813
2814        let waker = std::task::Waker::noop().clone();
2815        let mut poll_cx = Context::from_waker(&waker);
2816        assert!(t1.poll(&mut poll_cx).is_ready());
2817        assert!(t2.poll(&mut poll_cx).is_ready());
2818
2819        let result = block_on(scope.race(&cx, h1, h2));
2820        assert!(
2821            matches!(result, Err(JoinError::Panicked(_))),
2822            "loser panic must dominate race result, got {result:?}"
2823        );
2824    }
2825
2826    #[test]
2827    fn race_preserves_winner_panic_over_loser_panic() {
2828        use std::task::{Context, Poll};
2829
2830        let mut state = RuntimeState::new();
2831        let cx = test_cx();
2832        let region = state.create_root_region(Budget::INFINITE);
2833        let scope = test_scope(region, Budget::INFINITE);
2834
2835        let (h1, mut t1) = scope
2836            .spawn(&mut state, &cx, |_| async {
2837                std::panic::panic_any("winner panic");
2838            })
2839            .unwrap();
2840        let (h2, mut t2) = scope
2841            .spawn(&mut state, &cx, |_| {
2842                let mut first_poll = true;
2843                std::future::poll_fn(move |poll_cx| {
2844                    if first_poll {
2845                        first_poll = false;
2846                        poll_cx.waker().wake_by_ref();
2847                        Poll::Pending
2848                    } else {
2849                        std::panic::panic_any("loser panic");
2850                    }
2851                })
2852            })
2853            .unwrap();
2854
2855        let waker = std::task::Waker::noop().clone();
2856        let mut poll_cx = Context::from_waker(&waker);
2857        assert!(t1.poll(&mut poll_cx).is_ready());
2858        assert!(t2.poll(&mut poll_cx).is_pending());
2859
2860        let result = block_on(scope.race(&cx, h1, h2));
2861        match result {
2862            Err(JoinError::Panicked(payload)) => {
2863                assert_eq!(payload.message(), "winner panic");
2864            }
2865            other => unreachable!("winner panic must dominate race result, got {other:?}"),
2866        }
2867    }
2868
2869    #[test]
2870    fn race_all_surfaces_simultaneous_loser_panic() {
2871        use std::task::Context;
2872
2873        let mut state = RuntimeState::new();
2874        let cx = test_cx();
2875        let region = state.create_root_region(Budget::INFINITE);
2876        let scope = test_scope(region, Budget::INFINITE);
2877
2878        let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 1_i32 }).unwrap();
2879        let (h2, mut t2) = scope
2880            .spawn(&mut state, &cx, |_| async {
2881                std::panic::panic_any("simultaneous loser panic");
2882            })
2883            .unwrap();
2884        let (h3, mut t3) = scope.spawn(&mut state, &cx, |_| async { 3_i32 }).unwrap();
2885
2886        let waker = std::task::Waker::noop().clone();
2887        let mut poll_cx = Context::from_waker(&waker);
2888        assert!(t1.poll(&mut poll_cx).is_ready());
2889        assert!(t2.poll(&mut poll_cx).is_ready());
2890        assert!(t3.poll(&mut poll_cx).is_ready());
2891
2892        let result = block_on(scope.race_all(&cx, vec![h1, h2, h3]));
2893        assert!(
2894            matches!(result, Err(JoinError::Panicked(_))),
2895            "simultaneous loser panic must dominate race_all result, got {result:?}"
2896        );
2897    }
2898
2899    #[test]
2900    fn race_all_preserves_winner_panic_over_loser_panic() {
2901        use std::task::{Context, Poll};
2902
2903        let mut state = RuntimeState::new();
2904        let cx = test_cx();
2905        let region = state.create_root_region(Budget::INFINITE);
2906        let scope = test_scope(region, Budget::INFINITE);
2907
2908        let (h1, mut t1) = scope
2909            .spawn(&mut state, &cx, |_| async {
2910                std::panic::panic_any("winner panic");
2911            })
2912            .unwrap();
2913        let (h2, mut t2) = scope
2914            .spawn(&mut state, &cx, |_| {
2915                let mut first_poll = true;
2916                std::future::poll_fn(move |poll_cx| {
2917                    if first_poll {
2918                        first_poll = false;
2919                        poll_cx.waker().wake_by_ref();
2920                        Poll::Pending
2921                    } else {
2922                        std::panic::panic_any("loser panic");
2923                    }
2924                })
2925            })
2926            .unwrap();
2927
2928        let waker = std::task::Waker::noop().clone();
2929        let mut poll_cx = Context::from_waker(&waker);
2930        assert!(t1.poll(&mut poll_cx).is_ready());
2931        assert!(t2.poll(&mut poll_cx).is_pending());
2932
2933        let result = block_on(scope.race_all(&cx, vec![h1, h2]));
2934        match result {
2935            Err(JoinError::Panicked(payload)) => {
2936                assert_eq!(payload.message(), "winner panic");
2937            }
2938            other => unreachable!("winner panic must dominate race_all result, got {other:?}"),
2939        }
2940    }
2941
2942    #[test]
2943    fn race_all_empty_is_pending() {
2944        let mut state = RuntimeState::new();
2945        let cx = test_cx();
2946        let region = state.create_root_region(Budget::INFINITE);
2947        let scope = test_scope(region, Budget::INFINITE);
2948
2949        let fut = scope.race_all::<i32>(&cx, vec![]);
2950        let waker = std::task::Waker::noop();
2951        let mut poll_cx = std::task::Context::from_waker(waker);
2952        let pinned = std::pin::pin!(fut);
2953        let status = std::future::Future::poll(pinned, &mut poll_cx);
2954        assert!(status.is_pending());
2955    }
2956
2957    // =============================================================================
2958    // CONFORMANCE TESTS: Structured Concurrency Invariants
2959    // =============================================================================
2960    //
2961    // These tests verify the core structured concurrency guarantees that must hold
2962    // for the spawn/join contract to be sound.
2963
2964    #[test]
2965    fn conformance_spawn_creates_trackable_task() {
2966        // INVARIANT: Every spawned task creates a trackable record that belongs to the spawning region
2967        let mut state = RuntimeState::new();
2968        let cx = test_cx();
2969        let region = state.create_root_region(Budget::INFINITE);
2970        let scope = test_scope(region, Budget::INFINITE);
2971
2972        let (handle, _stored) = scope.spawn(&mut state, &cx, |_| async { 42_i32 }).unwrap();
2973
2974        // Task must exist and be trackable
2975        let task_record = state
2976            .task(handle.task_id())
2977            .expect("spawned task must have a record");
2978
2979        // Task must belong to the spawning region
2980        assert_eq!(
2981            task_record.owner, region,
2982            "spawned task must be owned by the spawning region"
2983        );
2984
2985        // Region must track the task
2986        let region_record = state.region(region).expect("spawning region must exist");
2987        assert!(
2988            region_record.task_ids().contains(&handle.task_id()),
2989            "spawning region must track the spawned task"
2990        );
2991    }
2992
2993    #[test]
2994    fn conformance_spawn_enforces_send_bounds() {
2995        // INVARIANT: spawn() enforces Send + 'static bounds for cross-worker task migration
2996        let mut state = RuntimeState::new();
2997        let cx = test_cx();
2998        let region = state.create_root_region(Budget::INFINITE);
2999        let scope = test_scope(region, Budget::INFINITE);
3000
3001        // This should compile - Send + 'static data
3002        let send_data = String::from("test");
3003        let (handle, _stored) = scope
3004            .spawn(&mut state, &cx, move |_| async move {
3005                send_data.len() // Uses Send + 'static String
3006            })
3007            .unwrap();
3008
3009        // Task record should reflect Send bounds
3010        let task_record = state
3011            .task(handle.task_id())
3012            .expect("Send task must have a record");
3013        assert_eq!(task_record.owner, region);
3014
3015        // NOTE: Non-Send compile-time test examples are in module docstring
3016        // (compile_fail tests with Rc<T> and borrowed references)
3017    }
3018
3019    #[test]
3020    fn conformance_join_awaits_task_completion() {
3021        // INVARIANT: TaskHandle.join() waits for task completion and returns the result
3022
3023        use std::task::Context;
3024
3025        let mut state = RuntimeState::new();
3026        let cx = test_cx();
3027        let region = state.create_root_region(Budget::INFINITE);
3028        let scope = test_scope(region, Budget::INFINITE);
3029
3030        let (mut handle, mut stored) = scope.spawn(&mut state, &cx, |_| async { 123_i32 }).unwrap();
3031
3032        let waker = std::task::Waker::noop().clone();
3033        let mut poll_cx = Context::from_waker(&waker);
3034
3035        // Before task completion, join should be pending
3036        let mut join_fut = std::pin::pin!(handle.join(&cx));
3037        assert!(
3038            join_fut.as_mut().poll(&mut poll_cx).is_pending(),
3039            "join must be pending before task completion"
3040        );
3041
3042        // Complete the task
3043        assert!(
3044            stored.poll(&mut poll_cx).is_ready(),
3045            "test task must complete in one poll"
3046        );
3047
3048        // After task completion, join should return the result
3049        match join_fut.as_mut().poll(&mut poll_cx) {
3050            std::task::Poll::Ready(Ok(result)) => {
3051                assert_eq!(result, 123, "join must return the task's result");
3052            }
3053            other => panic!("join must be Ready(Ok(123)) after task completion, got {other:?}"),
3054        }
3055    }
3056
3057    #[test]
3058    fn conformance_child_region_task_isolation() {
3059        // INVARIANT: Tasks spawned in child regions belong to the child, not the parent
3060
3061        use std::task::Context;
3062
3063        let mut state = RuntimeState::new();
3064        let cx = test_cx();
3065        let parent_region = state.create_root_region(Budget::INFINITE);
3066        let scope = test_scope(parent_region, Budget::INFINITE);
3067
3068        let outcome = block_on(scope.region(
3069            &mut state,
3070            &cx,
3071            crate::types::policy::FailFast,
3072            |child_scope, state| {
3073                let child_region = child_scope.region_id();
3074
3075                // Spawn task in child region
3076                let (handle, mut stored) = child_scope
3077                    .spawn(state, &cx, |_| async { 456_i32 })
3078                    .expect("spawn in child region must succeed");
3079
3080                // Verify task ownership invariants
3081                let task_record = state
3082                    .task(handle.task_id())
3083                    .expect("child task must have a record");
3084                let child_owns = task_record.owner == child_region;
3085                let parent_owns = task_record.owner == parent_region;
3086
3087                // Verify region tracking invariants
3088                let parent_tracks = state
3089                    .region(parent_region)
3090                    .is_some_and(|r| r.task_ids().contains(&handle.task_id()));
3091                let child_tracks = state
3092                    .region(child_region)
3093                    .is_some_and(|r| r.task_ids().contains(&handle.task_id()));
3094
3095                // Complete the task for clean shutdown
3096                let waker = std::task::Waker::noop().clone();
3097                let mut poll_cx = Context::from_waker(&waker);
3098                if let std::task::Poll::Ready(outcome) = stored.poll(&mut poll_cx) {
3099                    if let Some(task) = state.task_mut(handle.task_id()) {
3100                        task.complete(outcome.map_err(|_| {
3101                            crate::error::Error::new(crate::error::ErrorKind::Internal)
3102                        }));
3103                    }
3104                    let _ = state.task_completed(handle.task_id());
3105                }
3106
3107                std::future::ready(Outcome::Ok((
3108                    child_owns,
3109                    parent_owns,
3110                    child_tracks,
3111                    parent_tracks,
3112                )))
3113            },
3114        ))
3115        .expect("child region must complete");
3116
3117        let (child_owns, parent_owns, child_tracks, parent_tracks) = match outcome {
3118            Outcome::Ok(tuple) => tuple,
3119            other => panic!("expected Ok(ownership_data), got {other:?}"),
3120        };
3121
3122        assert!(
3123            child_owns,
3124            "task spawned in child region must be owned by child"
3125        );
3126        assert!(
3127            !parent_owns,
3128            "task spawned in child region must NOT be owned by parent"
3129        );
3130        assert!(child_tracks, "child region must track its spawned tasks");
3131        assert!(!parent_tracks, "parent region must NOT track child's tasks");
3132    }
3133
3134    #[test]
3135    fn conformance_capability_inheritance() {
3136        // INVARIANT: Spawned tasks inherit capabilities from the parent Cx
3137        use crate::cx::macaroon::MacaroonToken;
3138        use crate::cx::registry::RegistryHandle;
3139        use crate::remote::{NodeId, RemoteCap};
3140        use crate::security::key::AuthKey;
3141        use crate::types::SystemPressure;
3142        use std::sync::Arc;
3143        use std::task::Context;
3144
3145        let mut state = RuntimeState::new();
3146        let region = state.create_root_region(Budget::INFINITE);
3147        let scope = test_scope(region, Budget::INFINITE);
3148
3149        // Setup parent Cx with capabilities
3150        let registry = crate::cx::NameRegistry::new();
3151        let registry_handle = RegistryHandle::new(Arc::new(registry));
3152        let parent_registry_arc = registry_handle.as_arc();
3153        let parent_io_cap: Arc<dyn crate::io::IoCap> =
3154            Arc::new(crate::io::LabIoCap::new_for_tests());
3155        let parent_pressure = Arc::new(SystemPressure::new());
3156        parent_pressure.set_headroom(0.25);
3157        let auth_key = AuthKey::from_seed(7);
3158        let token = MacaroonToken::mint(&auth_key, "scope:spawn", "cx/scope");
3159
3160        let parent_cx = Cx::new_with_io(
3161            crate::types::RegionId::new_for_test(0, 0),
3162            crate::types::TaskId::new_for_test(0, 0),
3163            Budget::INFINITE,
3164            None,
3165            None,
3166            Some(Arc::clone(&parent_io_cap)),
3167            None,
3168        )
3169        .with_registry_handle(Some(registry_handle))
3170        .with_remote_cap(RemoteCap::new().with_local_node(NodeId::new("test-node")))
3171        .with_pressure(Arc::clone(&parent_pressure))
3172        .with_macaroon(token);
3173        let parent_macaroon = parent_cx
3174            .macaroon_handle()
3175            .expect("parent must retain macaroon capability");
3176
3177        let mut handle = scope
3178            .spawn_registered(&mut state, &parent_cx, move |child_cx| async move {
3179                // Verify registry inheritance
3180                let child_registry = child_cx
3181                    .registry_handle()
3182                    .expect("child must inherit registry capability");
3183                let same_registry = Arc::ptr_eq(&child_registry.as_arc(), &parent_registry_arc);
3184
3185                // Verify remote capability inheritance
3186                let child_remote = child_cx
3187                    .remote()
3188                    .expect("child must inherit remote capability");
3189                let node_name = child_remote.local_node().as_str().to_owned();
3190
3191                // Verify I/O capability inheritance
3192                let child_io_cap = child_cx
3193                    .io_cap_handle()
3194                    .expect("child must inherit I/O capability");
3195                let same_io_cap = Arc::ptr_eq(&child_io_cap, &parent_io_cap);
3196
3197                // Verify system pressure inheritance
3198                let child_pressure = child_cx
3199                    .pressure_handle()
3200                    .expect("child must inherit system pressure");
3201                let same_pressure = Arc::ptr_eq(&child_pressure, &parent_pressure);
3202
3203                // Verify macaroon inheritance
3204                let child_macaroon = child_cx
3205                    .macaroon_handle()
3206                    .expect("child must inherit macaroon capability");
3207                let same_macaroon = Arc::ptr_eq(&child_macaroon, &parent_macaroon);
3208
3209                // Verify timer capability inheritance (if parent has it)
3210                let has_timer = child_cx.has_timer();
3211
3212                (
3213                    same_registry,
3214                    node_name,
3215                    same_io_cap,
3216                    same_pressure,
3217                    same_macaroon,
3218                    has_timer,
3219                )
3220            })
3221            .unwrap();
3222
3223        // Complete the task and verify results
3224        let waker = std::task::Waker::noop().clone();
3225        let mut poll_cx = Context::from_waker(&waker);
3226
3227        let stored = state
3228            .get_stored_future(handle.task_id())
3229            .expect("spawn_registered must store the task");
3230        assert!(stored.poll(&mut poll_cx).is_ready());
3231
3232        let mut join_fut = std::pin::pin!(handle.join(&parent_cx));
3233        match join_fut.as_mut().poll(&mut poll_cx) {
3234            std::task::Poll::Ready(Ok((
3235                same_registry,
3236                node_name,
3237                same_io_cap,
3238                same_pressure,
3239                same_macaroon,
3240                has_timer,
3241            ))) => {
3242                assert!(
3243                    same_registry,
3244                    "child must inherit exact same registry instance"
3245                );
3246                assert_eq!(
3247                    node_name, "test-node",
3248                    "child must inherit remote capability"
3249                );
3250                assert!(same_io_cap, "child must inherit exact same I/O capability");
3251                assert!(
3252                    same_pressure,
3253                    "child must inherit exact same system pressure handle"
3254                );
3255                assert!(
3256                    same_macaroon,
3257                    "child must inherit exact same macaroon capability"
3258                );
3259                assert_eq!(
3260                    has_timer,
3261                    parent_cx.has_timer(),
3262                    "child timer capability should stay consistent with the runtime-backed parent"
3263                );
3264            }
3265            other => panic!("capability inheritance test failed: {other:?}"),
3266        }
3267    }
3268
3269    #[test]
3270    fn conformance_task_cancellation_propagation() {
3271        // INVARIANT: Cancelling a task via abort() propagates cancellation signal
3272
3273        use std::task::Context;
3274
3275        let mut state = RuntimeState::new();
3276        let cx = test_cx();
3277        let region = state.create_root_region(Budget::INFINITE);
3278        let scope = test_scope(region, Budget::INFINITE);
3279
3280        let (mut handle, mut stored) = scope
3281            .spawn(&mut state, &cx, |cx| async move {
3282                // Check cancellation status and respond accordingly
3283                if cx.checkpoint().is_err() {
3284                    "cancelled"
3285                } else {
3286                    "completed"
3287                }
3288            })
3289            .unwrap();
3290
3291        // Abort the task before it runs
3292        handle.abort();
3293
3294        let waker = std::task::Waker::noop().clone();
3295        let mut poll_cx = Context::from_waker(&waker);
3296
3297        // Task should complete and see the cancellation
3298        assert!(
3299            stored.poll(&mut poll_cx).is_ready(),
3300            "cancelled task must still complete"
3301        );
3302
3303        // Join should return the cancellation-aware result
3304        let mut join_fut = std::pin::pin!(handle.join(&cx));
3305        match join_fut.as_mut().poll(&mut poll_cx) {
3306            std::task::Poll::Ready(Ok(result)) => {
3307                assert_eq!(
3308                    result, "cancelled",
3309                    "cancelled task must observe cancellation via checkpoint()"
3310                );
3311            }
3312            other => panic!("cancelled task join failed: {other:?}"),
3313        }
3314    }
3315
3316    #[test]
3317    fn metamorphic_nested_scope_cancellation_closes_descendants_without_spawn_leaks() {
3318        use std::task::{Context, Poll};
3319
3320        struct YieldOnce(bool);
3321
3322        impl std::future::Future for YieldOnce {
3323            type Output = ();
3324
3325            fn poll(
3326                mut self: std::pin::Pin<&mut Self>,
3327                cx: &mut std::task::Context<'_>,
3328            ) -> Poll<()> {
3329                if self.0 {
3330                    Poll::Ready(())
3331                } else {
3332                    self.0 = true;
3333                    cx.waker().wake_by_ref();
3334                    Poll::Pending
3335                }
3336            }
3337        }
3338
3339        let mut state = RuntimeState::new();
3340        let cx = test_cx();
3341        let root = state.create_root_region(Budget::INFINITE);
3342        let child = state
3343            .create_child_region(root, Budget::INFINITE)
3344            .expect("child region");
3345        let grandchild = state
3346            .create_child_region(child, Budget::INFINITE)
3347            .expect("grandchild region");
3348        let child_scope = test_scope(child, Budget::INFINITE);
3349        let grandchild_scope = test_scope(grandchild, Budget::INFINITE);
3350        let finalizer_log = Arc::new(std::sync::Mutex::new(Vec::new()));
3351
3352        let child_log = Arc::clone(&finalizer_log);
3353        assert!(
3354            child_scope.defer_sync(&mut state, move || {
3355                child_log
3356                    .lock()
3357                    .expect("child finalizer log poisoned")
3358                    .push("child");
3359            }),
3360            "child finalizer should register before cancellation"
3361        );
3362
3363        let grandchild_log = Arc::clone(&finalizer_log);
3364        assert!(
3365            grandchild_scope.defer_sync(&mut state, move || {
3366                grandchild_log
3367                    .lock()
3368                    .expect("grandchild finalizer log poisoned")
3369                    .push("grandchild");
3370            }),
3371            "grandchild finalizer should register before cancellation"
3372        );
3373
3374        let mut child_handle = child_scope
3375            .spawn_registered(&mut state, &cx, |task_cx| async move {
3376                YieldOnce(false).await;
3377                if task_cx.checkpoint().is_err() {
3378                    "child_cancelled"
3379                } else {
3380                    "child_completed"
3381                }
3382            })
3383            .expect("spawn child task");
3384        let child_task_id = child_handle.task_id();
3385
3386        let mut grandchild_handle = grandchild_scope
3387            .spawn_registered(&mut state, &cx, |task_cx| async move {
3388                YieldOnce(false).await;
3389                if task_cx.checkpoint().is_err() {
3390                    "grandchild_cancelled"
3391                } else {
3392                    "grandchild_completed"
3393                }
3394            })
3395            .expect("spawn grandchild task");
3396        let grandchild_task_id = grandchild_handle.task_id();
3397
3398        let cancel_reason = CancelReason::shutdown().with_region(root);
3399        let cancelled = state.cancel_request(root, &cancel_reason, None);
3400        assert!(
3401            cancelled
3402                .iter()
3403                .any(|(task_id, _)| *task_id == child_task_id),
3404            "parent cancellation must reach child task"
3405        );
3406        assert!(
3407            cancelled
3408                .iter()
3409                .any(|(task_id, _)| *task_id == grandchild_task_id),
3410            "parent cancellation must reach grandchild task"
3411        );
3412
3413        let grandchild_tasks_before_failed_spawn = state
3414            .region(grandchild)
3415            .expect("grandchild region missing")
3416            .task_count();
3417        let live_tasks_before_failed_spawn = state.live_task_count();
3418        let failed_spawn = grandchild_scope.spawn(&mut state, &cx, |_| async { 99_u8 });
3419        let grandchild_tasks_after_failed_spawn = state
3420            .region(grandchild)
3421            .expect("grandchild region missing after failed spawn")
3422            .task_count();
3423        let live_tasks_after_failed_spawn = state.live_task_count();
3424
3425        let waker = std::task::Waker::noop().clone();
3426        let mut poll_cx = Context::from_waker(&waker);
3427        {
3428            let stored = state
3429                .get_stored_future(grandchild_task_id)
3430                .expect("grandchild stored task");
3431            let poll_result = stored.poll(&mut poll_cx);
3432            assert!(
3433                poll_result.is_pending(),
3434                "grandchild task should yield once before observing cancellation"
3435            );
3436        }
3437        {
3438            let stored = state
3439                .get_stored_future(grandchild_task_id)
3440                .expect("grandchild stored task");
3441            let poll_result = stored.poll(&mut poll_cx);
3442            let task_outcome = match poll_result {
3443                Poll::Ready(Outcome::Ok(())) => Outcome::Ok(()),
3444                Poll::Ready(Outcome::Panicked(payload)) => Outcome::Panicked(payload),
3445                other => panic!(
3446                    "grandchild task should complete once cancellation is observed: {other:?}"
3447                ),
3448            };
3449            if let Some(task_record) = state.task_mut(grandchild_task_id) {
3450                task_record.complete(task_outcome);
3451            }
3452        }
3453        let _ = state.task_completed(grandchild_task_id);
3454        state.advance_region_state(grandchild);
3455
3456        let mut grandchild_join_fut = std::pin::pin!(grandchild_handle.join(&cx));
3457        let grandchild_result = match grandchild_join_fut.as_mut().poll(&mut poll_cx) {
3458            Poll::Ready(Ok(result)) => result,
3459            other => panic!("grandchild cancellation join should succeed: {other:?}"),
3460        };
3461
3462        {
3463            let stored = state
3464                .get_stored_future(child_task_id)
3465                .expect("child stored task");
3466            let poll_result = stored.poll(&mut poll_cx);
3467            assert!(
3468                poll_result.is_pending(),
3469                "child task should yield once before observing cancellation"
3470            );
3471        }
3472        {
3473            let stored = state
3474                .get_stored_future(child_task_id)
3475                .expect("child stored task");
3476            let poll_result = stored.poll(&mut poll_cx);
3477            let task_outcome = match poll_result {
3478                Poll::Ready(Outcome::Ok(())) => Outcome::Ok(()),
3479                Poll::Ready(Outcome::Panicked(payload)) => Outcome::Panicked(payload),
3480                other => {
3481                    panic!("child task should complete once cancellation is observed: {other:?}")
3482                }
3483            };
3484            if let Some(task_record) = state.task_mut(child_task_id) {
3485                task_record.complete(task_outcome);
3486            }
3487        }
3488        let _ = state.task_completed(child_task_id);
3489        state.advance_region_state(child);
3490
3491        let mut child_join_fut = std::pin::pin!(child_handle.join(&cx));
3492        let child_result = match child_join_fut.as_mut().poll(&mut poll_cx) {
3493            Poll::Ready(Ok(result)) => result,
3494            other => panic!("child cancellation join should succeed: {other:?}"),
3495        };
3496
3497        assert_eq!(child_result, "child_cancelled");
3498        assert_eq!(grandchild_result, "grandchild_cancelled");
3499        assert!(
3500            matches!(failed_spawn, Err(SpawnError::RegionClosed(id)) if id == grandchild),
3501            "nested spawn after parent cancellation must fail against the closing grandchild region"
3502        );
3503        assert_eq!(
3504            grandchild_tasks_before_failed_spawn, grandchild_tasks_after_failed_spawn,
3505            "failed spawn after cancellation must not leak task membership into the grandchild region"
3506        );
3507        assert_eq!(
3508            live_tasks_before_failed_spawn, live_tasks_after_failed_spawn,
3509            "failed spawn after cancellation must not inflate runtime task count"
3510        );
3511        assert_eq!(
3512            *finalizer_log.lock().expect("finalizer log poisoned"), // ubs:ignore - test helper
3513            vec!["grandchild", "child"],
3514            "nested scope finalizers must run in reverse scope creation order"
3515        );
3516        assert!(
3517            state.region(grandchild).is_none(),
3518            "grandchild region should be reclaimed after close"
3519        );
3520        assert!(
3521            state.region(child).is_none(),
3522            "child region should be reclaimed after close"
3523        );
3524    }
3525
3526    #[test]
3527    fn conformance_race_loser_drain_invariant() {
3528        // INVARIANT: In race operations, losers are cancelled and fully drained
3529
3530        use std::task::{Context, Poll};
3531
3532        let mut state = RuntimeState::new();
3533        let cx = test_cx();
3534        let region = state.create_root_region(Budget::INFINITE);
3535        let scope = test_scope(region, Budget::INFINITE);
3536
3537        // Winner: completes immediately
3538        let (winner_handle, mut winner_stored) = scope
3539            .spawn(&mut state, &cx, |_| async { "winner" })
3540            .unwrap();
3541
3542        // Loser: would run longer, must be cancelled and drained
3543        let (loser_handle, mut loser_stored) = scope
3544            .spawn(&mut state, &cx, |cx| async move {
3545                // Simulate work that can be cancelled
3546                struct YieldOnce(bool);
3547                impl std::future::Future for YieldOnce {
3548                    type Output = ();
3549                    fn poll(
3550                        mut self: std::pin::Pin<&mut Self>,
3551                        cx: &mut std::task::Context<'_>,
3552                    ) -> std::task::Poll<()> {
3553                        if self.0 {
3554                            std::task::Poll::Ready(())
3555                        } else {
3556                            self.0 = true;
3557                            cx.waker().wake_by_ref();
3558                            std::task::Poll::Pending
3559                        }
3560                    }
3561                }
3562                YieldOnce(false).await;
3563
3564                // Check if we were cancelled
3565                if cx.checkpoint().is_err() {
3566                    "loser_cancelled"
3567                } else {
3568                    "loser_completed"
3569                }
3570            })
3571            .unwrap();
3572
3573        let waker = std::task::Waker::noop().clone();
3574        let mut poll_cx = Context::from_waker(&waker);
3575
3576        // Complete winner immediately
3577        assert!(winner_stored.poll(&mut poll_cx).is_ready());
3578
3579        // Start the race
3580        let handles = vec![winner_handle, loser_handle];
3581        let mut race_fut = std::pin::pin!(scope.race_all(&cx, handles));
3582
3583        // Race should be pending initially (waiting for loser drain)
3584        assert!(
3585            race_fut.as_mut().poll(&mut poll_cx).is_pending(),
3586            "race must wait for loser to be drained"
3587        );
3588
3589        // Drive loser to first yield (it's now pending, but abort signal propagates)
3590        assert!(loser_stored.poll(&mut poll_cx).is_pending());
3591
3592        // Still pending on race (loser not fully drained)
3593        assert!(race_fut.as_mut().poll(&mut poll_cx).is_pending());
3594
3595        // Drive loser to completion (should see cancellation)
3596        assert!(loser_stored.poll(&mut poll_cx).is_ready());
3597
3598        // Now race should complete with winner result
3599        match race_fut.as_mut().poll(&mut poll_cx) {
3600            Poll::Ready(Ok((result, winner_index))) => {
3601                assert_eq!(result, "winner", "race must return winner result");
3602                assert_eq!(winner_index, 0, "winner index must be correct");
3603            }
3604            other => panic!("race must complete after loser drain: {other:?}"),
3605        }
3606    }
3607
3608    #[test]
3609    fn conformance_region_quiescence_on_empty() {
3610        // INVARIANT: Empty regions reach quiescence and can be closed immediately
3611        let mut state = RuntimeState::new();
3612        let cx = test_cx();
3613        let parent_region = state.create_root_region(Budget::INFINITE);
3614        let scope = test_scope(parent_region, Budget::INFINITE);
3615
3616        // Create and immediately close a child region with no spawned tasks
3617        let outcome = block_on(scope.region(
3618            &mut state,
3619            &cx,
3620            crate::types::policy::FailFast,
3621            |_child_scope, _state| {
3622                // Don't spawn any tasks - region should be empty
3623                std::future::ready(Outcome::Ok("empty_region_completed"))
3624            },
3625        ))
3626        .expect("empty child region must complete");
3627
3628        match outcome {
3629            Outcome::Ok(result) => {
3630                assert_eq!(
3631                    result, "empty_region_completed",
3632                    "empty region must reach quiescence immediately"
3633                );
3634            }
3635            other => panic!("empty region must complete successfully: {other:?}"),
3636        }
3637
3638        // Child region should be cleaned up (no longer in parent's child list)
3639        let parent_record = state
3640            .region(parent_region)
3641            .expect("parent region must exist");
3642        assert!(
3643            parent_record.child_ids().is_empty(),
3644            "completed child region must be removed from parent"
3645        );
3646    }
3647
3648    #[test]
3649    fn region_close_future_wakes_all_registered_waiters() {
3650        use std::sync::atomic::{AtomicUsize, Ordering};
3651        use std::task::{Context, Waker};
3652
3653        struct CountWaker(Arc<AtomicUsize>);
3654
3655        impl std::task::Wake for CountWaker {
3656            fn wake(self: Arc<Self>) {
3657                self.0.fetch_add(1, Ordering::SeqCst);
3658            }
3659
3660            fn wake_by_ref(self: &Arc<Self>) {
3661                self.0.fetch_add(1, Ordering::SeqCst);
3662            }
3663        }
3664
3665        let notify = Arc::new(parking_lot::Mutex::new(
3666            crate::record::region::RegionCloseState {
3667                closed: false,
3668                waiters: Default::default(),
3669            },
3670        ));
3671        let wake_count_a = Arc::new(AtomicUsize::new(0));
3672        let wake_count_b = Arc::new(AtomicUsize::new(0));
3673        let waker_a = Waker::from(Arc::new(CountWaker(Arc::clone(&wake_count_a))));
3674        let waker_b = Waker::from(Arc::new(CountWaker(Arc::clone(&wake_count_b))));
3675        let mut cx_a = Context::from_waker(&waker_a);
3676        let mut cx_b = Context::from_waker(&waker_b);
3677        let mut future_a = RegionCloseFuture {
3678            state: Arc::clone(&notify),
3679        };
3680        let mut future_b = RegionCloseFuture { state: notify };
3681
3682        assert!(Pin::new(&mut future_a).poll(&mut cx_a).is_pending());
3683        assert!(Pin::new(&mut future_b).poll(&mut cx_b).is_pending());
3684
3685        let waiters = {
3686            let mut state = future_a.state.lock();
3687            state.closed = true;
3688            std::mem::take(&mut state.waiters)
3689        };
3690        for waker in waiters {
3691            waker.wake();
3692        }
3693
3694        assert_eq!(wake_count_a.load(Ordering::SeqCst), 1);
3695        assert_eq!(wake_count_b.load(Ordering::SeqCst), 1);
3696        assert!(Pin::new(&mut future_a).poll(&mut cx_a).is_ready());
3697        assert!(Pin::new(&mut future_b).poll(&mut cx_b).is_ready());
3698    }
3699
3700    #[test]
3701    fn conformance_spawn_into_closed_region_fails() {
3702        // INVARIANT: Cannot spawn tasks into regions that have begun closing
3703        let mut state = RuntimeState::new();
3704        let cx = test_cx();
3705        let region = state.create_root_region(Budget::INFINITE);
3706        let scope = test_scope(region, Budget::INFINITE);
3707
3708        // Close the region
3709        let region_record = state.region_mut(region).expect("region must exist");
3710        region_record.begin_close(None);
3711
3712        // Attempt to spawn should fail
3713        let spawn_result = scope.spawn(&mut state, &cx, |_| async { 42 });
3714
3715        assert!(
3716            matches!(spawn_result, Err(SpawnError::RegionClosed(_))),
3717            "spawning into closed region must fail with RegionClosed error"
3718        );
3719
3720        // State should remain consistent (no orphaned tasks)
3721        assert!(
3722            state.tasks_is_empty() || state.region(region).is_none_or(|r| r.task_ids().is_empty()),
3723            "failed spawn must not create orphaned tasks"
3724        );
3725    }
3726
3727    #[test]
3728    fn conformance_join_multiple_tasks_preserves_results() {
3729        // INVARIANT: join_all preserves all task results in order
3730
3731        use std::task::{Context, Poll};
3732
3733        let mut state = RuntimeState::new();
3734        let cx = test_cx();
3735        let region = state.create_root_region(Budget::INFINITE);
3736        let scope = test_scope(region, Budget::INFINITE);
3737
3738        // Spawn multiple tasks with different results
3739        let (h1, mut t1) = scope.spawn(&mut state, &cx, |_| async { 100_i32 }).unwrap();
3740        let (h2, mut t2) = scope.spawn(&mut state, &cx, |_| async { 200_i32 }).unwrap();
3741        let (h3, mut t3) = scope.spawn(&mut state, &cx, |_| async { 300_i32 }).unwrap();
3742
3743        let waker = std::task::Waker::noop().clone();
3744        let mut poll_cx = Context::from_waker(&waker);
3745
3746        // Complete all tasks
3747        assert!(t1.poll(&mut poll_cx).is_ready());
3748        assert!(t2.poll(&mut poll_cx).is_ready());
3749        assert!(t3.poll(&mut poll_cx).is_ready());
3750
3751        // Join all tasks
3752        let handles = vec![h1, h2, h3];
3753        let mut join_all_fut = std::pin::pin!(scope.join_all(&cx, handles));
3754
3755        match join_all_fut.as_mut().poll(&mut poll_cx) {
3756            Poll::Ready(results) => {
3757                assert_eq!(results.len(), 3, "join_all must return all results");
3758
3759                // Results must be in the same order as handles
3760                assert_eq!(
3761                    results[0].as_ref().unwrap(),
3762                    &100,
3763                    "first task result must be preserved"
3764                );
3765                assert_eq!(
3766                    results[1].as_ref().unwrap(),
3767                    &200,
3768                    "second task result must be preserved"
3769                );
3770                assert_eq!(
3771                    results[2].as_ref().unwrap(),
3772                    &300,
3773                    "third task result must be preserved"
3774                );
3775            }
3776            other @ Poll::Pending => panic!("join_all must complete with all results: {other:?}"),
3777        }
3778    }
3779
3780    #[test]
3781    fn conformance_panic_propagation_through_join() {
3782        // INVARIANT: Task panics are captured and propagated through join as JoinError::Panicked
3783
3784        use std::task::{Context, Poll};
3785
3786        let mut state = RuntimeState::new();
3787        let cx = test_cx();
3788        let region = state.create_root_region(Budget::INFINITE);
3789        let scope = test_scope(region, Budget::INFINITE);
3790
3791        // Spawn a task that panics with a specific message
3792        let (mut handle, mut stored) = scope
3793            .spawn(&mut state, &cx, |_| async {
3794                std::panic::panic_any("test_panic_message");
3795            })
3796            .unwrap();
3797
3798        let waker = std::task::Waker::noop().clone();
3799        let mut poll_cx = Context::from_waker(&waker);
3800
3801        // Task execution should complete with Panicked outcome
3802        match stored.poll(&mut poll_cx) {
3803            Poll::Ready(crate::types::Outcome::Panicked(_)) => {
3804                // Expected: panic was caught and wrapped as Outcome::Panicked
3805            }
3806            other => panic!("panicking task must complete with Panicked outcome: {other:?}"),
3807        }
3808
3809        // Join should propagate the panic as JoinError::Panicked
3810        let mut join_fut = std::pin::pin!(handle.join(&cx));
3811        match join_fut.as_mut().poll(&mut poll_cx) {
3812            Poll::Ready(Err(JoinError::Panicked(payload))) => {
3813                assert_eq!(
3814                    payload.message(),
3815                    "test_panic_message",
3816                    "join must preserve panic payload message"
3817                );
3818            }
3819            other => panic!("join of panicked task must return JoinError::Panicked: {other:?}"),
3820        }
3821    }
3822
3823    /// Regression: region_with_budget factory panicking AFTER scope.spawn(...)
3824    /// returned — but BEFORE the StoredTask was scheduled — must not leave an
3825    /// orphan TaskRecord bound to the child region. Without the panic-catch
3826    /// rollback in `region_with_budget`, `region.task_count()` stays non-zero
3827    /// and `advance_region_state` can never transition Closing → Finalizing.
3828    ///
3829    /// Bead asupersync-c7s5de.
3830    #[test]
3831    fn region_factory_panic_after_spawn_cleans_up_orphan_task_records() {
3832        let mut state = RuntimeState::new();
3833        let cx = test_cx();
3834        let parent = state.create_root_region(Budget::INFINITE);
3835        let scope = test_scope(parent, Budget::INFINITE);
3836
3837        let tasks_before = state.tasks_iter().count();
3838
3839        // Wrap the call in catch_unwind because the factory intentionally
3840        // panics after spawn — resume_unwind re-raises in region_with_budget.
3841        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3842            block_on(scope.region_with_budget(
3843                &mut state,
3844                &cx,
3845                Budget::INFINITE,
3846                crate::types::policy::FailFast,
3847                |child, state| {
3848                    // Spawn a task so a TaskRecord exists in Created state,
3849                    // bound to the child region.
3850                    let (_handle, _stored) = child
3851                        .spawn(state, &cx, |_| async { 42_i32 })
3852                        .expect("spawn must succeed before the panic");
3853                    // Immediately panic — StoredTask is dropped unscheduled.
3854                    std::panic::panic_any("factory panic after spawn");
3855                    #[allow(unreachable_code)]
3856                    std::future::ready(Outcome::Ok(0_i32))
3857                },
3858            ))
3859        }));
3860
3861        assert!(
3862            result.is_err(),
3863            "region_with_budget must re-raise the factory panic",
3864        );
3865
3866        // Orphan TaskRecord(s) created inside the factory must have been
3867        // removed by the panic-catch rollback. The task count should be
3868        // back to its pre-call value.
3869        let tasks_after = state.tasks_iter().count();
3870        assert_eq!(
3871            tasks_after, tasks_before,
3872            "post-panic tasks_iter count must match pre-call count \
3873             (orphan TaskRecord not cleaned up: before={tasks_before} after={tasks_after})",
3874        );
3875    }
3876
3877    #[test]
3878    fn spawn_local_wake_state_single_notify() {
3879        // Regression test for asupersync-3kwekm: ensure spawn_local only calls
3880        // notify() once and sets the task wake state correctly.
3881        use std::sync::Arc;
3882
3883        let mut state = RuntimeState::new();
3884        let cx = test_cx();
3885        let region = state.create_root_region(Budget::INFINITE);
3886        let scope = test_scope(region, Budget::INFINITE);
3887
3888        let local_ready = Arc::new(parking_lot::Mutex::new(std::collections::VecDeque::new()));
3889        let _local_ready_guard =
3890            crate::runtime::scheduler::three_lane::ScopedLocalReady::new(Arc::clone(&local_ready));
3891        let _worker_guard = crate::runtime::scheduler::three_lane::ScopedWorkerId::new(1);
3892
3893        let handle = scope
3894            .spawn_local(&mut state, &cx, |_| async move { "test_result" })
3895            .unwrap();
3896
3897        // Verify the task was created and is in the expected state
3898        let task_record = state.task(handle.task_id()).unwrap();
3899
3900        // The task should have been notified exactly once during spawn_local.
3901        // Subsequent notify() calls should return false (already notified).
3902        let second_notify_result = task_record.wake_state.notify();
3903        assert!(
3904            !second_notify_result,
3905            "Second notify() should return false - task already notified during spawn_local"
3906        );
3907
3908        // Verify task was properly scheduled
3909        let queued = {
3910            let queue = local_ready.lock();
3911            queue.contains(&handle.task_id())
3912        };
3913        assert!(queued, "spawn_local should enqueue task into local_ready");
3914    }
3915}