use super::*;
use futures::{
    FutureExt,
    channel::{mpsc, oneshot},
    executor::block_on,
    future,
    sink::SinkExt,
    stream::{FuturesUnordered, StreamExt},
};
use std::{
    cell::RefCell,
    collections::{BTreeSet, HashSet},
    pin::Pin,
    rc::Rc,
    sync::Arc,
    task::{Context, Poll, Waker},
};

#[test]
fn test_foreground_executor_spawn() {
    let result = TestScheduler::once(async |scheduler| {
        let task = scheduler.foreground().spawn(async move { 42 });
        task.await
    });
    assert_eq!(result, 42);
}

#[test]
fn test_background_executor_spawn() {
    TestScheduler::once(async |scheduler| {
        let task = scheduler.background().spawn(async move { 42 });
        let result = task.await;
        assert_eq!(result, 42);
    });
}

#[test]
fn test_scheduler_drops_with_stalled_detached_foreground_task() {
    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
    let weak_scheduler = Arc::downgrade(&scheduler);
    let (sender, receiver) = oneshot::channel::<()>();

    scheduler
        .foreground()
        .spawn(async move {
            receiver.await.ok();
        })
        .detach();
    scheduler.run();

    drop(scheduler);
    assert!(weak_scheduler.upgrade().is_none());
    drop(sender);
}

#[test]
fn test_scheduler_drops_with_stalled_detached_background_task() {
    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
    let weak_scheduler = Arc::downgrade(&scheduler);
    let (sender, receiver) = oneshot::channel::<()>();

    scheduler
        .background()
        .spawn(async move {
            receiver.await.ok();
        })
        .detach();
    scheduler.run();

    drop(scheduler);
    assert!(weak_scheduler.upgrade().is_none());
    drop(sender);
}

/// A dedicated task that is never polled must not keep the scheduler alive:
/// its runnable sits in the scheduler's own queue, so any strong scheduler
/// handle captured by the future would form a reference cycle and leak both.
#[test]
fn test_scheduler_drops_with_never_polled_dedicated_task() {
    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
    let weak_scheduler = Arc::downgrade(&scheduler);

    scheduler
        .background()
        .spawn_dedicated(|_executor| async move {})
        .detach();

    drop(scheduler);
    assert!(weak_scheduler.upgrade().is_none());
}

#[test]
fn test_foreground_ordering() {
    let mut traces = HashSet::new();

    TestScheduler::many(if cfg!(miri) { 5 } else { 100 }, async |scheduler| {
        #[derive(Hash, PartialEq, Eq)]
        struct TraceEntry {
            session: usize,
            task: usize,
        }

        let trace = Rc::new(RefCell::new(Vec::new()));

        let foreground_1 = scheduler.foreground();
        for task in 0..10 {
            foreground_1
                .spawn({
                    let trace = trace.clone();
                    async move {
                        trace.borrow_mut().push(TraceEntry { session: 0, task });
                    }
                })
                .detach();
        }

        let foreground_2 = scheduler.foreground();
        for task in 0..10 {
            foreground_2
                .spawn({
                    let trace = trace.clone();
                    async move {
                        trace.borrow_mut().push(TraceEntry { session: 1, task });
                    }
                })
                .detach();
        }

        scheduler.run();

        assert_eq!(
            trace
                .borrow()
                .iter()
                .filter(|entry| entry.session == 0)
                .map(|entry| entry.task)
                .collect::<Vec<_>>(),
            (0..10).collect::<Vec<_>>()
        );
        assert_eq!(
            trace
                .borrow()
                .iter()
                .filter(|entry| entry.session == 1)
                .map(|entry| entry.task)
                .collect::<Vec<_>>(),
            (0..10).collect::<Vec<_>>()
        );

        traces.insert(trace.take());
    });

    assert!(traces.len() > 1, "Expected at least two traces");
}

#[test]
fn test_timer_ordering() {
    TestScheduler::many(1, async |scheduler| {
        let background = scheduler.background();
        let futures = FuturesUnordered::new();
        futures.push(
            async {
                background.timer(Duration::from_millis(100)).await;
                2
            }
            .boxed(),
        );
        futures.push(
            async {
                background.timer(Duration::from_millis(50)).await;
                1
            }
            .boxed(),
        );
        futures.push(
            async {
                background.timer(Duration::from_millis(150)).await;
                3
            }
            .boxed(),
        );
        assert_eq!(futures.collect::<Vec<_>>().await, vec![1, 2, 3]);
    });
}

