arty 0.6.0

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

//! Asynchronous worker execution loop.

use std::fmt::Debug;
use std::marker::PhantomData;
use std::ops::ControlFlow;
use std::time::{Duration, Instant};

use arty_executor::{CycleOutcome, Executor, TaskSet};
use performables::arc::Arc;
use performables::sync::channel;
use thread_aware::Thread;
use tick::SimpleClock;
use tick::runtime::{ClockDriver, InactiveClock};

use crate::io::{Io, RuntimeDriverHost};
use crate::runtime::config::MonitorOptions;
use crate::runtime::dispatch::DispatcherClient;
use crate::runtime::shutdown::{Shutdown, ShutdownState};
use crate::runtime::worker::asynchronous::{Command, LocalWorker, Signal};
#[cfg(test)]
use crate::runtime::worker::blocking;
use crate::task::Context;

const COMMANDS_PER_CYCLE: usize = 256;

fn suspend_wait(deadline: Instant, now: Instant) -> Duration {
    deadline.saturating_duration_since(now)
}

fn can_suspend(cycle_outcome: &CycleOutcome, batch_exhausted: bool) -> bool {
    matches!(cycle_outcome, CycleOutcome::Suspend) && !batch_exhausted
}

/// The async worker has exclusive use of one runtime thread. Multiple worker
/// threads can share a processor when the exact worker count exceeds available
/// processor capacity. The worker executes async tasks that are woken by local
/// or remote futures, incoming commands, and timers.
///
/// This type contains the internal worker state. Runtime components send work
/// and signals to it through channels.
///
/// # Ownership
///
/// This type is exclusively owned by the thread entry point.
///
/// # Lifecycle
///
/// The worker has these lifecycle stages:
///
/// 1. Startup: Bootstrap creates the thread and worker from the
///    [`RuntimeBuilder`](crate::runtime::RuntimeBuilder) configuration. Bootstrap
///    and the worker coordinate initialization until the worker is ready.
/// 2. Running: The worker processes tasks, advances timers, and waits for
///    commands.
/// 3. Shutting down: The worker rejects queued and new submissions, stops
///    blocking work, cancels async tasks, and runs final timer callbacks.
/// 4. Dead: The worker completed shutdown and its thread terminated.
///
/// A running worker stops after
/// [`RuntimeControl::request_stop`](crate::runtime::RuntimeControl::request_stop),
/// `Runtime::stop`, or `Runtime` destruction.
///
/// # Thread safety
///
/// This type is single-threaded. It only exists on its designated worker thread.
#[derive(Debug)]
pub(crate) struct Worker {
    // Only top-level worker logic uses this field.
    // Shutdown drops the receiver. Later submissions then fail and discard their tasks.
    command_rx: Option<channel::Receiver<Command>>,

    // Dropped when shutdown starts so no more tasks can be enqueued and worker-local
    // services can release resources needed by shutdown.
    context: Option<Context>,

    shutdown: Shutdown,

    // Becomes None during shutdown when the executor is dropped to cancel and release its tasks.
    executor: Option<Executor>,

    // Connected to the `executor`. Submissions after shutdown resolve to an executor-shutdown
    // JoinError rather than registering a task.
    tasks: TaskSet,

    // This drives moves the timers registered with the clock forward.
    clock_driver: ClockDriver,

    // Cycle-local timestamps and wait budgets use precise system time.
    cycle_clock: SimpleClock,
    signal: Arc<Signal>,
    monitor_options: MonitorOptions,
    _single_threaded: PhantomData<*const ()>,
}

