beamr 0.16.2

//! Cooperative single-threaded scheduler for `wasm32-unknown-unknown` hosts.
//!
//! The host owns the event loop and calls [`WasmScheduler::run_until_idle`]
//! from `requestAnimationFrame`, a microtask, or an equivalent callback. No OS
//! threads, blocking I/O, dirty pools, or distribution services are started here.

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::rc::Rc;
use std::sync::{Arc, Mutex};

use crate::atom::{Atom, AtomTable};
use crate::error::ExecError;
use crate::ets::OwnedTerm;
use crate::interpreter::{ExecutionResult, NativeServices, run_with_native_services};
use crate::io_sink::IoSink;
use crate::mailbox::SendError;
use crate::module::ModuleRegistry;
use crate::namespace::NamespaceId;
use crate::native::{BifRegistryImpl, CapabilitySet, WasmAsyncNifFacility};
use crate::process::heap::DEFAULT_HEAP_SIZE;
use crate::process::{CodePosition, ExitReason, Priority, Process, ProcessStatus};
use crate::term::Term;
use crate::timer::TimerWheel;

/// Receive timer scheduled by the WASM scheduler and awaiting a host timeout.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct WasmScheduledTimer {
    /// Process that should be resumed when the timer expires.
    pub pid: u64,
    /// Opaque timer id stored on the process and mirrored by the host.
    pub timer_id: u64,
    /// Delay requested by the `receive after` instruction.
    pub milliseconds: u64,
}

/// Outcome returned when host async work completes.
#[derive(Debug)]
pub enum WasmAsyncCompletion {
    /// Promise fulfilled; inject the value into x(0) and advance past the NIF.
    Ok(OwnedTerm),
    /// Promise rejected; terminate the process with a NIF error mapping.
    Error(OwnedTerm),
}

/// Maximum number of reduction-bounded process slices executed in one host turn.
///
/// A finite wall guarantees that continuously runnable work returns control to
/// the host event loop so timers and I/O can be serviced. This is deliberately
/// not configurable: embedders observe the fairness result instead.
pub const MAX_SLICES_PER_DRAIN: usize = 1_024;

/// Post-drain scheduler state, separate from process exits and errors.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WasmRunState {
    /// No runnable process remains. This scheduler exposes but does not arm the
    /// later host-deadline service's native timer-wheel seam.
    Idle {
        /// Earliest absolute native timer-wheel deadline, if one is pending.
        next_native_deadline: Option<web_time::Instant>,
    },
    /// Runnable processes remain because this host turn reached its fairness wall.
    FairnessYield {
        /// Exact number of processes still present in the ready queues.
        runnable_remaining: usize,
    },
}

impl WasmRunState {
    /// Convert the exposed native deadline to non-negative milliseconds from now.
    #[must_use]
    pub fn next_native_deadline_millis_from_now(&self) -> Option<f64> {
        match self {
            Self::Idle {
                next_native_deadline: Some(deadline),
            } => Some(
                deadline
                    .saturating_duration_since(web_time::Instant::now())
                    .as_secs_f64()
                    * 1_000.0,
            ),
            Self::Idle { .. } | Self::FairnessYield { .. } => None,
        }
    }
}

/// Summary returned from one cooperative scheduler drain.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WasmRunSummary {
    /// Whether the drain reached true idle or yielded for host fairness.
    pub state: WasmRunState,
    /// Number of processes that received one reduction-bounded slice.
    pub executed: usize,
    /// PIDs that yielded and were requeued for a later host tick.
    pub yielded: Vec<u64>,
    /// PIDs that blocked waiting for a message or explicit wake.
    pub waiting: Vec<u64>,
    /// PIDs that exited during this turn.
    pub exited: Vec<u64>,
    /// PIDs that faulted with an interpreter error during this turn.
    pub errored: Vec<u64>,
}

impl Default for WasmRunSummary {
    fn default() -> Self {
        Self {
            state: WasmRunState::Idle {
                next_native_deadline: None,
            },
            executed: 0,
            yielded: Vec::new(),
            waiting: Vec::new(),
            exited: Vec::new(),
            errored: Vec::new(),
        }
    }
}

