tenant.openagents/omega

1use super::*;
2use futures::{
3    FutureExt,
4    channel::{mpsc, oneshot},
5    executor::block_on,
6    future,
7    sink::SinkExt,
8    stream::{FuturesUnordered, StreamExt},
9};
10use std::{
11    cell::RefCell,
12    collections::{BTreeSet, HashSet},
13    pin::Pin,
14    rc::Rc,
15    sync::Arc,
16    task::{Context, Poll, Waker},
17};
18
19#[test]
20fn test_foreground_executor_spawn() {
21    let result = TestScheduler::once(async |scheduler| {
22        let task = scheduler.foreground().spawn(async move { 42 });
23        task.await
24    });
25    assert_eq!(result, 42);
26}
27
28#[test]
29fn test_background_executor_spawn() {
30    TestScheduler::once(async |scheduler| {
31        let task = scheduler.background().spawn(async move { 42 });
32        let result = task.await;
33        assert_eq!(result, 42);
34    });
35}
36
37#[test]
38fn test_scheduler_drops_with_stalled_detached_foreground_task() {
39    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
40    let weak_scheduler = Arc::downgrade(&scheduler);
41    let (sender, receiver) = oneshot::channel::<()>();
42
43    scheduler
44        .foreground()
45        .spawn(async move {
46            receiver.await.ok();
47        })
48        .detach();
49    scheduler.run();
50
51    drop(scheduler);
52    assert!(weak_scheduler.upgrade().is_none());
53    drop(sender);
54}
55
56#[test]
57fn test_scheduler_drops_with_stalled_detached_background_task() {
58    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
59    let weak_scheduler = Arc::downgrade(&scheduler);
60    let (sender, receiver) = oneshot::channel::<()>();
61
62    scheduler
63        .background()
64        .spawn(async move {
65            receiver.await.ok();
66        })
67        .detach();
68    scheduler.run();
69
70    drop(scheduler);
71    assert!(weak_scheduler.upgrade().is_none());
72    drop(sender);
73}
74
75/// A dedicated task that is never polled must not keep the scheduler alive:
76/// its runnable sits in the scheduler's own queue, so any strong scheduler
77/// handle captured by the future would form a reference cycle and leak both.
78#[test]
79fn test_scheduler_drops_with_never_polled_dedicated_task() {
80    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
81    let weak_scheduler = Arc::downgrade(&scheduler);
82
83    scheduler
84        .background()
85        .spawn_dedicated(|_executor| async move {})
86        .detach();
87
88    drop(scheduler);
89    assert!(weak_scheduler.upgrade().is_none());
90}
91
92#[test]
93fn test_foreground_ordering() {
94    let mut traces = HashSet::new();
95
96    TestScheduler::many(if cfg!(miri) { 5 } else { 100 }, async |scheduler| {
97        #[derive(Hash, PartialEq, Eq)]
98        struct TraceEntry {
99            session: usize,
100            task: usize,
101        }
102
103        let trace = Rc::new(RefCell::new(Vec::new()));
104
105        let foreground_1 = scheduler.foreground();
106        for task in 0..10 {
107            foreground_1
108                .spawn({
109                    let trace = trace.clone();
110                    async move {
111                        trace.borrow_mut().push(TraceEntry { session: 0, task });
112                    }
113                })
114                .detach();
115        }
116
117        let foreground_2 = scheduler.foreground();
118        for task in 0..10 {
119            foreground_2
120                .spawn({
121                    let trace = trace.clone();
122                    async move {
123                        trace.borrow_mut().push(TraceEntry { session: 1, task });
124                    }
125                })
126                .detach();
127        }
128
129        scheduler.run();
130
131        assert_eq!(
132            trace
133                .borrow()
134                .iter()
135                .filter(|entry| entry.session == 0)
136                .map(|entry| entry.task)
137                .collect::<Vec<_>>(),
138            (0..10).collect::<Vec<_>>()
139        );
140        assert_eq!(
141            trace
142                .borrow()
143                .iter()
144                .filter(|entry| entry.session == 1)
145                .map(|entry| entry.task)
146                .collect::<Vec<_>>(),
147            (0..10).collect::<Vec<_>>()
148        );
149
150        traces.insert(trace.take());
151    });
152
153    assert!(traces.len() > 1, "Expected at least two traces");
154}
155
156#[test]
157fn test_timer_ordering() {
158    TestScheduler::many(1, async |scheduler| {
159        let background = scheduler.background();
160        let futures = FuturesUnordered::new();
161        futures.push(
162            async {
163                background.timer(Duration::from_millis(100)).await;
164                2
165            }
166            .boxed(),
167        );
168        futures.push(
169            async {
170                background.timer(Duration::from_millis(50)).await;
171                1
172            }
173            .boxed(),
174        );
175        futures.push(
176            async {
177                background.timer(Duration::from_millis(150)).await;
178                3
179            }
180            .boxed(),
181        );
182        assert_eq!(futures.collect::<Vec<_>>().await, vec![1, 2, 3]);
183    });
184}
185
186#[test]
187fn test_foreground_task_can_hold_mut_borrow_across_await() {
188    TestScheduler::once(async |scheduler| {
189        let foreground = scheduler.foreground();
190        let (sender, mut receiver) = mpsc::unbounded::<()>();
191
192        foreground
193            .spawn(async move {
194                receiver.next().await;
195            })
196            .detach();
197
198        scheduler.run();
199        sender.unbounded_send(()).unwrap();
200        scheduler.run();
201    });
202}
203
204#[test]
205fn test_send_from_bg_to_fg() {
206    TestScheduler::once(async |scheduler| {
207        let foreground = scheduler.foreground();
208        let background = scheduler.background();
209
210        let (sender, receiver) = oneshot::channel::<i32>();
211
212        background
213            .spawn(async move {
214                sender.send(42).unwrap();
215            })
216            .detach();
217
218        let task = foreground.spawn(async move { receiver.await.unwrap() });
219        let result = task.await;
220        assert_eq!(result, 42);
221    });
222}
223
224#[test]
225fn test_randomize_order() {
226    // Test deterministic mode: different seeds should produce same execution order
227    let mut deterministic_results = HashSet::new();
228    for seed in 0..10 {
229        let config = TestSchedulerConfig {
230            seed,
231            randomize_order: false,
232            ..Default::default()
233        };
234        let order = block_on(capture_execution_order(config));
235        assert_eq!(order.len(), 6);
236        deterministic_results.insert(order);
237    }
238
239    // All deterministic runs should produce the same result
240    assert_eq!(
241        deterministic_results.len(),
242        1,
243        "Deterministic mode should always produce same execution order"
244    );
245
246    // Test randomized mode: different seeds can produce different execution orders
247    let mut randomized_results = HashSet::new();
248    for seed in 0..20 {
249        let config = TestSchedulerConfig::with_seed(seed);
250        let order = block_on(capture_execution_order(config));
251        assert_eq!(order.len(), 6);
252        randomized_results.insert(order);
253    }
254
255    // Randomized mode should produce multiple different execution orders
256    assert!(
257        randomized_results.len() > 1,
258        "Randomized mode should produce multiple different orders"
259    );
260}
261
262async fn capture_execution_order(config: TestSchedulerConfig) -> Vec<String> {
263    let scheduler = Arc::new(TestScheduler::new(config));
264    let foreground = scheduler.foreground();
265    let background = scheduler.background();
266
267    let (sender, receiver) = mpsc::unbounded::<String>();
268
269    // Spawn foreground tasks
270    for i in 0..3 {
271        let mut sender = sender.clone();
272        foreground
273            .spawn(async move {
274                sender.send(format!("fg-{}", i)).await.ok();
275            })
276            .detach();
277    }
278
279    // Spawn background tasks
280    for i in 0..3 {
281        let mut sender = sender.clone();
282        background
283            .spawn(async move {
284                sender.send(format!("bg-{}", i)).await.ok();
285            })
286            .detach();
287    }
288
289    drop(sender); // Close sender to signal no more messages
290    scheduler.run();
291
292    receiver.collect().await
293}
294
295#[test]
296fn test_block() {
297    let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::default()));
298    let (tx, rx) = oneshot::channel();
299
300    // Spawn background task to send value
301    let _ = scheduler
302        .background()
303        .spawn(async move {
304            tx.send(42).unwrap();
305        })
306        .detach();
307
308    // Block on receiving the value
309    let result = scheduler.foreground().block_on(async { rx.await.unwrap() });
310    assert_eq!(result, 42);
311}
312
313#[test]
314#[should_panic(expected = "Parking forbidden.")]
315fn test_parking_panics() {
316    let config = TestSchedulerConfig {
317        capture_pending_traces: true,
318        ..Default::default()
319    };
320    let scheduler = Arc::new(TestScheduler::new(config));
321    scheduler.foreground().block_on(async {
322        let (_tx, rx) = oneshot::channel::<()>();
323        rx.await.unwrap(); // This will never complete
324    });
325}
326
327#[test]
328fn test_block_with_parking() {
329    let config = TestSchedulerConfig {
330        allow_parking: true,
331        ..Default::default()
332    };
333    let scheduler = Arc::new(TestScheduler::new(config));
334    let (tx, rx) = oneshot::channel();
335
336    // Spawn background task to send value
337    let _ = scheduler
338        .background()
339        .spawn(async move {
340            tx.send(42).unwrap();
341        })
342        .detach();
343
344    // Block on receiving the value (will park if needed)
345    let result = scheduler.foreground().block_on(async { rx.await.unwrap() });
346    assert_eq!(result, 42);
347}
348
349#[test]
350fn test_helper_methods() {
351    // Test the once method
352    let result = TestScheduler::once(async |scheduler: Arc<TestScheduler>| {
353        let background = scheduler.background();
354        background.spawn(async { 42 }).await
355    });
356    assert_eq!(result, 42);
357
358    // Test the many method
359    let results = TestScheduler::many(3, async |scheduler: Arc<TestScheduler>| {
360        let background = scheduler.background();
361        background.spawn(async { 10 }).await
362    });
363    assert_eq!(results, vec![10, 10, 10]);
364}
365
366#[test]
367fn test_many_with_arbitrary_seed() {
368    for seed in [0u64, 1, 5, 42] {
369        let mut seeds_seen = Vec::new();
370        let iterations = 3usize;
371
372        for current_seed in seed..seed + iterations as u64 {
373            let scheduler = Arc::new(TestScheduler::new(TestSchedulerConfig::with_seed(
374                current_seed,
375            )));
376            let captured_seed = current_seed;
377            scheduler
378                .foreground()
379                .block_on(async { seeds_seen.push(captured_seed) });
380            scheduler.run();
381        }
382
383        assert_eq!(
384            seeds_seen,
385            (seed..seed + iterations as u64).collect::<Vec<_>>(),
386            "Expected {iterations} iterations starting at seed {seed}"
387        );
388    }
389}
390
391#[test]
392fn test_block_with_timeout() {
393    // Test case: future completes within timeout
394    TestScheduler::once(async |scheduler| {
395        let foreground = scheduler.foreground();
396        let future = future::ready(42);
397        let output = foreground.block_with_timeout(Duration::from_millis(100), future);
398        assert_eq!(output.ok(), Some(42));
399    });
400
401    // Test case: future times out
402    TestScheduler::once(async |scheduler| {
403        // Make timeout behavior deterministic by forcing the timeout tick budget to be exactly 0.
404        // This prevents `block_with_timeout` from making progress via extra scheduler stepping and
405        // accidentally completing work that we expect to time out.
406        scheduler.set_timeout_ticks(0..=0);
407
408        let foreground = scheduler.foreground();
409        let future = future::pending::<()>();
410        let output = foreground.block_with_timeout(Duration::from_millis(50), future);
411        assert!(output.is_err(), "future should not have finished");
412    });
413
414    // Test case: future makes progress via timer but still times out
415    let mut results = BTreeSet::new();
416    TestScheduler::many(if cfg!(miri) { 5 } else { 100 }, async |scheduler| {
417        // Keep the existing probabilistic behavior here (do not force 0 ticks), since this subtest
418        // is explicitly checking that some seeds/timeouts can complete while others can time out.
419        let task = scheduler.background().spawn(async move {
420            Yield { polls: 10 }.await;
421            42
422        });
423        let output = scheduler
424            .foreground()
425            .block_with_timeout(Duration::from_millis(50), task);
426        results.insert(output.ok());
427    });
428    assert_eq!(
429        results.into_iter().collect::<Vec<_>>(),
430        if cfg!(miri) {
431            vec![Some(42)]
432        } else {
433            vec![None, Some(42)]
434        }
435    );
436
437    // Regression test:
438    // A timed-out future must not be cancelled. The returned future should still be
439    // pollable to completion later. We also want to ensure time only advances when we
440    // explicitly advance it (not by yielding).
441    TestScheduler::once(async |scheduler| {
442        // Force immediate timeout: the timeout tick budget is 0 so we will not step or
443        // advance timers inside `block_with_timeout`.
444        scheduler.set_timeout_ticks(0..=0);
445
446        let background = scheduler.background();
447
448        // This task should only complete once time is explicitly advanced.
449        let task = background.spawn({
450            let scheduler = scheduler.clone();
451            async move {
452                scheduler.timer(Duration::from_millis(100)).await;
453                123
454            }
455        });
456
457        // This should time out before we advance time enough for the timer to fire.
458        let timed_out = scheduler
459            .foreground()
460            .block_with_timeout(Duration::from_millis(50), task);
461        assert!(
462            timed_out.is_err(),
463            "expected timeout before advancing the clock enough for the timer"
464        );
465
466        // Now explicitly advance time and ensure the returned future can complete.
467        let mut task = timed_out.err().unwrap();
468        scheduler.advance_clock(Duration::from_millis(100));
469        scheduler.run();
470
471        let output = scheduler.foreground().block_on(&mut task);
472        assert_eq!(output, 123);
473    });
474}
475
476// When calling block, we shouldn't make progress on foreground-spawned futures with the same session id.
477#[test]
478fn test_block_does_not_progress_same_session_foreground() {
479    let mut task2_made_progress_once = false;
480    TestScheduler::many(if cfg!(miri) { 5 } else { 1000 }, async |scheduler| {
481        let foreground1 = scheduler.foreground();
482        let foreground2 = scheduler.foreground();
483
484        let task1 = foreground1.spawn(async move {});
485        let task2 = foreground2.spawn(async move {});
486
487        foreground1.block_on(async {
488            scheduler.yield_random().await;
489            assert!(!task1.is_ready());
490            task2_made_progress_once |= task2.is_ready();
491        });
492
493        task1.await;
494        task2.await;
495    });
496
497    assert!(
498        task2_made_progress_once,
499        "Expected task from different foreground executor to make progress (at least once)"
500    );
501}
502
503struct Yield {
504    polls: usize,
505}
506
507impl Future for Yield {
508    type Output = ();
509
510    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
511        self.polls -= 1;
512        if self.polls == 0 {
513            Poll::Ready(())
514        } else {
515            cx.waker().wake_by_ref();
516            Poll::Pending
517        }
518    }
519}
520
521#[test]
522fn test_nondeterministic_wake_detection() {
523    let config = TestSchedulerConfig {
524        allow_parking: false,
525        ..Default::default()
526    };
527    let scheduler = Arc::new(TestScheduler::new(config));
528
529    // A future that captures its waker and sends it to an external thread
530    struct SendWakerToThread {
531        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
532    }
533
534    impl Future for SendWakerToThread {
535        type Output = ();
536
537        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
538            if let Some(tx) = self.waker_tx.take() {
539                tx.send(cx.waker().clone()).ok();
540            }
541            Poll::Ready(())
542        }
543    }
544
545    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();
546
547    // Get a waker by running a future that sends it
548    scheduler.foreground().block_on(SendWakerToThread {
549        waker_tx: Some(waker_tx),
550    });
551
552    // Spawn a real OS thread that will call wake() on the waker
553    let handle = std::thread::spawn(move || {
554        if let Ok(waker) = waker_rx.recv() {
555            // This should trigger the non-determinism detection
556            waker.wake();
557        }
558    });
559
560    // Wait for the spawned thread to complete
561    handle.join().ok();
562
563    // The non-determinism error should be detected when end_test is called
564    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
565        scheduler.end_test();
566    }));
567    assert!(result.is_err(), "Expected end_test to panic");
568    let panic_payload = result.unwrap_err();
569    let panic_message = panic_payload
570        .downcast_ref::<String>()
571        .map(|s| s.as_str())
572        .or_else(|| panic_payload.downcast_ref::<&str>().copied())
573        .unwrap_or("<unknown panic>");
574    assert!(
575        panic_message.contains("Your test is not deterministic"),
576        "Expected panic message to contain non-determinism error, got: {}",
577        panic_message
578    );
579}
580
581#[test]
582fn test_nondeterministic_wake_allowed_with_parking() {
583    let config = TestSchedulerConfig {
584        allow_parking: true,
585        ..Default::default()
586    };
587    let scheduler = Arc::new(TestScheduler::new(config));
588
589    // A future that captures its waker and sends it to an external thread
590    struct WakeFromExternalThread {
591        waker_sent: bool,
592        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
593    }
594
595    impl Future for WakeFromExternalThread {
596        type Output = ();
597
598        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
599            if !self.waker_sent {
600                self.waker_sent = true;
601                if let Some(tx) = self.waker_tx.take() {
602                    tx.send(cx.waker().clone()).ok();
603                }
604                Poll::Pending
605            } else {
606                Poll::Ready(())
607            }
608        }
609    }
610
611    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();
612
613    // Spawn a real OS thread that will call wake() on the waker
614    std::thread::spawn(move || {
615        if let Ok(waker) = waker_rx.recv() {
616            // With allow_parking, this should NOT panic
617            waker.wake();
618        }
619    });
620
621    // This should complete without panicking
622    scheduler.foreground().block_on(WakeFromExternalThread {
623        waker_sent: false,
624        waker_tx: Some(waker_tx),
625    });
626}
627
628#[test]
629fn test_nondeterministic_waker_drop_detection() {
630    let config = TestSchedulerConfig {
631        allow_parking: false,
632        ..Default::default()
633    };
634    let scheduler = Arc::new(TestScheduler::new(config));
635
636    // A future that captures its waker and sends it to an external thread
637    struct SendWakerToThread {
638        waker_tx: Option<std::sync::mpsc::Sender<Waker>>,
639    }
640
641    impl Future for SendWakerToThread {
642        type Output = ();
643
644        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
645            if let Some(tx) = self.waker_tx.take() {
646                tx.send(cx.waker().clone()).ok();
647            }
648            Poll::Ready(())
649        }
650    }
651
652    let (waker_tx, waker_rx) = std::sync::mpsc::channel::<Waker>();
653
654    // Get a waker by running a future that sends it
655    scheduler.foreground().block_on(SendWakerToThread {
656        waker_tx: Some(waker_tx),
657    });
658
659    // Spawn a real OS thread that will drop the waker without calling wake
660    let handle = std::thread::spawn(move || {
661        if let Ok(waker) = waker_rx.recv() {
662            // This should trigger the non-determinism detection on drop
663            drop(waker);
664        }
665    });
666
667    // Wait for the spawned thread to complete
668    handle.join().ok();
669
670    // The non-determinism error should be detected when end_test is called
671    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
672        scheduler.end_test();
673    }));
674    assert!(result.is_err(), "Expected end_test to panic");
675    let panic_payload = result.unwrap_err();
676    let panic_message = panic_payload
677        .downcast_ref::<String>()
678        .map(|s| s.as_str())
679        .or_else(|| panic_payload.downcast_ref::<&str>().copied())
680        .unwrap_or("<unknown panic>");
681    assert!(
682        panic_message.contains("Your test is not deterministic"),
683        "Expected panic message to contain non-determinism error, got: {}",
684        panic_message
685    );
686}
687
688#[test]
689fn test_background_priority_scheduling() {
690    use parking_lot::Mutex;
691
692    // Run many iterations to get statistical significance
693    let mut high_before_low_count = 0;
694    let iterations = if cfg!(miri) { 5 } else { 100 };
695
696    for seed in 0..iterations {
697        let config = TestSchedulerConfig::with_seed(seed);
698        let scheduler = Arc::new(TestScheduler::new(config));
699        let background = scheduler.background();
700
701        let execution_order = Arc::new(Mutex::new(Vec::new()));
702
703        // Spawn low priority tasks first
704        for i in 0..3 {
705            let order = execution_order.clone();
706            background
707                .spawn_with_priority(Priority::Low, async move {
708                    order.lock().push(format!("low-{}", i));
709                })
710                .detach();
711        }
712
713        // Spawn high priority tasks second
714        for i in 0..3 {
715            let order = execution_order.clone();
716            background
717                .spawn_with_priority(Priority::High, async move {
718                    order.lock().push(format!("high-{}", i));
719                })
720                .detach();
721        }
722
723        scheduler.run();
724
725        // Count how many high priority tasks ran in the first half
726        let order = execution_order.lock();
727        let high_in_first_half = order
728            .iter()
729            .take(3)
730            .filter(|s| s.starts_with("high"))
731            .count();
732
733        if high_in_first_half >= 2 {
734            high_before_low_count += 1;
735        }
736    }
737
738    // High priority tasks should tend to run before low priority tasks
739    // With weights of 60 vs 10, high priority should dominate early execution
740    assert!(
741        high_before_low_count > iterations / 2,
742        "Expected high priority tasks to run before low priority tasks more often. \
743         Got {} out of {} iterations",
744        high_before_low_count,
745        iterations
746    );
747}
748
749#[test]
750fn test_spawn_dedicated_basic_round_trip() {
751    let result = TestScheduler::once(async |scheduler| {
752        scheduler
753            .background()
754            .spawn_dedicated(|_executor| async { 42 })
755            .await
756    });
757    assert_eq!(result, 42);
758}
759
760#[test]
761fn test_spawn_dedicated_not_send_future() {
762    let result = TestScheduler::once(async |scheduler| {
763        scheduler
764            .background()
765            .spawn_dedicated(|_executor| async move {
766                // `Rc<RefCell<_>>` is `!Send`. If `spawn_dedicated` required
767                // the returned future to be `Send`, this wouldn't compile.
768                let state = Rc::new(RefCell::new(0_i32));
769                for _ in 0..5 {
770                    *state.borrow_mut() += 1;
771                }
772                *state.borrow()
773            })
774            .await
775    });
776    assert_eq!(result, 5);
777}
778
779#[test]
780fn test_spawn_dedicated_send_closure_captures() {
781    use parking_lot::Mutex;
782
783    let observed = TestScheduler::once(async |scheduler| {
784        let shared = Arc::new(Mutex::new(0_i32));
785        let shared_for_closure = shared.clone();
786        let returned = scheduler
787            .background()
788            .spawn_dedicated(move |_executor| {
789                // `shared_for_closure` crossed the `Send` boundary of the
790                // closure; we then mutate it from inside the !Send future.
791                let local = shared_for_closure;
792                async move {
793                    *local.lock() = 7;
794                }
795            })
796            .await;
797        let _: () = returned;
798        *shared.lock()
799    });
800    assert_eq!(observed, 7);
801}
802
803#[test]
804fn test_spawn_dedicated_inner_spawn_local() {
805    let result = TestScheduler::once(async |scheduler| {
806        scheduler
807            .background()
808            .spawn_dedicated(|executor| async move {
809                // The provided executor can spawn additional `!Send` work
810                // onto the same dedicated session.
811                let inner = Rc::new(RefCell::new(0_i32));
812                let inner_for_child = inner.clone();
813                let child = executor.spawn(async move {
814                    *inner_for_child.borrow_mut() = 99;
815                    *inner_for_child.borrow()
816                });
817                child.await
818            })
819            .await
820    });
821    assert_eq!(result, 99);
822}
823
824#[test]
825fn test_spawn_dedicated_determinism_under_many() {
826    use parking_lot::Mutex;
827
828    let outcomes = TestScheduler::many(if cfg!(miri) { 4 } else { 20 }, async |scheduler| {
829        let trace = Arc::new(Mutex::new(Vec::<u32>::new()));
830
831        let background = scheduler.background();
832        let mut tasks = Vec::new();
833        for id in 0..4_u32 {
834            let trace = trace.clone();
835            let task = background.spawn_dedicated(move |executor| async move {
836                for step in 0..3 {
837                    trace.lock().push(id * 100 + step);
838                    executor.spawn(async {}).await;
839                }
840                id
841            });
842            tasks.push(task);
843        }
844
845        let mut outputs = Vec::new();
846        for task in tasks {
847            outputs.push(task.await);
848        }
849
850        (trace.lock().clone(), outputs)
851    });
852
853    // Re-running with the same seed should produce the same trace. Run a
854    // second pass with identical seeds and compare to the first.
855    let outcomes_replay = TestScheduler::many(if cfg!(miri) { 4 } else { 20 }, async |scheduler| {
856        let trace = Arc::new(Mutex::new(Vec::<u32>::new()));
857
858        let background = scheduler.background();
859        let mut tasks = Vec::new();
860        for id in 0..4_u32 {
861            let trace = trace.clone();
862            let task = background.spawn_dedicated(move |executor| async move {
863                for step in 0..3 {
864                    trace.lock().push(id * 100 + step);
865                    executor.spawn(async {}).await;
866                }
867                id
868            });
869            tasks.push(task);
870        }
871
872        let mut outputs = Vec::new();
873        for task in tasks {
874            outputs.push(task.await);
875        }
876
877        (trace.lock().clone(), outputs)
878    });
879
880    assert_eq!(
881        outcomes, outcomes_replay,
882        "per-seed outcomes should be reproducible"
883    );
884
885    // Sanity: at least one seed produced a non-monotonic trace,
886    // demonstrating that dedicated tasks really do interleave under the
887    // scheduler's randomization.
888    let any_interleaved = outcomes.iter().any(|(trace, _)| {
889        trace
890            .windows(2)
891            .any(|window| window[0] / 100 != window[1] / 100)
892    });
893    assert!(
894        any_interleaved,
895        "expected at least one seed to interleave dedicated tasks"
896    );
897}
898
899#[test]
900fn test_spawn_dedicated_dropping_task_cancels_future() {
901    use parking_lot::Mutex;
902
903    let counter_after = TestScheduler::once(async |scheduler| {
904        let counter = Arc::new(Mutex::new(0_u32));
905        let (resume_tx, resume_rx) = oneshot::channel::<()>();
906
907        let task = {
908            let counter = counter.clone();
909            scheduler
910                .background()
911                .spawn_dedicated(move |_executor| async move {
912                    *counter.lock() = 1;
913                    // Park here until the test resumes us. If the task is
914                    // dropped before this resolves, the second assignment
915                    // below must never happen.
916                    let _ = resume_rx.await;
917                    *counter.lock() = 2;
918                })
919        };
920
921        // Let the dedicated future make its first observable step.
922        scheduler.run();
923        assert_eq!(*counter.lock(), 1);
924
925        // Cancel by dropping the root task, then unblock the parked oneshot.
926        // The future must not advance past the await: counter stays at 1.
927        drop(task);
928        let _ = resume_tx.send(());
929        scheduler.run();
930
931        *counter.lock()
932    });
933
934    assert_eq!(
935        counter_after, 1,
936        "dropping the dedicated task must cancel the root future before its second write"
937    );
938}
939
940#[test]
941fn test_spawn_dedicated_detached_child_runs_after_root_completes() {
942    use parking_lot::Mutex;
943
944    let child_ran = TestScheduler::once(async |scheduler| {
945        let child_ran = Arc::new(Mutex::new(false));
946
947        let task = {
948            let child_ran = child_ran.clone();
949            scheduler
950                .background()
951                .spawn_dedicated(move |executor| async move {
952                    executor
953                        .spawn(async move {
954                            *child_ran.lock() = true;
955                        })
956                        .detach();
957                    // Root returns immediately, before the child has had a
958                    // chance to run.
959                })
960        };
961
962        task.await;
963
964        // Drain the dedicated session. The detached child must run.
965        scheduler.run();
966
967        *child_ran.lock()
968    });
969
970    assert!(
971        child_ran,
972        "detached child must complete after the root, not be cancelled with it"
973    );
974}
975
976// The production smoke test for `spawn_dedicated` lives in the `gpui` crate
977// alongside `PlatformScheduler`, which is the real production implementation
978// of the `Scheduler` trait. See `crates/gpui/src/platform_scheduler.rs`.
979