impl Worker {
    #[expect(
        clippy::too_many_arguments,
        reason = "worker construction wires its independently owned runtime subsystems"
    )]
    pub(crate) fn new(
        command_rx: channel::Receiver<Command>,
        dispatcher: DispatcherClient,
        thread: Thread,
        io: Io,
        clock: InactiveClock,
        signal: Arc<Signal>,
        monitor_options: MonitorOptions,
        panic_policy: crate::runtime::PanicPolicy,
    ) -> Self {
        let worker_waker = Signal::waker(&signal);
        let shutdown = dispatcher.shutdown().clone();
        let (clock, clock_driver) = clock.activate(worker_waker.clone());
        let cycle_clock = SimpleClock::new_system();

        let executor = Executor::builder()
            .owner_waker(worker_waker)
            .thread(thread.clone())
            .panic_policy(panic_policy.into_executor_policy())
            .build();
        let tasks = executor.tasks();
        let executor_shutdown = executor.shutdown();

        let context = Context::new(dispatcher, clock, io, thread);
        LocalWorker::register_scheduler(context.clone(), tasks.clone(), executor_shutdown);

        Self {
            command_rx: Some(command_rx),
            context: Some(context),
            shutdown,
            executor: Some(executor),
            tasks,
            clock_driver,
            cycle_clock,
            signal,
            monitor_options,
            _single_threaded: PhantomData,
        }
    }

    /// Worker thread entry point, invoked by bootstrap after startup coordination completes.
    /// The configuration-facing `RuntimeBuilder` delegates construction to bootstrap.
    /// After this method returns, the thread will end.
    pub(crate) fn run(&mut self) {
        // Keep the worker-owned policy available at the boundary where monitoring
        // will be attached in a later stage.
        let _monitor_policy = self.monitor_options.policy();
        self.execute_phase();
    }

    #[cfg_attr(test, mutants::skip)] // Critical for code execution to occur in async contexts.
    fn execute_phase(&mut self) {
        while matches!(self.execute_step(), ControlFlow::Continue(())) {}
    }

    fn execute_step(&mut self) -> ControlFlow<()> {
        let action = self.prepare_cycle();
        let max_wait = match action {
            CycleAction::Shutdown => return self.finish_shutdown(),
            CycleAction::WaitForIo(deadline) => self.io_wait(deadline),
            CycleAction::Poll | CycleAction::WaitForSignal(_) => Duration::ZERO,
        };

        if RuntimeDriverHost::execute_cycle_current(max_wait).is_err() {
            return self.finish_shutdown();
        }

        if let CycleAction::WaitForSignal(deadline) = action {
            match deadline {
                Some(deadline) => self.signal.wait(self.wait_until(deadline)),
                None => self.signal.wait_for_notification(),
            }
        }

        ControlFlow::Continue(())
    }

    fn prepare_cycle(&mut self) -> CycleAction {
        let batch_exhausted = self.process_commands();

        if matches!(self.shutdown.state(), ShutdownState::Started) {
            return CycleAction::Shutdown;
        }

        let executor = self
            .executor
            .as_mut()
            .expect("executor removal publishes shutdown, which is handled above");

        let cycle_outcome = executor.execute_cycle();

        if let CycleOutcome::Shutdown(cause) = cycle_outcome {
            self.shutdown.begin(cause);
            return CycleAction::Shutdown;
        }

        let cycle_finished = self.cycle_clock.instant();
        let next_timer = self.clock_driver.advance_timers(cycle_finished).ok().flatten();
        if matches!(self.shutdown.state(), ShutdownState::Started) {
            return CycleAction::Shutdown;
        }

        if !can_suspend(&cycle_outcome, batch_exhausted) {
            CycleAction::Poll
        } else if RuntimeDriverHost::has_primary_current() {
            CycleAction::WaitForIo(next_timer)
        } else {
            CycleAction::WaitForSignal(next_timer)
        }
    }

    fn io_wait(&self, deadline: Option<Instant>) -> Duration {
        if self.signal.take_notification() {
            return Duration::ZERO;
        }

        deadline.map_or(Duration::MAX, |deadline| self.wait_until(deadline))
    }

    fn wait_until(&self, deadline: Instant) -> Duration {
        suspend_wait(deadline, self.cycle_clock.instant())
    }

    fn finish_shutdown(&mut self) -> ControlFlow<()> {
        self.begin_shutdown();
        // Final task cancellation can register a timer after the cycle sample.
        let now = self.cycle_clock.instant();
        _ = self.clock_driver.advance_timers(now);
        ControlFlow::Break(())
    }

    #[cfg_attr(test, mutants::skip)] // Mutation testing requires a fine level of control over what is in the channel, which is too bothersome just for mutation testing.
    fn process_commands(&mut self) -> bool {
        let Some(context) = self.context.as_ref() else {
            debug_assert!(self.command_rx.is_none());

            // We are shutting down and have dropped worker-local services.
            // The command channel is also closed, so no more commands can be received.
            return false;
        };

        let command_rx = self.command_rx.as_ref().expect("context and command channel are dropped together");

        for _ in 0..COMMANDS_PER_CYCLE {
            if matches!(self.shutdown.state(), ShutdownState::Started) {
                self.begin_shutdown();
                return false;
            }
            match command_rx.try_recv() {
                Ok(command) => match command {
                    Command::EnqueueTask { future_factory } => {
                        let _registered = future_factory.into_inner().register(context.clone(), &self.tasks);
                    }
                    Command::AccessIo { access } => {
                        access.into_inner()(context);
                    }
                    Command::Shutdown => {
                        self.begin_shutdown();

                        // Shutdown closes the command channel. No more commands can arrive.
                        return false;
                    }
                },
                Err(error) => {
                    // Every initialized worker retains its own dispatcher until shutdown.
                    assert!(error.is_empty(), "async worker command channel disconnected without shutdown");
                    return false;
                }
            }
        }
        // An exhausted batch can leave commands queued. Read the queue again before parking.
        true
    }

    #[cfg_attr(test, mutants::skip)] // If mutated, shutdown process will never finish - will hang.
    fn begin_shutdown(&mut self) {
        let Some(executor) = self.executor.take() else {
            return;
        };

        let cause = self
            .context
            .as_ref()
            .expect("a running worker retains its context until shutdown")
            .dispatcher()
            .stop(self.shutdown.cause().unwrap_or_default());

        LocalWorker::clear_scheduler();
        // Drop worker-local services first. Their futures and waiters can hold
        // resources that block executor shutdown.
        self.context = None;

        // The command channel is now closed. No more commands can arrive.
        self.command_rx = None;

        // Safe executor shutdown synchronously cancels and releases all remaining task storage.
        executor.shutdown().trigger(cause);
        drop(executor);
        let driver_host = RuntimeDriverHost::uninstall();
        drop(driver_host);
    }
}

impl Drop for Worker {
    #[cfg_attr(test, mutants::skip)] // Worker-owned services must retire before the thread can finish shutdown.
    fn drop(&mut self) {
        if self.executor.is_some() {
            self.begin_shutdown();
        }
    }
}

#[derive(Clone, Copy, Debug)]
enum CycleAction {
    Poll,
    WaitForIo(Option<Instant>),
    WaitForSignal(Option<Instant>),
    Shutdown,
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))] // Test scaffolding is not runtime behavior.
mod tests {
    use std::panic::{AssertUnwindSafe, catch_unwind};
    #[cfg(not(miri))]
    use std::pin::Pin;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::sync::mpsc;
    #[cfg(not(miri))]
    use std::task::Wake;
    use std::task::Waker;
    use std::thread;