/// Single-threaded cooperative scheduler for WASM.
pub struct WasmScheduler {
    atom_table: Arc<AtomTable>,
    module_registry: Arc<ModuleRegistry>,
    bif_registry: Arc<BifRegistryImpl>,
    pub(super) processes: BTreeMap<u64, Process>,
    pub(super) ready: ReadyQueues,
    pub(super) waiting: BTreeSet<u64>,
    exit_reasons: BTreeMap<u64, ExitReason>,
    exit_results: BTreeMap<u64, OwnedTerm>,
    exit_errors: BTreeMap<u64, ExecError>,
    next_timer_id: u64,
    pending_timer_schedules: Vec<WasmScheduledTimer>,
    pending_timer_cancellations: Vec<u64>,
    pub(super) async_results: BTreeMap<u64, WasmAsyncCompletion>,
    pub(super) wasm_async_nif_facility: Option<Rc<dyn WasmAsyncNifFacility>>,
    /// Shared pid counter used by the cooperative native path (WR-0). It is an
    /// `Arc<Mutex<…>>` so the `Send + Sync` cooperative `SpawnFacility` can
    /// allocate child pids from inside a slice; uncontended (single thread).
    pub(super) shared_next_pid: Arc<Mutex<u64>>,
    /// Exit reasons captured by the cooperative native path.
    pub(super) native_exit_reasons: BTreeMap<u64, ExitReason>,
    /// Shared timer wheel for cooperative native `Deliver` timers (WR-4).
    ///
    /// `NativeContext::send_after`/`schedule` and the bytecode timer BIFs build
    /// `TimerKind::Deliver` entries carrying a message payload. The wheel is
    /// ticked at the start of every cooperative turn (see
    /// [`WasmScheduler::tick_native_timers`]) and its earliest deadline is
    /// reported by [`WasmRunState::Idle`], where the wrapper's unified deadline
    /// service arms the single host one-shot (WPORT-3). It is an
    /// `Arc<Mutex<…>>` so the `Send + Sync` [`NativeContext`] can hold it;
    /// uncontended on the single host thread.
    pub(super) native_timers: Arc<Mutex<TimerWheel>>,
    /// One-shot latch set by an external empty-ready to runnable mutation.
    external_runnable_edge_pending: bool,
    /// Suppresses external-edge requests for work produced inside an active drain.
    drain_in_progress: bool,
    /// Wrapper-injected output sink for the `io`-family BIFs (WPORT-5 R2
    /// item 4). `None` until the embedder installs one; injected into
    /// bytecode `NativeServices` per slice. The sink contract is PUSH-ONLY
    /// (NO-POLLING law): writes reach it synchronously at the writing BIF.
    io_sink: Option<Arc<dyn IoSink>>,
}

impl WasmScheduler {
    /// Create a scheduler around the VM-global registries used by module loading
    /// and native import resolution.
    #[must_use]
    pub fn new(
        atom_table: Arc<AtomTable>,
        module_registry: Arc<ModuleRegistry>,
        bif_registry: Arc<BifRegistryImpl>,
    ) -> Self {
        Self {
            atom_table,
            module_registry,
            bif_registry,
            processes: BTreeMap::new(),
            ready: ReadyQueues::default(),
            waiting: BTreeSet::new(),
            exit_reasons: BTreeMap::new(),
            exit_results: BTreeMap::new(),
            exit_errors: BTreeMap::new(),
            next_timer_id: 1,
            pending_timer_schedules: Vec::new(),
            pending_timer_cancellations: Vec::new(),
            async_results: BTreeMap::new(),
            wasm_async_nif_facility: None,
            shared_next_pid: Arc::new(Mutex::new(1)),
            native_exit_reasons: BTreeMap::new(),
            native_timers: Arc::new(Mutex::new(TimerWheel::new())),
            external_runnable_edge_pending: false,
            drain_in_progress: false,
            io_sink: None,
        }
    }

    /// Access the atom table backing this scheduler.
    #[must_use]
    pub fn atom_table(&self) -> &Arc<AtomTable> {
        &self.atom_table
    }

    /// Access the module registry backing this scheduler.
    #[must_use]
    pub fn module_registry(&self) -> &Arc<ModuleRegistry> {
        &self.module_registry
    }

    /// Access the BIF registry backing this scheduler.
    #[must_use]
    pub fn bif_registry(&self) -> &Arc<BifRegistryImpl> {
        &self.bif_registry
    }

    /// Install the single-threaded host bridge used by WASM async NIF stubs.
    pub fn set_wasm_async_nif_facility(&mut self, facility: Option<Rc<dyn WasmAsyncNifFacility>>) {
        self.wasm_async_nif_facility = facility;
    }