#[test]
fn test_foreground_task_can_hold_mut_borrow_across_await() {
    TestScheduler::once(async |scheduler| {
        let foreground = scheduler.foreground();
        let (sender, mut receiver) = mpsc::unbounded::<()>();

        foreground
            .spawn(async move {
                receiver.next().await;
            })
            .detach();

        scheduler.run();
        sender.unbounded_send(()).unwrap();
        scheduler.run();
    });
}

#[test]
fn test_send_from_bg_to_fg() {
    TestScheduler::once(async |scheduler| {
        let foreground = scheduler.foreground();
        let background = scheduler.background();

        let (sender, receiver) = oneshot::channel::<i32>();

        background
            .spawn(async move {
                sender.send(42).unwrap();
            })
            .detach();

        let task = foreground.spawn(async move { receiver.await.unwrap() });
        let result = task.await;
        assert_eq!(result, 42);
    });
}

#[test]
fn test_randomize_order() {
    // Test deterministic mode: different seeds should produce same execution order
    let mut deterministic_results = HashSet::new();
    for seed in 0..10 {
        let config = TestSchedulerConfig {
            seed,
            randomize_order: false,
            ..Default::default()
        };
        let order = block_on(capture_execution_order(config));
        assert_eq!(order.len(), 6);
        deterministic_results.insert(order);
    }

    // All deterministic runs should produce the same result
    assert_eq!(
        deterministic_results.len(),
        1,
        "Deterministic mode should always produce same execution order"
    );

    // Test randomized mode: different seeds can produce different execution orders
    let mut randomized_results = HashSet::new();
    for seed in 0..20 {
        let config = TestSchedulerConfig::with_seed(seed);
        let order = block_on(capture_execution_order(config));
        assert_eq!(order.len(), 6);
        randomized_results.insert(order);
    }

    // Randomized mode should produce multiple different execution orders
    assert!(
        randomized_results.len() > 1,
        "Randomized mode should produce multiple different orders"
    );
}

async fn capture_execution_order(config: TestSchedulerConfig) -> Vec<String> {
    let scheduler = Arc::new(TestScheduler::new(config));
    let foreground = scheduler.foreground();
    let background = scheduler.background();

    let (sender, receiver) = mpsc::unbounded::<String>();

    // Spawn foreground tasks
    for i in 0..3 {
        let mut sender = sender.clone();
        foreground
            .spawn(async move {
                sender.send(format!("fg-{}", i)).await.ok();
            })
            .detach();
    }

    // Spawn background tasks
    for i in 0..3 {
        let mut sender = sender.clone();
        background
            .spawn(async move {
                sender.send(format!("bg-{}", i)).await.ok();
            })
            .detach();
    }

    drop(sender); // Close sender to signal no more messages
    scheduler.run();

    receiver.collect().await
}

#[test]
fn test_block() {
    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
    let (tx, rx) = oneshot::channel();

    // Spawn background task to send value
    let _ = scheduler
        .background()
        .spawn(async move {
            tx.send(42).unwrap();
        })
        .detach();

    // Block on receiving the value
    let result = scheduler.foreground().block_on(async { rx.await.unwrap() });
    assert_eq!(result, 42);
}

#[test]
#[should_panic(expected = "Parking forbidden.")]
fn test_parking_panics() {
    let config = TestSchedulerConfig {
        capture_pending_traces: true,
        ..Default::default()
    };
    let scheduler = Arc::new(TestScheduler::new(config));
    scheduler.foreground().block_on(async {
        let (_tx, rx) = oneshot::channel::<()>();
        rx.await.unwrap(); // This will never complete
    });
}

#[test]
fn test_block_with_parking() {
    let config = TestSchedulerConfig {
        allow_parking: true,
        ..Default::default()
    };
    let scheduler = Arc::new(TestScheduler::new(config));
    let (tx, rx) = oneshot::channel();

    // Spawn background task to send value
    let _ = scheduler
        .background()
        .spawn(async move {
            tx.send(42).unwrap();
        })
        .detach();

    // Block on receiving the value (will park if needed)
    let result = scheduler.foreground().block_on(async { rx.await.unwrap() });
    assert_eq!(result, 42);
}