    use arty_executor::{ShutdownCause, TaskCompletion, TaskOptions, TaskPanicPayload, discard_task_panic_payload};
    use arty_io_core::{Cycle, Driver, DriverError, DriverOptions, IoContext};
    #[cfg(not(miri))]
    use events_once::BoxedSender;
    use events_once::{Event, IntoValueError};
    use nonempty::NonEmpty;
    use testing_aids::{TEST_TIMEOUT, async_test};
    use thread_aware::{ThreadAware, ThreadBuilder};
    use tick::ClockControl;

    use super::*;
    use crate::runtime::config::MonitorPolicy;
    use crate::runtime::dispatch::{DispatcherCore, WorkerEndpoint};
    use crate::runtime::worker::asynchronous::{QueuedFactory, QueuedIoAccess};
    use crate::task::preparation::RemoteFutureFactory;

    fn test_blocking_worker() -> (Arc<blocking::Worker>, Shutdown) {
        let shutdown = Shutdown::default();
        (
            blocking::Worker::new(blocking::Pool::new(None), shutdown.clone(), blocking::PanicPolicy::Ignore),
            shutdown,
        )
    }

    trait TestTaskSetExt {
        fn add_test<F, R>(&self, future: F)
        where
            F: IntoFuture<Output = R> + 'static,
            R: 'static;
    }

    impl TestTaskSetExt for TaskSet {
        fn add_test<F, R>(&self, future: F)
        where
            F: IntoFuture<Output = R> + 'static,
            R: 'static,
        {
            self.add(future, TaskOptions::default(), DiscardCompletion);
        }
    }

    #[derive(Debug)]
    struct DiscardCompletion;

    impl<R: 'static> TaskCompletion<R> for DiscardCompletion {
        fn completed(self, result: R) {
            if let Err(panic) = catch_unwind(AssertUnwindSafe(|| drop(result))) {
                discard_task_panic_payload(panic);
            }
        }

        fn panicked(self, panic: TaskPanicPayload, _cause: Option<ShutdownCause>) {
            discard_task_panic_payload(panic);
        }