    /// Install the output sink injected into bytecode `NativeServices`
    /// (WPORT-5 R2 item 4). The wrapper installs its host sink at
    /// construction so `io`-family output goes somewhere with zero
    /// configuration; the sink is PUSH-ONLY (NO-POLLING law) — writes reach
    /// it synchronously at the writing BIF, with no flush timer or recurring
    /// callback on any branch.
    pub fn set_io_sink(&mut self, sink: Arc<dyn IoSink>) {
        self.io_sink = Some(sink);
    }

    /// Drain receive timers that the host must schedule with `setTimeout`.
    pub fn take_pending_timer_schedules(&mut self) -> Vec<WasmScheduledTimer> {
        std::mem::take(&mut self.pending_timer_schedules)
    }

    /// Drain receive timers that the host must cancel with `clearTimeout`.
    pub fn take_pending_timer_cancellations(&mut self) -> Vec<u64> {
        std::mem::take(&mut self.pending_timer_cancellations)
    }

    /// Expire a host timer callback if it still matches the waiting process.
    pub fn timer_fired(&mut self, pid: u64, timer_id: u64) -> bool {
        let Some(process) = self.processes.get_mut(&pid) else {
            return false;
        };
        if process.receive_timer_ref() != Some(timer_id) {
            return false;
        }
        process.set_receive_timer_ref(None);
        if let Some(position) = process
            .receive_timeout()
            .map(|timeout| timeout.timeout_position)
        {
            process.set_code_position(Some(position));
        }
        self.wake(pid)
    }

    /// Record an async NIF completion and wake the suspended process.
    ///
    /// The bytecode async-NIF path stores the completion in `async_results`,
    /// where the next bytecode slice injects it into `x(0)` and advances past the
    /// NIF (see `WasmScheduler::apply_async_completion`). A NATIVE process has
    /// no `x(0)`/instruction pointer, so for a native target the completion is
    /// delivered as a mailbox message instead — `{ok, Value}` on success or
    /// `{error, Reason}` on rejection — which the parked handler reads via
    /// [`NativeContext::recv`](crate::native::native_process::NativeContext::recv)
    /// when it resumes (WR-7). Both paths share the same pid-keyed `wake`.
    pub fn complete_async(&mut self, pid: u64, completion: WasmAsyncCompletion) -> bool {
        let Some(process) = self.processes.get(&pid) else {
            return false;
        };
        if process.is_native() {
            return self.deliver_native_async_completion(pid, completion);
        }
        self.async_results.insert(pid, completion);
        self.wake(pid)
    }

    /// Spawn a process at an exported module/function/arity entrypoint.
    pub fn spawn(
        &mut self,
        entry_module: Atom,
        entry_function: Atom,
        args: Vec<Term>,
    ) -> Result<u64, ExecError> {
        self.spawn_in(NamespaceId::DEFAULT, entry_module, entry_function, args)
    }

    /// Spawn a process with arguments held in owned detached storage.
    pub fn spawn_owned(
        &mut self,
        entry_module: Atom,
        entry_function: Atom,
        args: Vec<OwnedTerm>,
    ) -> Result<u64, ExecError> {
        self.spawn_in_owned(NamespaceId::DEFAULT, entry_module, entry_function, args)
    }

    /// Spawn a process in a namespace. WASM is single-node and currently only
    /// supports the default namespace.
    pub fn spawn_in(
        &mut self,
        namespace: NamespaceId,
        entry_module: Atom,
        entry_function: Atom,
        args: Vec<Term>,
    ) -> Result<u64, ExecError> {
        if namespace != NamespaceId::DEFAULT {
            return Err(ExecError::Badarg);
        }
        let arity = u8::try_from(args.len()).map_err(|_| ExecError::Badarg)?;
        let entry = self
            .module_registry
            .lookup_mfa(entry_module, entry_function, arity)?;
        let instruction_pointer = entry.module.label_ip(entry.label)?;

        // One pid space for every spawn path (WPORT-5 R2 item 2): the shared
        // counter also feeds native root spawns and the cooperative
        // spawn/MFA-spawn facilities, so bytecode and native pids can never
        // collide (the former private `next_pid` counter overlapped it).
        let pid = self.alloc_pid();

        let mut process = Process::with_capabilities(pid, DEFAULT_HEAP_SIZE, CapabilitySet::all());
        process.set_group_leader(Term::pid(pid));
        process.set_priority(Priority::Normal);
        process.set_namespace_id(namespace);
        process.set_code_position(Some(CodePosition {
            module: entry_module,
            instruction_pointer,
        }));
        process.set_current_module(entry.module);
        for (index, arg) in args.into_iter().enumerate().take(1024) {
            if let Ok(register) = u16::try_from(index) {
                process.set_x_reg(register, arg);
            }
        }
        let ready_was_empty = self.ready.len() == 0;
        self.ready.push(pid, process.priority());
        self.note_ready_push(ready_was_empty);
        self.processes.insert(pid, process);
        Ok(pid)
    }