#[test]
fn test_helper_methods() {
    // Test the once method
    let result = TestScheduler::once(async |scheduler: Arc<TestScheduler>| {
        let background = scheduler.background();
        background.spawn(async { 42 }).await
    });
    assert_eq!(result, 42);

    // Test the many method
    let results = TestScheduler::many(3, async |scheduler: Arc<TestScheduler>| {
        let background = scheduler.background();
        background.spawn(async { 10 }).await
    });
    assert_eq!(results, vec![10, 10, 10]);
}

#[test]
fn test_many_with_arbitrary_seed() {
    for seed in [0u64, 1, 5, 42] {
        let mut seeds_seen = Vec::new();
        let iterations = 3usize;

        for current_seed in seed..seed + iterations as u64 {
            let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::with_seed(
                current_seed,
            )));
            let captured_seed = current_seed;
            scheduler
                .foreground()
                .block_on(async { seeds_seen.push(captured_seed) });
            scheduler.run();
        }

        assert_eq!(
            seeds_seen,
            (seed..seed + iterations as u64).collect::<Vec<_>>(),
            "Expected {iterations} iterations starting at seed {seed}"
        );
    }
}

#[test]
fn test_block_with_timeout() {
    // Test case: future completes within timeout
    TestScheduler::once(async |scheduler| {
        let foreground = scheduler.foreground();
        let future = future::ready(42);
        let output = foreground.block_with_timeout(Duration::from_millis(100), future);
        assert_eq!(output.ok(), Some(42));
    });

    // Test case: future times out
    TestScheduler::once(async |scheduler| {
        // Make timeout behavior deterministic by forcing the timeout tick budget to be exactly 0.
        // This prevents `block_with_timeout` from making progress via extra scheduler stepping and
        // accidentally completing work that we expect to time out.
        scheduler.set_timeout_ticks(0..=0);

        let foreground = scheduler.foreground();
        let future = future::pending::<()>();
        let output = foreground.block_with_timeout(Duration::from_millis(50), future);
        assert!(output.is_err(), "future should not have finished");
    });

    // Test case: future makes progress via timer but still times out
    let mut results = BTreeSet::new();
    TestScheduler::many(if cfg!(miri) { 5 } else { 100 }, async |scheduler| {
        // Keep the existing probabilistic behavior here (do not force 0 ticks), since this subtest
        // is explicitly checking that some seeds/timeouts can complete while others can time out.
        let task = scheduler.background().spawn(async move {
            Yield { polls: 10 }.await;
            42
        });
        let output = scheduler
            .foreground()
            .block_with_timeout(Duration::from_millis(50), task);
        results.insert(output.ok());
    });
    assert_eq!(
        results.into_iter().collect::<Vec<_>>(),
        if cfg!(miri) {
            vec![Some(42)]
        } else {
            vec![None, Some(42)]
        }
    );

    // Regression test:
    // A timed-out future must not be cancelled. The returned future should still be
    // pollable to completion later. We also want to ensure time only advances when we
    // explicitly advance it (not by yielding).
    TestScheduler::once(async |scheduler| {
        // Force immediate timeout: the timeout tick budget is 0 so we will not step or
        // advance timers inside `block_with_timeout`.
        scheduler.set_timeout_ticks(0..=0);

        let background = scheduler.background();

        // This task should only complete once time is explicitly advanced.
        let task = background.spawn({
            let scheduler = scheduler.clone();
            async move {
                scheduler.timer(Duration::from_millis(100)).await;
                123
            }
        });

        // This should time out before we advance time enough for the timer to fire.
        let timed_out = scheduler
            .foreground()
            .block_with_timeout(Duration::from_millis(50), task);
        assert!(
            timed_out.is_err(),
            "expected timeout before advancing the clock enough for the timer"
        );

        // Now explicitly advance time and ensure the returned future can complete.
        let mut task = timed_out.err().unwrap();
        scheduler.advance_clock(Duration::from_millis(100));
        scheduler.run();

        let output = scheduler.foreground().block_on(&mut task);
        assert_eq!(output, 123);
    });
}