        fn cancelled(self, _cause: ShutdownCause) {}
    }

    fn enqueue(factory: RemoteFutureFactory) -> Command {
        Command::EnqueueTask {
            future_factory: QueuedFactory::new(factory),
        }
    }

    #[derive(Clone)]
    struct WaitRecordingContext(Arc<std::sync::Mutex<Vec<Duration>>>);

    impl ThreadAware for WaitRecordingContext {
        fn relocate(&mut self, _source: Option<&Thread>, _destination: &Thread) {}
    }

    impl IoContext for WaitRecordingContext {
        type ContextOptions = Arc<std::sync::Mutex<Vec<Duration>>>;
        type Driver = WaitRecordingDriver;

        fn context(options: Self::ContextOptions) -> Self {
            Self(options)
        }

        fn driver(&mut self, _options: DriverOptions) -> Result<Self::Driver, DriverError> {
            Ok(WaitRecordingDriver(Arc::clone(&self.0)))
        }
    }

    struct WaitRecordingDriver(Arc<std::sync::Mutex<Vec<Duration>>>);

    impl Driver for WaitRecordingDriver {
        fn waker(&self) -> Waker {
            Waker::noop().clone()
        }

        fn execute_cycle(&mut self, cycle: &mut Cycle) -> Result<(), DriverError> {
            self.0.lock().unwrap().push(cycle.max_wait());
            Ok(())
        }

        fn shutdown(self) -> Result<(), DriverError> {
            Ok(())
        }
    }

    fn test_worker(
        command_tx: channel::Sender<Command>,
        command_rx: channel::Receiver<Command>,
        blocking: (Arc<blocking::Worker>, Shutdown),
        clock: InactiveClock,
        signal: Arc<Signal>,
    ) -> Worker {
        test_worker_with_monitor_options(command_tx, command_rx, blocking, clock, signal, MonitorOptions::default())
    }

    fn test_worker_with_monitor_options(
        command_tx: channel::Sender<Command>,
        command_rx: channel::Receiver<Command>,
        blocking: (Arc<blocking::Worker>, Shutdown),
        clock: InactiveClock,
        signal: Arc<Signal>,
        monitor_options: MonitorOptions,
    ) -> Worker {
        let (blocking_worker, shutdown) = blocking;
        let thread = ThreadBuilder::default().build(thread::current().id());
        let endpoint = WorkerEndpoint {
            command_tx,
            waker: Signal::waker(&signal),
            thread: thread.clone(),
            blocking_worker: Arc::clone(&blocking_worker),
        };
        let dispatcher = DispatcherClient::new(DispatcherCore::new(NonEmpty::new(endpoint), shutdown));
        let io_runtime = arty_io_host::IoRuntime::new(observed::Sink::noop());
        let io = crate::io::Io::new(io_runtime.clone(), dispatcher.clone(), thread.clone());
        if !LocalWorker::is_current() {
            LocalWorker::attach();
        }
        let driver_waker = crate::io::RuntimeDriverHost::install(&io_runtime, dispatcher.clone(), &thread, Signal::waker(&signal));
        signal.set_driver_waker(driver_waker);

        Worker::new(
            command_rx,
            dispatcher,
            thread,
            io,
            clock,
            signal,
            monitor_options,
            crate::runtime::PanicPolicy::default(),
        )
    }

    fn replenishing_command(sender: channel::Sender<Command>, processed: Arc<AtomicUsize>) -> Command {
        enqueue(RemoteFutureFactory::new(move |_, _| {
            processed.fetch_add(1, Ordering::Relaxed);
            sender.send(replenishing_command(sender.clone(), Arc::clone(&processed))).unwrap();
        }))
    }

    #[cfg(not(miri))]
    struct GeneratedWorkerScenario {
        command_tx: channel::Sender<Command>,
        ready_task: Option<BoxedSender<()>>,
        control: ClockControl,
        task_done: Arc<AtomicBool>,
        timer_done: Arc<AtomicBool>,
        factory_invocations: Arc<AtomicUsize>,
        forbidden_factory_invocations: Arc<AtomicUsize>,
        signal: Arc<Signal>,
        worker: Worker,
        timer_ready: bool,
        timer_advanced_by_step: bool,
        queued_shutdown: bool,
        queued_factories: usize,
        sent_factories: usize,
        progress_failure: Option<&'static str>,
    }

    #[cfg(not(miri))]
    impl GeneratedWorkerScenario {
        const MAX_OPERATIONS: usize = 12;
        const MAX_CLEANUP_STEPS: usize = Self::MAX_OPERATIONS + 8;

        fn new() -> Self {
            let (command_tx, command_rx) = channel::unbounded();
            let (ready_task, task_ready) = Event::boxed();
            let control = ClockControl::new();
            let task_done = Arc::new(AtomicBool::new(false));
            let timer_done = Arc::new(AtomicBool::new(false));
            let factory_invocations = Arc::new(AtomicUsize::new(0));
            let forbidden_factory_invocations = Arc::new(AtomicUsize::new(0));
            let signal = Arc::new(Signal::default());
            let task_observer = Arc::clone(&task_done);
            let timer_observer = Arc::clone(&timer_done);

            let worker = test_worker(
                command_tx.clone(),
                command_rx,
                test_blocking_worker(),
                control.clone().into(),
                Arc::clone(&signal),
            );
            worker.tasks.add_test(async move {
                task_ready.await.unwrap();
                task_observer.store(true, Ordering::Relaxed);
            });
            let clock = worker.context.as_ref().unwrap().clock().clone();
            worker.tasks.add_test(async move {
                clock.delay(Duration::from_secs(1)).await;
                timer_observer.store(true, Ordering::Relaxed);
            });
            let mut scenario = Self {
                command_tx,
                ready_task: Some(ready_task),
                control,
                task_done,
                timer_done,
                factory_invocations,
                forbidden_factory_invocations,
                signal,
                worker,
                timer_ready: false,
                timer_advanced_by_step: false,
                queued_shutdown: false,
                queued_factories: 0,
                sent_factories: 0,
                progress_failure: None,
            };
            scenario.setup_step("worker stopped while polling controlled work");
            scenario
        }

        fn setup_step(&mut self, failure: &'static str) {
            Signal::waker(&self.signal).wake_by_ref();
            if self.worker.execute_step().is_break() {
                self.progress_failure = Some(failure);
            }
        }

        fn apply(&mut self, operation: u8) {
            match operation % 5 {
                0 => self.queue_factory_batch(operation),
                1 => {
                    if let Some(ready_task) = self.ready_task.take() {
                        ready_task.send(());
                    }
                }
                2 => self.ready_timer(),
                3 => self.step(),
                _ => self.queue_shutdown(),
            }
        }

        fn queue_factory_batch(&mut self, operation: u8) {
            let batch_size = match (operation / 6) % 4 {
                0 => 1,
                1 => COMMANDS_PER_CYCLE - 1,
                2 => COMMANDS_PER_CYCLE,
                _ => COMMANDS_PER_CYCLE + 1,
            };
            for _ in 0..batch_size {
                let invocations = Arc::clone(&self.factory_invocations);
                let forbidden = Arc::clone(&self.forbidden_factory_invocations);
                let forbidden_after_queued_shutdown = self.queued_shutdown;
                if self
                    .command_tx
                    .send(enqueue(RemoteFutureFactory::new(move |_, _| {
                        invocations.fetch_add(1, Ordering::Relaxed);
                        if forbidden_after_queued_shutdown {
                            forbidden.fetch_add(1, Ordering::Relaxed);
                        }
                    })))
                    .is_err()
                {
                    break;
                }
                self.queued_factories += 1;
                self.sent_factories += 1;
            }
        }

        fn ready_timer(&mut self) {
            if !self.timer_ready {
                self.control.advance(Duration::from_secs(1));
                self.timer_ready = true;
            }
        }

        fn step(&mut self) {
            if self.worker.executor.is_none() {
                return;
            }
            Signal::waker(&self.signal).wake_by_ref();
            let full_batch = self.queued_factories >= COMMANDS_PER_CYCLE && !self.queued_shutdown;
            let task_should_progress =
                full_batch && self.ready_task.is_none() && !self.task_done.load(Ordering::Relaxed) && self.progress_failure.is_none();
            let timer_should_progress =
                full_batch && self.timer_advanced_by_step && !self.timer_done.load(Ordering::Relaxed) && self.progress_failure.is_none();
            let running = self.worker.execute_step().is_continue();
            if running && task_should_progress && !self.task_done.load(Ordering::Relaxed) {
                self.progress_failure = Some("ready task must progress between full command batches");
            }
            if running && timer_should_progress && !self.timer_done.load(Ordering::Relaxed) {
                self.progress_failure = Some("ready timer must progress between full command batches");
            }
            if full_batch {
                self.queued_factories -= COMMANDS_PER_CYCLE;
            } else if !self.queued_shutdown {
                self.queued_factories = 0;
            }
            self.timer_advanced_by_step |= self.timer_ready;
        }

        fn queue_shutdown(&mut self) {
            if !self.queued_shutdown {
                self.worker.shutdown.begin(ShutdownCause::default());
            }
            if !self.queued_shutdown && self.command_tx.send(Command::Shutdown).is_ok() {
                self.queued_shutdown = true;
            }
        }

        fn finish(mut self) {
            if let Some(ready_task) = self.ready_task.take() {
                ready_task.send(());
            }
            self.ready_timer();
            self.queue_shutdown();

            let mut stopped = self.worker.executor.is_none();
            for _ in 0..Self::MAX_CLEANUP_STEPS {
                if stopped {
                    break;
                }
                Signal::waker(&self.signal).wake_by_ref();
                stopped = self.worker.execute_step().is_break();
            }
            if !stopped {
                std::mem::forget(self.worker);
                panic!("bounded cleanup must complete worker shutdown");
            }
            assert!(self.progress_failure.is_none(), "{}", self.progress_failure.unwrap_or_default());
            assert_eq!(
                self.forbidden_factory_invocations.load(Ordering::Relaxed),
                0,
                "factories queued after shutdown or pending at shutdown publication must not be invoked"
            );
            assert!(self.factory_invocations.load(Ordering::Relaxed) <= self.sent_factories);
        }
    }

    #[test]
    fn suspend_budget_requires_suspend_outcome_and_non_exhausted_batch() {
        assert!(can_suspend(&CycleOutcome::Suspend, false));
        assert!(!can_suspend(&CycleOutcome::Suspend, true));
        assert!(!can_suspend(&CycleOutcome::Continue, false));
        assert!(!can_suspend(&CycleOutcome::Continue, true));
        assert!(!can_suspend(&CycleOutcome::Shutdown(ShutdownCause::default()), false));
        assert!(!can_suspend(&CycleOutcome::Shutdown(ShutdownCause::default()), true));
    }

    #[test]
    fn private_suspend_wait_uses_current_time() {
        // The exact wait calculation is private worker-loop policy; public timer wake and
        // completion behavior is covered by integration tests in `tests/time.rs`.
        let now = Instant::now();

        assert_eq!(
            suspend_wait(now + Duration::from_secs(2), now + Duration::from_secs(1)),
            Duration::from_secs(1)
        );
        assert_eq!(suspend_wait(now + Duration::from_millis(20), now), Duration::from_millis(20));
        assert_eq!(
            suspend_wait(
                now.checked_sub(Duration::from_millis(1))
                    .expect("the current instant supports a 1 ms subtraction"),
                now
            ),
            Duration::ZERO
        );
    }

    #[test]
    fn worker_preserves_monitor_options() {
        let (command_tx, command_rx) = channel::unbounded();
        let signal = Arc::new(Signal::default());
        let worker = test_worker_with_monitor_options(
            command_tx,
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            signal,
            MonitorPolicy::disabled().into(),
        );

        assert!(!worker.monitor_options.policy().is_enabled());
    }

    #[cfg(not(miri))]
    #[test]
    fn worker_without_primary_waits_for_timer_or_notification() {
        let (worker_ready, worker_setup) = mpsc::channel();
        let (step_done, step_completion) = mpsc::channel();
        let worker_thread = thread::spawn(move || {
            let (command_tx, command_rx) = channel::unbounded();
            let signal = Arc::new(Signal::default());
            let mut worker = test_worker(
                command_tx.clone(),
                command_rx,
                test_blocking_worker(),
                InactiveClock::new_isolated(),
                Arc::clone(&signal),
            );
            let clock = worker.context.as_ref().unwrap().clock().clone();
            let mut delay = Box::pin(clock.delay(Duration::from_mins(1)));
            assert!(delay.as_mut().poll(&mut std::task::Context::from_waker(Waker::noop())).is_pending());
            signal.wait(Duration::ZERO);
            worker_ready.send(Signal::waker(&signal)).unwrap();

            assert!(worker.execute_step().is_continue());
            step_done.send(()).unwrap();

            drop(delay);
            drop(clock);
            command_tx.send(Command::Shutdown).unwrap();
            worker.run();
        });
        let worker_waker = worker_setup.recv_timeout(TEST_TIMEOUT).unwrap();

        assert_eq!(
            step_completion.recv_timeout(Duration::from_millis(50)),
            Err(mpsc::RecvTimeoutError::Timeout)
        );
        worker_waker.wake_by_ref();
        step_completion.recv_timeout(TEST_TIMEOUT).unwrap();
        worker_thread.join().unwrap();
    }

    #[cfg(not(miri))]
    #[test]
    fn first_cross_thread_timer_interrupts_established_idle_wait() {
        struct TimerWake(mpsc::Sender<()>);

        impl Wake for TimerWake {
            fn wake(self: std::sync::Arc<Self>) {
                _ = self.0.send(());
            }
        }

        let (wait_entered, wait_entry) = mpsc::channel();
        let (worker_ready, worker_setup) = mpsc::channel();
        let worker_thread = thread::spawn(move || {
            let (command_tx, command_rx) = channel::unbounded();
            let signal = Arc::new(Signal::with_wait_observer(wait_entered));
            let worker_waker = Signal::waker(&signal);
            let mut worker = test_worker(
                command_tx.clone(),
                command_rx,
                test_blocking_worker(),
                InactiveClock::new_isolated(),
                signal,
            );
            let clock = worker.context.as_ref().unwrap().clock().clone();
            worker_ready.send((command_tx, worker_waker, clock)).unwrap();
            worker.run();
        });
        let (command_tx, worker_waker, clock) = worker_setup.recv_timeout(TEST_TIMEOUT).unwrap();
        wait_entry.recv_timeout(TEST_TIMEOUT).unwrap();
        let (timer_fired, timer_completion) = mpsc::channel();
        let timer_waker = Waker::from(std::sync::Arc::new(TimerWake(timer_fired)));
        let mut delay = Box::pin(clock.delay(Duration::from_millis(1)));

        assert!(delay.as_mut().poll(&mut std::task::Context::from_waker(&timer_waker)).is_pending());
        timer_completion.recv_timeout(TEST_TIMEOUT).unwrap();

        drop(delay);
        drop(clock);
        command_tx.send(Command::Shutdown).unwrap();
        worker_waker.wake_by_ref();
        worker_thread.join().unwrap();
    }

    #[cfg(not(miri))]
    #[test]
    fn timer_callback_direct_submission_notifies_worker() {
        struct SpawnOnWake {
            scheduler: crate::task::Scheduler,
            invoked: Arc<AtomicBool>,
        }

        impl Wake for SpawnOnWake {
            fn wake(self: std::sync::Arc<Self>) {
                let invoked = Arc::clone(&self.invoked);
                drop(self.scheduler.spawn(move |_| async move {
                    invoked.store(true, Ordering::Release);
                }));
            }
        }

        let control = ClockControl::new();
        let (command_tx, command_rx) = channel::unbounded();
        let signal = Arc::new(Signal::default());
        let mut worker = test_worker(
            command_tx,
            command_rx,
            test_blocking_worker(),
            control.clone().into(),
            Arc::clone(&signal),
        );
        let clock = worker.context.as_ref().unwrap().clock().clone();
        let invoked = Arc::new(AtomicBool::new(false));
        let callback_waker = Waker::from(std::sync::Arc::new(SpawnOnWake {
            scheduler: worker.context.as_ref().unwrap().scheduler().clone(),
            invoked: Arc::clone(&invoked),
        }));
        let mut delay = Box::pin(clock.delay(Duration::from_secs(1)));
        assert!(
            delay
                .as_mut()
                .poll(&mut std::task::Context::from_waker(&callback_waker))
                .is_pending()
        );
        while signal.take_notification() {}

        control.advance(Duration::from_secs(1));

        assert!(signal.is_notified());
        assert!(worker.execute_step().is_continue());
        assert!(invoked.load(Ordering::Acquire));
    }

    #[cfg(not(miri))]
    #[test]
    fn generated_command_sequences_preserve_progress_and_close_admission() {
        bolero::check!().for_each(|input: &[u8]| {
            let mut scenario = GeneratedWorkerScenario::new();
            for operation in input.iter().copied().take(GeneratedWorkerScenario::MAX_OPERATIONS) {
                scenario.apply(operation);
            }
            scenario.finish();
        });
    }

    #[test]
    fn a_never_empty_command_queue_allows_registered_tasks_and_timers_to_progress() {
        let (command_tx, command_rx) = channel::unbounded();
        let (proceed, progress) = Event::boxed();
        let control = ClockControl::new();
        let task_done = Arc::new(AtomicBool::new(false));
        let timer_done = Arc::new(AtomicBool::new(false));
        let processed = Arc::new(AtomicUsize::new(0));
        let task_observer = Arc::clone(&task_done);
        let timer_observer = Arc::clone(&timer_done);
        let shutdown = command_tx.clone();

        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            control.clone().into(),
            Arc::new(Signal::default()),
        );
        worker.tasks.add_test(async move {
            progress.await.unwrap();
            task_observer.store(true, Ordering::Relaxed);
        });
        let clock = worker.context.as_ref().unwrap().clock().clone();
        worker.tasks.add_test(async move {
            clock.delay(Duration::from_secs(1)).await;
            timer_observer.store(true, Ordering::Relaxed);
            shutdown.send(Command::Shutdown).unwrap();
        });
        _ = worker.executor.as_mut().unwrap().execute_cycle();
        assert!(!task_done.load(Ordering::Relaxed));
        assert!(!timer_done.load(Ordering::Relaxed));

        command_tx
            .send(replenishing_command(command_tx.clone(), Arc::clone(&processed)))
            .unwrap();
        proceed.send(());
        control.advance(Duration::from_secs(1));
        worker.run();

        assert!(task_done.load(Ordering::Relaxed));
        assert!(timer_done.load(Ordering::Relaxed));
        assert!((COMMANDS_PER_CYCLE..=2 * COMMANDS_PER_CYCLE + 1).contains(&processed.load(Ordering::Relaxed)));
    }

    #[test]
    fn a_full_command_batch_skips_waiting_for_the_worker_signal() {
        let (command_tx, command_rx) = channel::unbounded();
        let signal = Arc::new(Signal::default());

        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::clone(&signal),
        );
        Signal::waker(&signal).wake_by_ref();
        assert!(worker.execute_step().is_continue());

        Signal::waker(&signal).wake_by_ref();
        assert!(signal.is_notified());
        assert!(worker.execute_step().is_continue());
        assert!(!signal.is_notified());

        for _ in 0..COMMANDS_PER_CYCLE {
            command_tx.send(enqueue(RemoteFutureFactory::new(|_, _| {}))).unwrap();
        }
        Signal::waker(&signal).wake_by_ref();
        assert!(signal.is_notified());
        assert!(worker.execute_step().is_continue());
        assert!(signal.is_notified());

        command_tx.send(Command::Shutdown).unwrap();
        worker.run();
    }

    #[test]
    fn a_full_command_batch_gives_the_primary_driver_no_wait_budget_and_skips_signal_wait() {
        let (command_tx, command_rx) = channel::unbounded();
        let signal = Arc::new(Signal::default());
        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::clone(&signal),
        );
        let waits = Arc::new(std::sync::Mutex::new(Vec::new()));
        let _context = worker
            .context
            .as_ref()
            .unwrap()
            .io()
            .context::<WaitRecordingContext>(&Arc::clone(&waits));
        for _ in 0..COMMANDS_PER_CYCLE {
            command_tx.send(enqueue(RemoteFutureFactory::new(|_, _| {}))).unwrap();
        }
        Signal::waker(&signal).wake_by_ref();
        assert!(signal.is_notified());

        assert!(worker.execute_step().is_continue());
        assert_eq!(*waits.lock().unwrap(), vec![Duration::ZERO]);
        assert!(signal.is_notified());

        command_tx.send(Command::Shutdown).unwrap();
        worker.run();
    }

    #[cfg(not(miri))]
    #[test]
    fn final_timer_panic_occurs_after_executor_drop() {
        struct PanicWake;

        impl Wake for PanicWake {
            fn wake(self: std::sync::Arc<Self>) {
                panic!("final timer callback");
            }
        }

        let (command_tx, command_rx) = channel::unbounded();
        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );
        let clock = worker.context.as_ref().unwrap().clock().clone();
        let mut timer = clock.delay(Duration::from_millis(1));
        let waker = Waker::from(Arc::into_std_arc(Arc::new(PanicWake)));
        assert!(Pin::new(&mut timer).poll(&mut std::task::Context::from_waker(&waker)).is_pending());
        // Forgetting a timer is legal and avoids a second panic from the poisoned timer lock.
        std::mem::forget(timer);
        thread::sleep(Duration::from_millis(2));
        command_tx.send(Command::Shutdown).unwrap();

        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| worker.run()));
        assert_eq!(*outcome.unwrap_err().downcast::<&str>().unwrap(), "final timer callback");
    }

    #[test]
    fn smoke_test() {
        async_test(async || {
            // Run worker, execute one remote task which awaits a nested task, then shut down.
            let (command_tx, command_rx) = channel::unbounded();
            let worker_command_tx = command_tx.clone();
            let signal = Arc::new(Signal::default());
            let worker_signal = Arc::clone(&signal);

            let async_worker_thread = thread::spawn(move || {
                let blocking_worker = test_blocking_worker();

                let mut worker = test_worker(
                    worker_command_tx,
                    command_rx,
                    (Arc::clone(&blocking_worker.0), blocking_worker.1.clone()),
                    InactiveClock::new_isolated(),
                    worker_signal,
                );
                worker.run();

                assert!(blocking_worker.0.is_shutting_down());
            });

            // Signals that the task waiting on the nested task has completed.
            let (outer_completed_tx, outer_completed_rx) = Event::boxed();

            // Signals that the nested task has completed.
            let (inner_completed_tx, inner_completed_rx) = Event::boxed();

            command_tx
                .send(enqueue(RemoteFutureFactory::new({
                    move |_: Context, tasks: &TaskSet| {
                        tasks.add_test(async move {
                            inner_completed_tx.send(());
                        });
                        tasks.add_test(async move {
                            inner_completed_rx.await.unwrap();
                            outer_completed_tx.send(());
                        });
                    }
                })))
                .unwrap();
            Signal::waker(&signal).wake_by_ref();

            // Wait for the command to complete.
            outer_completed_rx.await.unwrap();

            // Shut down after the test body completes.
            command_tx.send(Command::Shutdown).unwrap();
            Signal::waker(&signal).wake_by_ref();

            // Join the worker so this test reports shutdown errors.
            async_worker_thread.join().unwrap();
        });
    }

    #[test]
    fn worker_setup_can_register_new_tasks() {
        async_test(async || {
            // Observe both tasks.
            let (initial_completed_tx, initial_completed_rx) = Event::boxed();
            let (spawned_completed_tx, spawned_completed_rx) = Event::boxed();

            let (command_tx, command_rx) = channel::unbounded();
            let worker_command_tx = command_tx.clone();
            let signal = Arc::new(Signal::default());
            let worker_signal = Arc::clone(&signal);
            let worker_thread = thread::spawn(move || {
                let mut worker = test_worker(
                    worker_command_tx,
                    command_rx,
                    test_blocking_worker(),
                    InactiveClock::new_isolated(),
                    worker_signal,
                );
                worker.tasks.add_test(async move { initial_completed_tx.send(()) });

                worker.run();
            });

            command_tx
                .send(enqueue(RemoteFutureFactory::new({
                    move |_: Context, tasks: &TaskSet| {
                        tasks.add_test(async move {
                            spawned_completed_tx.send(());
                        });
                    }
                })))
                .unwrap();
            Signal::waker(&signal).wake_by_ref();

            spawned_completed_rx.await.unwrap();
            initial_completed_rx.into_value().unwrap();

            // Shut down after both tasks complete.
            command_tx.send(Command::Shutdown).unwrap();
            Signal::waker(&signal).wake_by_ref();

            // Join the worker so this test reports shutdown errors.
            worker_thread.join().unwrap();
        });
    }

    #[test]
    fn task_after_shutdown_is_ignored() {
        // The asynchronous submission path can enqueue a new-task command after
        // shutdown starts.
        //
        // The worker drops these tasks so their remote join handles report
        // shutdown and release all task resources.

        let (command_tx, command_rx) = channel::unbounded();
        let worker_command_tx = command_tx.clone();
        let signal = Arc::new(Signal::default());
        let worker_signal = Arc::clone(&signal);

        let async_worker_thread = thread::spawn(move || {
            let mut worker = test_worker(
                worker_command_tx,
                command_rx,
                test_blocking_worker(),
                InactiveClock::new_isolated(),
                worker_signal,
            );
            worker.run();
        });

        command_tx.send(Command::Shutdown).unwrap();
        Signal::waker(&signal).wake_by_ref();

        let (completed_tx, completed_rx) = Event::boxed();

        // The channel can already be closed because the worker is shutting down.
        let _ = command_tx.send(enqueue(RemoteFutureFactory::new({
            move |_: Context, tasks: &TaskSet| {
                tasks.add_test(async move {
                    completed_tx.send(14);
                });
            }
        })));

        // This waits for the worker to shut down.
        async_worker_thread.join().unwrap();

        let completed_result = completed_rx.into_value();
        assert!(matches!(completed_result, Err(IntoValueError::Disconnected)));
    }

    #[test]
    fn shutdown_command_cancels_registered_tasks_before_polling() {
        let (command_tx, command_rx) = channel::unbounded();
        let invoked = Arc::new(AtomicBool::new(false));
        let invoked_by_task = Arc::clone(&invoked);

        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );

        command_tx
            .send(enqueue(RemoteFutureFactory::new(move |_, tasks| {
                tasks.add_test(async move {
                    invoked_by_task.store(true, Ordering::Relaxed);
                });
            })))
            .unwrap();
        command_tx.send(Command::Shutdown).unwrap();

        worker.run();

        assert!(!invoked.load(Ordering::Relaxed));
    }

    #[test]
    fn same_worker_stop_skips_a_later_initialized_task() {
        let (command_tx, command_rx) = channel::unbounded();
        let mut worker = test_worker(
            command_tx,
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );
        let control = worker.context.as_ref().unwrap().control().clone();
        let invoked = Arc::new(AtomicBool::new(false));
        let invoked_by_task = Arc::clone(&invoked);
        worker.tasks.add_test(async move {
            control.request_stop();
        });
        worker.tasks.add_test(async move {
            invoked_by_task.store(true, Ordering::Release);
        });
        let executor = worker.executor.as_mut().unwrap();

        assert!(matches!(executor.execute_cycle(), CycleOutcome::Shutdown(_)));
        assert!(!invoked.load(Ordering::Acquire));
    }

    #[test]
    fn repeated_shutdown_cleanup_is_idempotent() {
        let (command_tx, command_rx) = channel::unbounded();
        let mut worker = test_worker(
            command_tx,
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );

        worker.begin_shutdown();
        worker.begin_shutdown();
    }

    #[test]
    fn published_shutdown_discards_queued_io_access_without_invoking_it() {
        struct DropSignal(mpsc::Sender<()>);

        impl Drop for DropSignal {
            fn drop(&mut self) {
                _ = self.0.send(());
            }
        }

        let (command_tx, command_rx) = channel::unbounded();
        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );
        let invoked = Arc::new(AtomicBool::new(false));
        let (dropped_tx, dropped_rx) = mpsc::channel();
        let drop_signal = DropSignal(dropped_tx);
        command_tx
            .send(Command::AccessIo {
                access: QueuedIoAccess::new(Box::new({
                    let invoked = Arc::clone(&invoked);
                    move |_| {
                        let _ = &drop_signal;
                        invoked.store(true, Ordering::Release);
                    }
                })),
            })
            .unwrap();
        worker.shutdown.begin(ShutdownCause::default());

        worker.run();

        dropped_rx.recv_timeout(TEST_TIMEOUT).unwrap();
        assert!(!invoked.load(Ordering::Acquire));
    }

    #[test]
    fn shutdown_published_during_final_batch_registration_cancels_before_polling() {
        let (command_tx, command_rx) = channel::unbounded();
        let invoked = Arc::new(AtomicUsize::new(0));
        let mut worker = test_worker(
            command_tx.clone(),
            command_rx,
            test_blocking_worker(),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );
        let shutdown = worker.shutdown.clone();

        for index in 0..COMMANDS_PER_CYCLE {
            let invoked = Arc::clone(&invoked);
            let shutdown = shutdown.clone();
            command_tx
                .send(enqueue(RemoteFutureFactory::new(move |_, tasks| {
                    if index + 1 == COMMANDS_PER_CYCLE {
                        shutdown.begin(ShutdownCause::default());
                    }
                    tasks.add_test(async move {
                        invoked.fetch_add(1, Ordering::Relaxed);
                    });
                })))
                .unwrap();
        }

        worker.run();

        assert_eq!(invoked.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn dropping_worker_without_run_cleans_up_safely() {
        let (commands, receiver) = channel::unbounded();
        let blocking_worker = test_blocking_worker();
        let worker = test_worker(
            commands,
            receiver,
            (Arc::clone(&blocking_worker.0), blocking_worker.1.clone()),
            InactiveClock::new_isolated(),
            Arc::new(Signal::default()),
        );

        drop(worker);

        assert!(blocking_worker.0.is_shutting_down());
    }

    #[test]
    fn explicit_executor_shutdown_completes_worker_shutdown() {
        let (commands, receiver) = channel::unbounded();
        let blocking_worker = test_blocking_worker();
        let signal = Arc::new(Signal::default());
        let mut worker = test_worker(
            commands.clone(),
            receiver,
            (Arc::clone(&blocking_worker.0), blocking_worker.1.clone()),
            InactiveClock::new_isolated(),
            Arc::clone(&signal),
        );
        Signal::waker(&signal).wake_by_ref();
        assert!(worker.execute_step().is_continue());
        worker.executor.as_ref().unwrap().shutdown().trigger(ShutdownCause::default());

        let outcome = worker.execute_step();

        assert!(outcome.is_break());
        assert!(worker.executor.is_none());
        assert!(blocking_worker.0.is_shutting_down());
        assert!(!worker.process_commands());
        drop(commands);
    }
}