    /// Spawn a process in a namespace with arguments copied from owned storage
    /// into the new process heap.
    pub fn spawn_in_owned(
        &mut self,
        namespace: NamespaceId,
        entry_module: Atom,
        entry_function: Atom,
        args: Vec<OwnedTerm>,
    ) -> Result<u64, ExecError> {
        if namespace != NamespaceId::DEFAULT {
            return Err(ExecError::Badarg);
        }
        let arity = u8::try_from(args.len()).map_err(|_| ExecError::Badarg)?;
        let entry = self
            .module_registry
            .lookup_mfa(entry_module, entry_function, arity)?;
        let instruction_pointer = entry.module.label_ip(entry.label)?;

        // Shared pid counter — see `spawn_in` (WPORT-5 R2 item 2).
        let pid = self.alloc_pid();

        let mut process = Process::with_capabilities(pid, DEFAULT_HEAP_SIZE, CapabilitySet::all());
        process.set_group_leader(Term::pid(pid));
        process.set_priority(Priority::Normal);
        process.set_namespace_id(namespace);
        process.set_code_position(Some(CodePosition {
            module: entry_module,
            instruction_pointer,
        }));
        process.set_current_module(entry.module);
        for (index, arg) in args.iter().enumerate().take(1024) {
            if let Ok(register) = u16::try_from(index) {
                let copied = arg
                    .copy_to_heap(process.heap_mut())
                    .map_err(|_| ExecError::Badarg)?;
                process.set_x_reg(register, copied);
            }
        }
        let ready_was_empty = self.ready.len() == 0;
        self.ready.push(pid, process.priority());
        self.note_ready_push(ready_was_empty);
        self.processes.insert(pid, process);
        Ok(pid)
    }

    /// Wake a previously waiting process so it can be run by a later host tick.
    pub fn wake(&mut self, pid: u64) -> bool {
        if !self.waiting.remove(&pid) {
            return false;
        }
        let Some(process) = self.processes.get_mut(&pid) else {
            return false;
        };
        if process.transition_to(ProcessStatus::Running).is_err() {
            return false;
        }
        let ready_was_empty = self.ready.len() == 0;
        self.ready.push(pid, process.priority());
        self.note_ready_push(ready_was_empty);
        true
    }

    /// Deliver an immediate message to a local process and wake it if it was blocked.
    pub fn send(&mut self, pid: u64, message: Term) -> bool {
        self.enqueue_owned_message(pid, message)
    }

    /// Deliver an owned host term to a local process and wake it if it was blocked.
    ///
    /// Boxed values are first copied into the receiver's heap, preserving the
    /// normal mailbox ownership contract for messages entering from JavaScript.
    pub fn send_owned(&mut self, pid: u64, message: &OwnedTerm) -> Result<(), ExecError> {
        let Some(process) = self.processes.get_mut(&pid) else {
            return Err(ExecError::Badarg);
        };
        let sender = process.mailbox().sender();
        sender
            .send(message.root(), process.heap_mut())
            .map_err(send_error_to_exec)?;
        self.after_successful_enqueue(pid);
        Ok(())
    }

    fn enqueue_owned_message(&mut self, pid: u64, message: Term) -> bool {
        let Some(_process) = self.processes.get(&pid) else {
            return false;
        };
        self.enqueue_copied_message(pid, message);
        true
    }

    fn enqueue_copied_message(&mut self, pid: u64, message: Term) {
        if let Some(process) = self.processes.get_mut(&pid) {
            process.mailbox_mut().push_owned(message);
        }
        self.after_successful_enqueue(pid);
    }

    pub(super) fn after_successful_enqueue(&mut self, pid: u64) {
        if let Some(process) = self.processes.get_mut(&pid)
            && let Some(timer_id) = process.receive_timer_ref()
        {
            process.set_receive_timer_ref(None);
            self.pending_timer_cancellations.push(timer_id);
        }
        if self.waiting.contains(&pid) {
            let _woken = self.wake(pid);
        }
    }