// When calling block, we shouldn't make progress on foreground-spawned futures with the same session id.
#[test]
fn test_block_does_not_progress_same_session_foreground() {
    let mut task2_made_progress_once = false;
    TestScheduler::many(if cfg!(miri) { 5 } else { 1000 }, async |scheduler| {
        let foreground1 = scheduler.foreground();
        let foreground2 = scheduler.foreground();

        let task1 = foreground1.spawn(async move {});
        let task2 = foreground2.spawn(async move {});

        foreground1.block_on(async {
            scheduler.yield_random().await;
            assert!(!task1.is_ready());
            task2_made_progress_once |= task2.is_ready();
        });

        task1.await;
        task2.await;
    });

    assert!(
        task2_made_progress_once,
        "Expected task from different foreground executor to make progress (at least once)"
    );
}

struct Yield {
    polls: usize,
}

impl Future for Yield {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.polls -= 1;
        if self.polls == 0 {
            Poll::Ready(())
        } else {
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

#[test]
fn test_nondeterministic_wake_detection() {
    let config = TestSchedulerConfig {
        allow_parking: false,
        ..Default::default()
    };
    let scheduler = Arc::new(TestScheduler::new(config));

    // A future that captures its waker and sends it to an external thread
    struct SendWakerToThread {
        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
    }

    impl Future for SendWakerToThread {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if let Some(tx) = self.waker_tx.take() {
                tx.send(cx.waker().clone()).ok();
            }
            Poll::Ready(())
        }
    }

    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();

    // Get a waker by running a future that sends it
    scheduler.foreground().block_on(SendWakerToThread {
        waker_tx: Some(waker_tx),
    });

    // Spawn a real OS thread that will call wake() on the waker
    let handle = std::thread::spawn(move || {
        if let Ok(waker) = waker_rx.recv() {
            // This should trigger the non-determinism detection
            waker.wake();
        }
    });

    // Wait for the spawned thread to complete
    handle.join().ok();

    // The non-determinism error should be detected when end_test is called
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        scheduler.end_test();
    }));
    assert!(result.is_err(), "Expected end_test to panic");
    let panic_payload = result.unwrap_err();
    let panic_message = panic_payload
        .downcast_ref::<String>()
        .map(|s| s.as_str())
        .or_else(|| panic_payload.downcast_ref::<&str>().copied())
        .unwrap_or("<unknown panic>");
    assert!(
        panic_message.contains("Your test is not deterministic"),
        "Expected panic message to contain non-determinism error, got: {}",
        panic_message
    );
}

#[test]
fn test_nondeterministic_wake_allowed_with_parking() {
    let config = TestSchedulerConfig {
        allow_parking: true,
        ..Default::default()
    };
    let scheduler = Arc::new(TestScheduler::new(config));

    // A future that captures its waker and sends it to an external thread
    struct WakeFromExternalThread {
        waker_sent: bool,
        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
    }

    impl Future for WakeFromExternalThread {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if !self.waker_sent {
                self.waker_sent = true;
                if let Some(tx) = self.waker_tx.take() {
                    tx.send(cx.waker().clone()).ok();
                }
                Poll::Pending
            } else {
                Poll::Ready(())
            }
        }
    }

    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();

    // Spawn a real OS thread that will call wake() on the waker
    std::thread::spawn(move || {
        if let Ok(waker) = waker_rx.recv() {
            // With allow_parking, this should NOT panic
            waker.wake();
        }
    });

    // This should complete without panicking
    scheduler.foreground().block_on(WakeFromExternalThread {
        waker_sent: false,
        waker_tx: Some(waker_tx),
    });
}

#[test]
fn test_nondeterministic_waker_drop_detection() {
    let config = TestSchedulerConfig {
        allow_parking: false,
        ..Default::default()
    };
    let scheduler = Arc::new(TestScheduler::new(config));

    // A future that captures its waker and sends it to an external thread
    struct SendWakerToThread {
        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
    }

    impl Future for SendWakerToThread {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if let Some(tx) = self.waker_tx.take() {
                tx.send(cx.waker().clone()).ok();
            }
            Poll::Ready(())
        }
    }

    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();

    // Get a waker by running a future that sends it
    scheduler.foreground().block_on(SendWakerToThread {
        waker_tx: Some(waker_tx),
    });

    // Spawn a real OS thread that will drop the waker without calling wake
    let handle = std::thread::spawn(move || {
        if let Ok(waker) = waker_rx.recv() {
            // This should trigger the non-determinism detection on drop
            drop(waker);
        }
    });

    // Wait for the spawned thread to complete
    handle.join().ok();

    // The non-determinism error should be detected when end_test is called
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        scheduler.end_test();
    }));
    assert!(result.is_err(), "Expected end_test to panic");
    let panic_payload = result.unwrap_err();
    let panic_message = panic_payload
        .downcast_ref::<String>()
        .map(|s| s.as_str())
        .or_else(|| panic_payload.downcast_ref::<&str>().copied())
        .unwrap_or("<unknown panic>");
    assert!(
        panic_message.contains("Your test is not deterministic"),
        "Expected panic message to contain non-determinism error, got: {}",
        panic_message
    );
}

#[test]
fn test_background_priority_scheduling() {
    use parking_lot::Mutex;

    // Run many iterations to get statistical significance
    let mut high_before_low_count = 0;
    let iterations = if cfg!(miri) { 5 } else { 100 };

    for seed in 0..iterations {
        let config = TestSchedulerConfig::with_seed(seed);
        let scheduler = Arc::new(TestScheduler::new(config));
        let background = scheduler.background();

        let execution_order = Arc::new(Mutex::new(Vec::new()));

        // Spawn low priority tasks first
        for i in 0..3 {
            let order = execution_order.clone();
            background
                .spawn_with_priority(Priority::Low, async move {
                    order.lock().push(format!("low-{}", i));
                })
                .detach();
        }

        // Spawn high priority tasks second
        for i in 0..3 {
            let order = execution_order.clone();
            background
                .spawn_with_priority(Priority::High, async move {
                    order.lock().push(format!("high-{}", i));
                })
                .detach();
        }

        scheduler.run();

        // Count how many high priority tasks ran in the first half
        let order = execution_order.lock();
        let high_in_first_half = order
            .iter()
            .take(3)
            .filter(|s| s.starts_with("high"))
            .count();

        if high_in_first_half >= 2 {
            high_before_low_count += 1;
        }
    }

    // High priority tasks should tend to run before low priority tasks
    // With weights of 60 vs 10, high priority should dominate early execution
    assert!(
        high_before_low_count > iterations / 2,
        "Expected high priority tasks to run before low priority tasks more often. \
         Got {} out of {} iterations",
        high_before_low_count,
        iterations
    );
}

#[test]
fn test_spawn_dedicated_basic_round_trip() {
    let result = TestScheduler::once(async |scheduler| {
        scheduler
            .background()
            .spawn_dedicated(|_executor| async { 42 })
            .await
    });
    assert_eq!(result, 42);
}

#[test]
fn test_spawn_dedicated_not_send_future() {
    let result = TestScheduler::once(async |scheduler| {
        scheduler
            .background()
            .spawn_dedicated(|_executor| async move {
                // `Rc<RefCell<_>>` is `!Send`. If `spawn_dedicated` required
                // the returned future to be `Send`, this wouldn't compile.
                let state = Rc::new(RefCell::new(0_i32));
                for _ in 0..5 {
                    *state.borrow_mut() += 1;
                }
                *state.borrow()
            })
            .await
    });
    assert_eq!(result, 5);
}

#[test]
fn test_spawn_dedicated_send_closure_captures() {
    use parking_lot::Mutex;

    let observed = TestScheduler::once(async |scheduler| {
        let shared = Arc::new(Mutex::new(0_i32));
        let shared_for_closure = shared.clone();
        let returned = scheduler
            .background()
            .spawn_dedicated(move |_executor| {
                // `shared_for_closure` crossed the `Send` boundary of the
                // closure; we then mutate it from inside the !Send future.
                let local = shared_for_closure;
                async move {
                    *local.lock() = 7;
                }
            })
            .await;
        let _: () = returned;
        *shared.lock()
    });
    assert_eq!(observed, 7);
}

#[test]
fn test_spawn_dedicated_inner_spawn_local() {
    let result = TestScheduler::once(async |scheduler| {
        scheduler
            .background()
            .spawn_dedicated(|executor| async move {
                // The provided executor can spawn additional `!Send` work
                // onto the same dedicated session.
                let inner = Rc::new(RefCell::new(0_i32));
                let inner_for_child = inner.clone();
                let child = executor.spawn(async move {
                    *inner_for_child.borrow_mut() = 99;
                    *inner_for_child.borrow()
                });
                child.await
            })
            .await
    });
    assert_eq!(result, 99);
}