    /// Drain native `Deliver` timers that expired by wall-clock now, delivering
    /// each timer's message to its target mailbox and waking the target.
    ///
    /// Returns the pids woken by an expiring timer. This is the host-facing
    /// entry point; [`WasmScheduler::run_until_idle`] calls it at the start of
    /// every turn so a parked native actor whose self-tick has come due is
    /// runnable in the same turn. The deterministic counterpart used by tests
    /// (and the seam a browser host drives with a `performance.now()`-derived
    /// instant) is [`WasmScheduler::tick_native_timers_at`].
    ///
    /// When no native timers are pending this returns immediately WITHOUT
    /// reading the wall clock, an early-out that predates the `web_time` clock
    /// base. The clock the wheel reads when timers ARE pending is
    /// `web_time::Instant::now()` (WR-10), which is `performance.now()`-backed on
    /// `wasm32-unknown-unknown` and a re-export of `std::time::Instant::now()` on
    /// native — so this no longer panics in the browser when a timer is armed.
    pub fn tick_native_timers(&mut self) -> Vec<u64> {
        let expired = {
            let mut wheel = lock_timers(&self.native_timers);
            if wheel.is_empty() {
                return Vec::new();
            }
            wheel.tick()
        };
        self.deliver_expired_native_timers(expired)
    }

    /// Deterministic variant of [`WasmScheduler::tick_native_timers`]: expire
    /// every native `Deliver` timer due at `now` and deliver it.
    ///
    /// Tests drive the cooperative timer path through this seam — schedule a
    /// self-tick from a native handler, advance `now` past the delay, and the
    /// rescheduled actor receives the delivered message — with no browser and no
    /// wall-clock dependency, exactly as the threaded `expire_timers_for_test`
    /// seam does for the threaded scheduler.
    pub fn tick_native_timers_at(&mut self, now: web_time::Instant) -> Vec<u64> {
        let expired = lock_timers(&self.native_timers).tick_at(now);
        self.deliver_expired_native_timers(expired)
    }

    /// Route a drained set of expired timers: `Deliver` timers push their
    /// message into the target mailbox and wake it; `ReceiveTimeout` timers are
    /// never produced by the native path (it only schedules `Deliver`) and are
    /// ignored defensively. Returns the woken pids.
    fn deliver_expired_native_timers(
        &mut self,
        expired: Vec<crate::timer::ExpiredTimer>,
    ) -> Vec<u64> {
        let mut woken = Vec::new();
        for timer in expired {
            if timer.kind != crate::timer::TimerKind::Deliver {
                continue;
            }
            let pid = timer.target_pid;
            let Some(process) = self.processes.get_mut(&pid) else {
                continue;
            };
            process.mailbox_mut().push_owned(timer.message);
            self.after_successful_enqueue(pid);
            woken.push(pid);
        }
        woken
    }

    /// Drain runnable work until true idle or the fixed host-turn fairness wall.
    pub fn run_until_idle(&mut self) -> WasmRunSummary {
        self.run_until_idle_with_limit(MAX_SLICES_PER_DRAIN)
    }

    #[cfg(test)]
    pub(super) fn run_until_idle_with_test_limit(&mut self, limit: usize) -> WasmRunSummary {
        self.run_until_idle_with_limit(limit)
    }

    fn run_until_idle_with_limit(&mut self, limit: usize) -> WasmRunSummary {
        self.drain_in_progress = true;
        self.external_runnable_edge_pending = false;
        // WR-4: expire any native `Deliver` timers due now so a parked actor
        // whose self-tick has come due is runnable within this same turn.
        let _woken = self.tick_native_timers();
        let mut summary = WasmRunSummary::default();
        let mut yielded_next_tick = Vec::new();

        while summary.executed < limit {
            let Some(pid) = self.ready.pop() else {
                break;
            };
            if self.waiting.contains(&pid) {
                continue;
            }
            let Some(mut process) = self.processes.remove(&pid) else {
                continue;
            };
            let priority = process.priority();
            if !matches!(process.status(), ProcessStatus::Running) {
                let _transition = process.transition_to(ProcessStatus::Running);
            }

            // WR-3: native processes are driven by the cooperative native slice
            // in the same turn as bytecode, branching on `is_native()` the way
            // the threaded `core::execute_slice` does. Native bodies carry no
            // bytecode module, so they must be dispatched before the
            // async-completion / current-module bytecode path below.
            if process.is_native() {
                self.dispatch_native_in_turn(
                    pid,
                    priority,
                    process,
                    &mut summary,
                    &mut yielded_next_tick,
                );
                continue;
            }

            // LATENT GAP, record-only (WPORT-4 tear Ruling 7): bytecode-process
            // exits — this arm and the `ExecutionResult::Exited` arm below —
            // perform NO link propagation, unlike the cooperative native path
            // (`wasm_native.rs::propagate_native_exit`). Still unreachable
            // after WPORT-5 R2: the injected cooperative spawn facility
            // refuses every link/monitor-bearing variant (plain `spawn/3`
            // only — OQ8 ruled) and no link/supervision facility is injected,
            // so a bytecode process cannot hold links. The guarding wall
            // belongs to the future bytecode-linking brief that makes this
            // path reachable (arc-board line recorded).
            if let Some(reason) = self.apply_async_completion(&mut process) {
                let x0 = process.x_reg(0);
                let _transition = process.transition_to(ProcessStatus::Exited(reason));
                self.exit_reasons.insert(pid, reason);
                self.exit_results
                    .insert(pid, super::exit_capture::capture_term(x0));
                summary.exited.push(pid);
                continue;
            }
            process.reset_reductions(crate::scheduler::DEFAULT_REDUCTION_BUDGET);

            let Some(module) = process.current_module().cloned() else {
                self.exit_errors
                    .insert(pid, ExecError::InvalidOperand("current module"));
                summary.errored.push(pid);
                continue;
            };

            // WPORT-5 R2 items 1+2: mirror the native-slice shape at the
            // bytecode seam — per-slice effect facilities over a shared
            // `DeferredEffects` buffer, applied after the slice returns and
            // the process is back in the map (or dropped on exit). This is
            // what makes cross-process `Pid ! Msg` deliver and plain
            // `erlang:spawn/3` run instead of silently dropping/refusing.
            let (effects, local_send, spawn_facility) = self.bytecode_effect_facilities();
            let services = NativeServices {
                local_send: Some(local_send),
                spawn_facility: Some(spawn_facility),
                ..self.native_services()
            };
            let result = run_with_native_services(
                &mut process,
                module.as_ref(),
                self.module_registry.as_ref(),
                &services,
            );
            summary.executed += 1;

            match result {
                Ok(ExecutionResult::Yielded) => {
                    let _transition = process.transition_to(ProcessStatus::Yielded);
                    self.processes.insert(pid, process);
                    yielded_next_tick.push((pid, priority));
                    summary.yielded.push(pid);
                }
                Ok(ExecutionResult::Waiting) => {
                    let _transition = process.transition_to(ProcessStatus::Waiting);
                    self.register_receive_timer(&mut process);
                    self.processes.insert(pid, process);
                    self.waiting.insert(pid);
                    summary.waiting.push(pid);
                }
                Ok(ExecutionResult::Exited(reason)) => {
                    let x0 = process.x_reg(0);
                    let _transition = process.transition_to(ProcessStatus::Exited(reason));
                    self.exit_reasons.insert(pid, reason);
                    // Deep-copy while the process heap is still alive; the
                    // process is dropped at the end of this scope.
                    self.exit_results
                        .insert(pid, super::exit_capture::capture_term(x0));
                    summary.exited.push(pid);
                }
                Ok(ExecutionResult::DirtyCall { .. }) => {
                    self.exit_errors.insert(
                        pid,
                        ExecError::UnsupportedOpcode {
                            name: "dirty native call on wasm",
                        },
                    );
                    summary.errored.push(pid);
                }
                Err(error) => {
                    self.exit_errors.insert(pid, error);
                    summary.errored.push(pid);
                }
            }

            // Apply the slice's deferred sends/spawns now that its borrow of
            // the process has settled (WPORT-5 R2): delivered messages wake
            // their targets and spawned children become runnable within this
            // same drain's budget discipline.
            self.apply_bytecode_effects(&effects);
        }

        for (pid, priority) in yielded_next_tick {
            self.ready.push(pid, priority);
        }
        self.drain_in_progress = false;
        let runnable_remaining = self.ready.len();
        summary.state = if runnable_remaining == 0 {
            WasmRunState::Idle {
                next_native_deadline: lock_timers(&self.native_timers).earliest_deadline(),
            }
        } else {
            WasmRunState::FairnessYield { runnable_remaining }
        };
        summary
    }