#[test]
fn test_spawn_dedicated_determinism_under_many() {
    use parking_lot::Mutex;

    let outcomes = TestScheduler::many(if cfg!(miri) { 4 } else { 20 }, async |scheduler| {
        let trace = Arc::new(Mutex::new(Vec::<u32>::new()));

        let background = scheduler.background();
        let mut tasks = Vec::new();
        for id in 0..4_u32 {
            let trace = trace.clone();
            let task = background.spawn_dedicated(move |executor| async move {
                for step in 0..3 {
                    trace.lock().push(id * 100 + step);
                    executor.spawn(async {}).await;
                }
                id
            });
            tasks.push(task);
        }

        let mut outputs = Vec::new();
        for task in tasks {
            outputs.push(task.await);
        }

        (trace.lock().clone(), outputs)
    });

    // Re-running with the same seed should produce the same trace. Run a
    // second pass with identical seeds and compare to the first.
    let outcomes_replay = TestScheduler::many(if cfg!(miri) { 4 } else { 20 }, async |scheduler| {
        let trace = Arc::new(Mutex::new(Vec::<u32>::new()));

        let background = scheduler.background();
        let mut tasks = Vec::new();
        for id in 0..4_u32 {
            let trace = trace.clone();
            let task = background.spawn_dedicated(move |executor| async move {
                for step in 0..3 {
                    trace.lock().push(id * 100 + step);
                    executor.spawn(async {}).await;
                }
                id
            });
            tasks.push(task);
        }

        let mut outputs = Vec::new();
        for task in tasks {
            outputs.push(task.await);
        }

        (trace.lock().clone(), outputs)
    });

    assert_eq!(
        outcomes, outcomes_replay,
        "per-seed outcomes should be reproducible"
    );

    // Sanity: at least one seed produced a non-monotonic trace,
    // demonstrating that dedicated tasks really do interleave under the
    // scheduler's randomization.
    let any_interleaved = outcomes.iter().any(|(trace, _)| {
        trace
            .windows(2)
            .any(|window| window[0] / 100 != window[1] / 100)
    });
    assert!(
        any_interleaved,
        "expected at least one seed to interleave dedicated tasks"
    );
}

#[test]
fn test_spawn_dedicated_dropping_task_cancels_future() {
    use parking_lot::Mutex;

    let counter_after = TestScheduler::once(async |scheduler| {
        let counter = Arc::new(Mutex::new(0_u32));
        let (resume_tx, resume_rx) = oneshot::channel::<()>();

        let task = {
            let counter = counter.clone();
            scheduler
                .background()
                .spawn_dedicated(move |_executor| async move {
                    *counter.lock() = 1;
                    // Park here until the test resumes us. If the task is
                    // dropped before this resolves, the second assignment
                    // below must never happen.
                    let _ = resume_rx.await;
                    *counter.lock() = 2;
                })
        };

        // Let the dedicated future make its first observable step.
        scheduler.run();
        assert_eq!(*counter.lock(), 1);

        // Cancel by dropping the root task, then unblock the parked oneshot.
        // The future must not advance past the await: counter stays at 1.
        drop(task);
        let _ = resume_tx.send(());
        scheduler.run();

        *counter.lock()
    });

    assert_eq!(
        counter_after, 1,
        "dropping the dedicated task must cancel the root future before its second write"
    );
}

#[test]
fn test_spawn_dedicated_detached_child_runs_after_root_completes() {
    use parking_lot::Mutex;

    let child_ran = TestScheduler::once(async |scheduler| {
        let child_ran = Arc::new(Mutex::new(false));

        let task = {
            let child_ran = child_ran.clone();
            scheduler
                .background()
                .spawn_dedicated(move |executor| async move {
                    executor
                        .spawn(async move {
                            *child_ran.lock() = true;
                        })
                        .detach();
                    // Root returns immediately, before the child has had a
                    // chance to run.
                })
        };

        task.await;

        // Drain the dedicated session. The detached child must run.
        scheduler.run();

        *child_ran.lock()
    });

    assert!(
        child_ran,
        "detached child must complete after the root, not be cancelled with it"
    );
}

// The production smoke test for `spawn_dedicated` lives in the `gpui` crate
// alongside `PlatformScheduler`, which is the real production implementation
// of the `Scheduler` trait. See `crates/gpui/src/platform_scheduler.rs`.