    /// Consume the one-shot external idle-to-runnable edge latch.
    pub fn take_external_runnable_edge(&mut self) -> bool {
        std::mem::take(&mut self.external_runnable_edge_pending)
    }

    /// Return a process exit result captured from x(0), if available.
    ///
    /// The result is an owning deep copy that outlives the exited process.
    #[must_use]
    pub fn take_exit_result(&mut self, pid: u64) -> Option<OwnedTerm> {
        self.exit_results.remove(&pid)
    }

    /// Return all currently recorded exit results without consuming them.
    ///
    /// The returned terms borrow storage owned by this scheduler; they stay
    /// valid until the corresponding entry is removed via `take_exit_result`.
    #[must_use]
    pub fn exit_results(&self) -> Vec<(u64, Term)> {
        self.exit_results
            .iter()
            .map(|(pid, owned)| (*pid, owned.root()))
            .collect()
    }

    /// Allocate the next pid from the shared counter used by both the native
    /// root spawn and the cooperative spawn facility.
    pub(super) fn alloc_pid(&self) -> u64 {
        let mut guard = self
            .shared_next_pid
            .lock()
            .unwrap_or_else(|error| error.into_inner());
        let pid = *guard;
        *guard = guard.saturating_add(1);
        pid
    }

    pub(super) fn note_ready_push(&mut self, ready_was_empty: bool) {
        if ready_was_empty && !self.drain_in_progress {
            self.external_runnable_edge_pending = true;
        }
    }

    /// Return whether an interpreter error has been retained for `pid`.
    #[must_use]
    pub fn has_exit_error(&self, pid: u64) -> bool {
        self.exit_errors.contains_key(&pid)
    }

    /// Consume and return the interpreter error retained for `pid`, if any
    /// (WPORT-5 R2 item 7; the cooperative counterpart of the threaded
    /// `Scheduler::take_exit_error`). This is what makes refusal REASONS —
    /// facility-absent `Badarg` vs `Undef`-with-MFA vs `UnsupportedOpcode` —
    /// observable at the embedder boundary instead of a bare boolean.
    #[must_use]
    pub fn take_exit_error(&mut self, pid: u64) -> Option<ExecError> {
        self.exit_errors.remove(&pid)
    }

    /// Return the retained interpreter error for `pid` WITHOUT consuming it.
    ///
    /// Completion surfaces use this peek so repeated completion queries for
    /// an errored pid keep answering (exactly as `has_exit_error` always
    /// did), while [`WasmScheduler::take_exit_error`] remains the consuming
    /// accessor.
    #[must_use]
    pub fn exit_error(&self, pid: u64) -> Option<&ExecError> {
        self.exit_errors.get(&pid)
    }

    /// Record an interpreter error for a process that could not be
    /// materialized or run (used by the deferred MFA-spawn path).
    pub(super) fn record_exit_error(&mut self, pid: u64, error: ExecError) {
        self.exit_errors.insert(pid, error);
    }

    /// Return the number of currently runnable processes.
    #[must_use]
    pub fn runnable_count(&self) -> usize {
        self.ready.len()
    }

    /// Total number of ready-queued processes across all priorities.
    pub(super) fn ready_len(&self) -> usize {
        self.ready.len()
    }

    /// Whether a process is ready to run or a native `Deliver` timer is armed
    /// (a later tick will deliver its message and wake the target).
    ///
    /// An introspection predicate for tests and embedders. It deliberately does
    /// NOT count processes parked in `waiting` with no armed timer: those are
    /// blocked on an external event (an inbound `send`/`cast`, a `timer_fired`,
    /// or an async completion), and each of those entry points raises the
    /// edge-triggered arbiter turn that resumes execution. It is not a drive
    /// signal: progress is arbiter-scheduled, never polled, and the drain
    /// result ([`WasmRunState`]) is the host-facing idle/fairness contract.
    #[must_use]
    pub fn has_pending_work(&self) -> bool {
        if self.ready.len() != 0 {
            return true;
        }
        !lock_timers(&self.native_timers).is_empty()
    }

    /// Record a cooperative native-process exit (reason + captured x(0) result).
    pub(super) fn record_native_exit(&mut self, pid: u64, reason: ExitReason, result: OwnedTerm) {
        self.native_exit_reasons.insert(pid, reason);
        self.exit_results.insert(pid, result);
    }

    /// Return the exit reason recorded for a cooperative native process, if any.
    #[must_use]
    pub fn native_exit_reason(&self, pid: u64) -> Option<ExitReason> {
        self.native_exit_reasons.get(&pid).copied()
    }

    fn native_services(&self) -> NativeServices {
        NativeServices {
            atom_table: Some(Arc::clone(&self.atom_table)),
            wasm_async_nif_facility: self.wasm_async_nif_facility.clone(),
            // WPORT-3 R2: cooperative bytecode shares the native `Deliver`
            // wheel, so `erlang:send_after/3`, `start_timer/3`, and
            // `cancel_timer/1` reach a real timer facility instead of the
            // missing-service `badarg`/`false` refusal.
            timers: Some(Arc::clone(&self.native_timers)),
            // WPORT-5 R2 item 3: the registry the scheduler already holds,
            // so export-fun/`apply` dispatch to a registered BIF succeeds
            // where static `CallExt` always did (the closure-dispatch `Undef`
            // dies).
            bif_registry: Some(Arc::clone(&self.bif_registry)),
            // WPORT-5 R2 item 4: the wrapper-injected host output sink;
            // `None` leaves the `NullSink` context default in place.
            io_sink: self.io_sink.clone(),
            ..NativeServices::default()
        }
    }

    pub(super) fn register_receive_timer(&mut self, process: &mut Process) {
        let Some(timeout) = process.receive_timeout() else {
            return;
        };
        if process.receive_timer_ref().is_some() {
            return;
        }
        let timer_id = self.next_timer_id;
        self.next_timer_id = self.next_timer_id.saturating_add(1);
        process.set_receive_timer_ref(Some(timer_id));
        self.pending_timer_schedules.push(WasmScheduledTimer {
            pid: process.pid(),
            timer_id,
            milliseconds: timeout.milliseconds,
        });
    }

    pub(super) fn apply_async_completion(&mut self, process: &mut Process) -> Option<ExitReason> {
        let completion = self.async_results.remove(&process.pid())?;
        match completion {
            WasmAsyncCompletion::Ok(term) => {
                let result = term
                    .copy_to_heap(process.heap_mut())
                    .unwrap_or_else(|_| Term::atom(Atom::BADARG));
                process.set_x_reg(0, result);
                advance_past_current_instruction(process);
                None
            }
            WasmAsyncCompletion::Error(term) => {
                let result = term
                    .copy_to_heap(process.heap_mut())
                    .unwrap_or_else(|_| Term::atom(Atom::BADARG));
                process.set_x_reg(0, result);
                Some(ExitReason::Error)
            }
        }
    }
}

/// Lock the native timer wheel, recovering from poisoning (which cannot occur
/// without a panic across the uncontended single-threaded lock, but must be
/// handled to keep the cooperative path panic-free).
fn lock_timers(timers: &Mutex<TimerWheel>) -> std::sync::MutexGuard<'_, TimerWheel> {
    timers
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

fn send_error_to_exec(error: SendError) -> ExecError {
    match error {
        SendError::HeapFull(error) => ExecError::from(error),
        SendError::InvalidBoxedTerm => ExecError::Badarg,
    }
}

fn advance_past_current_instruction(process: &mut Process) {
    if let Some(position) = process.code_position() {
        process.set_code_position(Some(CodePosition {
            module: position.module,
            instruction_pointer: position.instruction_pointer.saturating_add(1),
        }));
    }
}

#[derive(Default)]
pub(super) struct ReadyQueues {
    max: VecDeque<u64>,
    high: VecDeque<u64>,
    normal: VecDeque<u64>,
    low: VecDeque<u64>,
}

impl ReadyQueues {
    pub(super) fn push(&mut self, pid: u64, priority: Priority) {
        match priority {
            Priority::Max => self.max.push_back(pid),
            Priority::High => self.high.push_back(pid),
            Priority::Normal => self.normal.push_back(pid),
            Priority::Low => self.low.push_back(pid),
        }
    }

    pub(super) fn pop(&mut self) -> Option<u64> {
        self.max
            .pop_front()
            .or_else(|| self.high.pop_front())
            .or_else(|| self.normal.pop_front())
            .or_else(|| self.low.pop_front())
    }

    pub(super) fn len(&self) -> usize {
        self.max.len() + self.high.len() + self.normal.len() + self.low.len()
    }
}

#[cfg(test)]
#[path = "wasm_tests.rs"]
mod tests;