beamr 0.6.4

//! Minimal process-facing context exposed to native code.
//!
//! Native functions deliberately receive this allocation subset instead of the
//! full process so they cannot inspect scheduler, mailbox, or process internals.

use std::fmt;
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::atom::AtomTable;
use crate::distribution::control::DistributionSendFacility;
use crate::distribution::pg::PgFacility;
use crate::distribution::remote_link::DistributionControlFacility;
use crate::io::resource::FdInner;
use crate::io::{
    CompletionRing, IoCompletion, IoError, IoFacility, IoOp, IoSink, NullSink, ResultMode,
};
use crate::native::ets_bifs::EtsFoldlState;
use crate::native::stdlib_stubs::{
    lists_hof_bifs::ListsHofState,
    maps_bifs::{ContinuationStep, MapsHofState},
};
use crate::process::{Priority, Process};
use crate::replay::ReplayDriver;
use crate::term::Term;
use crate::term::compare;
use crate::timer::{TimerRef, TimerWheel};

use super::distribution_bifs::GlobalNameFacility;
use super::ets_bifs::EtsFacility;
use super::group_leader::GroupLeaderFacility;
use super::io_message::IoMessageFacility;
use super::links::LinkFacility;
use super::process_info_bifs::ProcessInfoFacility;
use super::registry::RegistryFacility;
use super::select::SelectFacility;
use super::spawn::SpawnFacility;
use super::supervision::SupervisionFacility;
use super::system_info_bifs::SystemInfoFacility;
use super::{NativeKey, code_management_bifs::CodeManagementFacility};

/// Handle to a contiguous run of terms registered as GC roots by
/// [`ProcessContext::with_rooted`].
///
/// Indices are relative to the handle (`0..len`), not to the process root
/// stack, so nested `with_rooted` scopes compose.
#[derive(Clone, Copy, Debug)]
pub struct RootedTerms {
    base: usize,
    len: usize,
}

/// Trampoline request from a BIF that needs interpreter re-entry.
///
/// When a BIF returns normally but needs the interpreter to call a BEAM
/// closure and use the closure's return value as the BIF's result, it stores
/// a `TrampolineRequest` in the process context. The interpreter checks for
/// this after each BIF call.
#[derive(Clone, Debug)]
pub struct TrampolineRequest {
    /// The closure (fun) term to invoke.
    pub fun: Term,
    /// Arguments to pass to the closure.
    pub args: Vec<Term>,
    /// Optional native continuation to resume after the closure returns.
    pub continuation: Option<NativeContinuation>,
}

/// Native continuation state for collection BIFs that call closures repeatedly.
#[derive(Clone, Debug)]
pub enum NativeContinuation {
    /// Continuation for maps higher-order BIFs.
    Maps(MapsHofState),
    /// Continuation for lists higher-order BIFs.
    Lists(ListsHofState),
    /// Continuation for ets:foldl/3.
    EtsFoldl(EtsFoldlState),
    /// Continuation for Aion with_timeout NIF trampoline.
    AionTimeout(AionTimeoutContinuation),
}

/// Aion timeout continuation state — carries an opaque state ID and a resume
/// function. No heap Terms are held, so GC tracing is a no-op.
#[derive(Clone)]
pub struct AionTimeoutContinuation {
    /// Opaque identifier for the timeout state in the Aion runtime.
    pub state_id: u64,
    /// Resume function called when the closure returns.
    pub resume: fn(
        AionTimeoutContinuation,
        Term,
        &mut ProcessContext<'_>,
    ) -> Result<ContinuationStep, Term>,
}

impl std::fmt::Debug for AionTimeoutContinuation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AionTimeoutContinuation")
            .field("state_id", &self.state_id)
            .finish()
    }
}

impl NativeContinuation {
    /// Visit every term held by this continuation, for GC root snapshots.
    ///
    /// Continuation state survives across closure trampolines and therefore
    /// across collections; any variant that stores terms MUST visit all of
    /// them here and in [`NativeContinuation::for_each_term_mut`], or those
    /// terms dangle after a GC. The matches are exhaustive on purpose so a
    /// new variant fails to compile until it declares its roots.
    pub(crate) fn for_each_term(&self, f: &mut dyn FnMut(Term)) {
        match self {
            Self::Maps(state) => state.for_each_term(f),
            Self::Lists(state) => state.for_each_term(f),
            Self::EtsFoldl(state) => state.for_each_term(f),
            Self::AionTimeout(_) => {}
        }
    }

    /// Visit every term held by this continuation mutably, so GC can forward
    /// moved terms in place.
    pub(crate) fn for_each_term_mut(&mut self, f: &mut dyn FnMut(&mut Term)) {
        match self {
            Self::Maps(state) => state.for_each_term_mut(f),
            Self::Lists(state) => state.for_each_term_mut(f),
            Self::EtsFoldl(state) => state.for_each_term_mut(f),
            Self::AionTimeout(_) => {}
        }
    }
}

/// File I/O continuation data used when a suspended file BIF resumes.
#[derive(Clone, Debug)]
pub enum FileIoContinuation {
    /// `erlang:open_file/2` completion.
    Open,
    /// `erlang:close_file/1` completion.
    Close { fd: Arc<FdInner> },
    /// `erlang:read_file/2` completion.
    Read { fd: Option<Arc<FdInner>> },
    /// `erlang:write_file/2` completion.
    Write {
        fd: Option<Arc<FdInner>>,
        expected_len: usize,
    },
    /// `erlang:file_seek/3` EOF completion.
    SeekEof { fd: Arc<FdInner>, offset: i64 },
    /// `erlang:file_info/1` completion.
    FileInfo,
    /// `erlang:list_dir/1` completion.
    ListDir,
    /// `erlang:make_dir/1` completion.
    MakeDir,
    /// `erlang:del_file/1` completion.
    DelFile,
    /// `erlang:del_dir/1` completion.
    DelDir,
    /// `erlang:rename/2` completion.
    Rename,
    /// `erlang:tcp_accept/1,2` completion.
    Accept,
    /// `erlang:udp_send/4` completion.
    UdpSend { expected_len: usize },
    /// `erlang:udp_recv/2,3` completion.
    UdpRecv,
    /// Active-mode UDP receive (scheduler-driven, not BIF-resumed).
    UdpActiveRecv { fd: Arc<FdInner> },
    /// Active-mode TCP receive (scheduler-driven, not BIF-resumed).
    TcpActiveRecv { fd: Arc<FdInner> },
    /// `erlang:tcp_connect/3` completion.
    Connect { fd: Arc<FdInner> },
    /// `erlang:tcp_send/2` completion.
    TcpWrite {
        fd: Arc<FdInner>,
        remaining: Vec<u8>,
        bytes_written: usize,
    },
    /// `erlang:tcp_recv/2,3` completion.
    TcpRead {
        fd: Arc<FdInner>,
        requested_len: usize,
        accumulated: Vec<u8>,
        timeout_ms: Option<u64>,
    },
}

/// Completion facility used by file BIFs to submit ring work and retrieve resume completions.
pub trait FileIoFacility: Send + Sync {
    /// Submit an operation for `pid`, tagged with the BIF continuation metadata.
    fn submit_file_io(&self, pid: u64, op: IoOp, continuation: FileIoContinuation) -> u64;

    /// Associate an already-submitted operation with `pid` and continuation metadata.
    fn track_submitted_file_io(&self, pid: u64, op_id: u64, continuation: FileIoContinuation);

    /// Take a completion that woke `pid`, if any.
    fn take_file_io_completion(&self, pid: u64) -> Option<FileIoCompletion>;

    /// Drop pending operations and future completions for `pid` after a timed wait expires.
    fn cancel_pending_file_io_for_pid(&self, pid: u64);

    /// Completion ring used by `FdInner::explicit_close`.
    fn ring(&self) -> &dyn CompletionRing;
}

/// File I/O completion delivered back to a suspended process.
#[derive(Debug)]
pub struct FileIoCompletion {
    /// Operation id returned by the ring.
    pub op_id: u64,
    /// BIF continuation associated with the operation.
    pub continuation: FileIoContinuation,
    /// Backend completion result.
    pub completion: IoCompletion,
}

/// Active TCP read-loop submission facility used by socket option BIFs.
pub trait TcpIoFacility: Send + Sync {
    /// Start an active TCP read loop for `socket` using `buf_len` read buffers.
    fn submit_active_tcp_read(&self, socket: Arc<FdInner>, buf_len: usize) -> Option<u64>;
}

/// Facility used by node-qualified spawn BIFs to request process creation on a remote node.
pub trait RemoteSpawnFacility: Send + Sync {
    /// Send a SPAWN_REQUEST to `node` and return the SPAWN_REPLY PID components.
    fn remote_spawn(
        &self,
        caller_pid: u64,
        node: crate::atom::Atom,
        module: crate::atom::Atom,
        function: crate::atom::Atom,
        args: Vec<Term>,
        options: super::spawn::SpawnOptions,
    ) -> Result<RemoteSpawnResult, RemoteSpawnError>;
}

/// Successful remote spawn reply, ready to allocate as an external PID.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct RemoteSpawnResult {
    /// Remote node that owns the spawned process.
    pub node: crate::atom::Atom,
    /// PID number on the remote node.
    pub pid_number: u64,
    /// PID serial on the remote node.
    pub serial: u64,
    /// Monitor reference when spawn_monitor was requested.
    pub monitor_reference: Option<u64>,
}

/// Error returned by remote spawn facilities.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum RemoteSpawnError {
    /// No remote spawn facility is available.
    Unavailable,
    /// The remote spawn request failed.
    Failed,
}

/// Suspend request from a BIF that wants the process to wait.
#[derive(Copy, Clone, Debug)]
pub struct SuspendRequest {
    /// Optional timeout in milliseconds. `None` means wait indefinitely.
    pub timeout_ms: Option<u64>,
    /// True for message-wakeable suspends (`request_suspend`: select-style
    /// mailbox scans and marker-style awaits, whose natives are re-entrant):
    /// any message arrival wakes the process and the native re-executes.
    /// False for gated host awaits (`request_await_suspend`): only the
    /// matching completion (or the timeout) resumes — re-executing the
    /// native would repeat its host side effect.
    pub wake_on_message: bool,
    /// Suspension call id allocated at request time when a process was
    /// attached. `None` for detached (dirty-thread) contexts: the owning
    /// scheduler thread allocates the id when it applies the request.
    pub call_id: Option<u64>,
}

/// Scheduler-side registry that makes a host-await suspension's call identity
/// visible to completion publishers (`Scheduler::wake_with_result*`) before
/// the requesting native even returns, closing the race in which a host
/// completion arrives while the process is still mid-slice.
pub trait SuspensionRegistrar: Send + Sync {
    /// Publish `(pid, call_id)` as the process's current host-await
    /// suspension. `wake_on_message` mirrors the request flavor so the wake
    /// gate knows whether plain message arrivals may wake the process.
    fn register_host_await(&self, pid: u64, call_id: u64, wake_on_message: bool);

    /// Withdraw a registration whose suspend request was abandoned (the
    /// native raised an exception after requesting suspension).
    fn cancel_host_await(&self, pid: u64, call_id: u64);
}

/// Exception classes that BIFs can request when returning `Err(reason)`.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ExceptionClass {
    /// Ordinary error exception class.
    Error,
    /// Non-local throw exception class.
    Throw,
    /// Process exit exception class.
    Exit,
}

/// Single-threaded host facility for WASM async native functions.
///
/// Implementations start host I/O for the currently executing native call,
/// arrange for a later completion, and request suspension on the supplied
/// process context.
pub trait WasmAsyncNifFacility {
    /// Start an async native function call.
    fn start_async_nif(
        &self,
        mfa: NativeKey,
        args: &[Term],
        context: &mut ProcessContext<'_>,
    ) -> Result<Term, Term>;
}

pub struct ProcessContext<'process> {
    pid: Option<u64>,
    current_native: Option<NativeKey>,
    local_node: Option<crate::distribution::Node>,
    net_kernel: Option<Arc<crate::distribution::NetKernel>>,
    distribution_send: Option<Arc<dyn DistributionSendFacility>>,
    process: Option<&'process mut Process>,
    detached_allocations: Vec<Box<[u64]>>,
    live_x: usize,
    timers: Option<Arc<Mutex<TimerWheel>>>,
    atom_table: Option<Arc<AtomTable>>,
    spawn_facility: Option<Arc<dyn SpawnFacility>>,
    remote_spawn_facility: Option<Arc<dyn RemoteSpawnFacility>>,
    link_facility: Option<Arc<dyn LinkFacility>>,
    distribution_control_facility: Option<Arc<dyn DistributionControlFacility>>,
    global_name_facility: Option<Arc<dyn GlobalNameFacility>>,
    group_leader_facility: Option<Arc<dyn GroupLeaderFacility>>,
    supervision_facility: Option<Arc<dyn SupervisionFacility>>,
    code_management_facility: Option<Arc<dyn CodeManagementFacility>>,
    process_info_facility: Option<Arc<dyn ProcessInfoFacility>>,
    registry_facility: Option<Arc<dyn RegistryFacility>>,
    select_facility: Option<Arc<dyn SelectFacility>>,
    system_info_facility: Option<Arc<dyn SystemInfoFacility>>,
    ets_facility: Option<Arc<dyn EtsFacility>>,
    pg_facility: Option<Arc<dyn PgFacility>>,
    io_facility: Option<Arc<dyn IoFacility>>,
    io_message_facility: Option<Arc<dyn IoMessageFacility>>,
    file_io_facility: Option<Arc<dyn FileIoFacility>>,
    tcp_io_facility: Option<Arc<dyn TcpIoFacility>>,
    io_sink: Arc<dyn IoSink>,
    exception_class: ExceptionClass,
    exception_stacktrace: Term,
    shutdown_requested: bool,
    trampoline: Option<TrampolineRequest>,
    suspend: Option<SuspendRequest>,
    suspension_registrar: Option<Arc<dyn SuspensionRegistrar>>,
    replay_driver: Option<Arc<Mutex<ReplayDriver>>>,
    wasm_async_nif_facility: Option<Rc<dyn WasmAsyncNifFacility>>,
    nif_private_data: Option<Arc<dyn std::any::Any + Send + Sync>>,
}

impl fmt::Debug for ProcessContext<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ProcessContext")
            .field("pid", &self.pid)
            .field("current_native", &self.current_native)
            .field("local_node", &self.local_node)
            .field("net_kernel", &self.net_kernel.as_ref().map(|_| ".."))
            .field(
                "distribution_send",
                &self.distribution_send.as_ref().map(|_| ".."),
            )
            .field("process_heap", &self.process.as_ref().map(|_| ".."))
            .field("live_x", &self.live_x)
            .field("timers", &self.timers)
            .field("atom_table", &self.atom_table.as_ref().map(|_| ".."))
            .field(
                "spawn_facility",
                &self.spawn_facility.as_ref().map(|_| ".."),
            )
            .field(
                "remote_spawn_facility",
                &self.remote_spawn_facility.as_ref().map(|_| ".."),
            )
            .field("link_facility", &self.link_facility.as_ref().map(|_| ".."))
            .field(
                "distribution_control_facility",
                &self.distribution_control_facility.as_ref().map(|_| ".."),
            )
            .field(
                "global_name_facility",
                &self.global_name_facility.as_ref().map(|_| ".."),
            )
            .field(
                "group_leader_facility",
                &self.group_leader_facility.as_ref().map(|_| ".."),
            )
            .field(
                "supervision_facility",
                &self.supervision_facility.as_ref().map(|_| ".."),
            )
            .field(
                "code_management_facility",
                &self.code_management_facility.as_ref().map(|_| ".."),
            )
            .field(
                "process_info_facility",
                &self.process_info_facility.as_ref().map(|_| ".."),
            )
            .field(
                "registry_facility",
                &self.registry_facility.as_ref().map(|_| ".."),
            )
            .field(
                "select_facility",
                &self.select_facility.as_ref().map(|_| ".."),
            )
            .field(
                "system_info_facility",
                &self.system_info_facility.as_ref().map(|_| ".."),
            )
            .field("ets_facility", &self.ets_facility.as_ref().map(|_| ".."))
            .field("pg_facility", &self.pg_facility.as_ref().map(|_| ".."))
            .field("io_facility", &self.io_facility.as_ref().map(|_| ".."))
            .field(
                "io_message_facility",
                &self.io_message_facility.as_ref().map(|_| ".."),
            )
            .field(
                "file_io_facility",
                &self.file_io_facility.as_ref().map(|_| ".."),
            )
            .field(
                "tcp_io_facility",
                &self.tcp_io_facility.as_ref().map(|_| ".."),
            )
            .field("io_sink", &"..")
            .field("exception_class", &self.exception_class)
            .field("shutdown_requested", &self.shutdown_requested)
            .field("trampoline", &self.trampoline)
            .field("suspend", &self.suspend)
            .field("exception_stacktrace", &self.exception_stacktrace)
            .field("replay_driver", &self.replay_driver.as_ref().map(|_| ".."))
            .field(
                "wasm_async_nif_facility",
                &self.wasm_async_nif_facility.as_ref().map(|_| ".."),
            )
            .field(
                "nif_private_data",
                &self.nif_private_data.as_ref().map(|_| ".."),
            )
            .finish()
    }
}

impl Default for ProcessContext<'_> {
    fn default() -> Self {
        Self::new()
    }
}

impl<'process> ProcessContext<'process> {
    /// Creates an empty process context.
    #[must_use]
    pub fn new() -> Self {
        Self {
            pid: None,
            current_native: None,
            local_node: None,
            net_kernel: None,
            distribution_send: None,
            process: None,
            detached_allocations: Vec::new(),
            live_x: 256,
            timers: None,
            atom_table: None,
            spawn_facility: None,
            remote_spawn_facility: None,
            link_facility: None,
            distribution_control_facility: None,
            global_name_facility: None,
            group_leader_facility: None,
            supervision_facility: None,
            code_management_facility: None,
            process_info_facility: None,
            registry_facility: None,
            select_facility: None,
            system_info_facility: None,
            ets_facility: None,
            pg_facility: None,
            io_facility: None,
            io_message_facility: None,
            file_io_facility: None,
            tcp_io_facility: None,
            io_sink: Arc::new(NullSink),
            exception_class: ExceptionClass::Error,
            exception_stacktrace: Term::NIL,
            trampoline: None,
            suspend: None,
            suspension_registrar: None,
            shutdown_requested: false,
            replay_driver: None,
            wasm_async_nif_facility: None,
            nif_private_data: None,
        }
    }

    /// Creates a context with timer services for asynchronous timer BIFs.
    #[must_use]
    pub fn with_timer_services(pid: u64, timers: Arc<Mutex<TimerWheel>>) -> Self {
        Self {
            pid: Some(pid),
            current_native: None,
            local_node: None,
            net_kernel: None,
            distribution_send: None,
            process: None,
            detached_allocations: Vec::new(),
            live_x: 256,
            timers: Some(timers),
            atom_table: None,
            spawn_facility: None,
            remote_spawn_facility: None,
            link_facility: None,
            distribution_control_facility: None,
            global_name_facility: None,
            group_leader_facility: None,
            supervision_facility: None,
            code_management_facility: None,
            process_info_facility: None,
            registry_facility: None,
            select_facility: None,
            system_info_facility: None,
            ets_facility: None,
            pg_facility: None,
            io_facility: None,
            io_message_facility: None,
            file_io_facility: None,
            tcp_io_facility: None,
            io_sink: Arc::new(NullSink),
            exception_class: ExceptionClass::Error,
            exception_stacktrace: Term::NIL,
            trampoline: None,
            suspend: None,
            suspension_registrar: None,
            shutdown_requested: false,
            replay_driver: None,
            wasm_async_nif_facility: None,
            nif_private_data: None,
        }
    }

    /// Return the embedder-supplied NIF private data for this runtime.
    ///
    /// This is the moral equivalent of ERTS `enif_priv_data`: one opaque
    /// value installed per scheduler (per embedded runtime instance) that
    /// every native call can recover, so embedders never need process-wide
    /// globals — which break when multiple runtimes share an OS process.
    #[must_use]
    pub fn nif_private_data(&self) -> Option<&Arc<dyn std::any::Any + Send + Sync>> {
        self.nif_private_data.as_ref()
    }

    /// Set the embedder-supplied NIF private data for this runtime.
    pub fn set_nif_private_data(&mut self, data: Option<Arc<dyn std::any::Any + Send + Sync>>) {
        self.nif_private_data = data;
    }

    /// Return the replay driver when running under deterministic replay.
    #[must_use]
    pub fn replay_driver(&self) -> Option<&Arc<Mutex<ReplayDriver>>> {
        self.replay_driver.as_ref()
    }

    /// Set the replay driver for native BIFs that consume recorded decisions.
    pub fn set_replay_driver(&mut self, driver: Option<Arc<Mutex<ReplayDriver>>>) {
        self.replay_driver = driver;
    }

    /// Return the calling process id when provided by the runtime.
    #[must_use]
    pub fn pid(&self) -> Option<u64> {
        self.pid
    }

    /// Return the currently executing native MFA, if the interpreter supplied it.
    #[must_use]
    pub fn current_native(&self) -> Option<NativeKey> {
        self.current_native
    }

    /// Set the currently executing native MFA for host facilities that need it.
    pub fn set_current_native(&mut self, native: Option<NativeKey>) {
        self.current_native = native;
    }

    /// Return the WASM async NIF bridge, if one has been configured.
    #[must_use]
    pub fn wasm_async_nif_facility(&self) -> Option<Rc<dyn WasmAsyncNifFacility>> {
        self.wasm_async_nif_facility.clone()
    }

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

    /// Return the immutable local node identity when provided by the runtime.
    #[must_use]
    pub fn local_node(&self) -> Option<crate::distribution::Node> {
        self.local_node
    }

    /// Set the immutable local node identity for node-aware BIFs.
    pub fn set_local_node(&mut self, node: Option<crate::distribution::Node>) {
        self.local_node = node;
    }

    /// Return the net-kernel distribution facade, if one has been configured.
    #[must_use]
    pub fn net_kernel(&self) -> Option<&crate::distribution::NetKernel> {
        self.net_kernel.as_deref()
    }

    /// Set the net-kernel distribution facade for distribution BIFs.
    pub fn set_net_kernel(&mut self, net_kernel: Option<Arc<crate::distribution::NetKernel>>) {
        self.net_kernel = net_kernel;
    }

    /// Return the distribution send facility, if one has been configured.
    #[must_use]
    pub fn distribution_send_facility(&self) -> Option<&dyn DistributionSendFacility> {
        self.distribution_send.as_deref()
    }

    /// Set the distribution send facility for remote PID messaging.
    pub fn set_distribution_send_facility(
        &mut self,
        facility: Option<Arc<dyn DistributionSendFacility>>,
    ) {
        self.distribution_send = facility;
    }

    /// Returns true when the attached process is re-entering a timed suspend after expiry.
    #[must_use]
    pub fn receive_timeout_expired(&self) -> bool {
        self.process
            .as_ref()
            .is_some_and(|process| process.receive_timeout().is_some())
    }

    /// Clear timed-suspend metadata after a native timed wait has resolved.
    pub fn clear_receive_timeout(&mut self) {
        if let Some(process) = self.process.as_deref_mut() {
            process.set_receive_timeout(None);
            process.set_receive_timer_ref(None);
        }
    }

    /// Cancel any file I/O operation tracked for the attached process.
    pub fn cancel_pending_file_io_for_current_process(&self) {
        if let (Some(pid), Some(facility)) = (self.pid, self.file_io_facility.as_ref()) {
            facility.cancel_pending_file_io_for_pid(pid);
        }
    }

    /// Set the calling process id.
    pub fn set_pid(&mut self, pid: Option<u64>) {
        self.pid = pid;
    }

    /// Attach the calling process for process-heap native result allocation.
    pub fn attach_process(&mut self, process: &'process mut Process, live_x: usize) {
        self.pid = Some(process.pid());
        self.process = Some(process);
        self.live_x = live_x;
    }

    /// Detach the calling process before the interpreter resumes using it directly.
    pub fn detach_process(&mut self) {
        self.process = None;
    }

    /// Return the calling process heap, when this context is heap-backed.
    #[must_use]
    pub fn process_heap(&self) -> Option<&crate::process::heap::Heap> {
        self.process.as_ref().map(|process| process.heap())
    }

    /// Return the attached calling process for native operations that must use process APIs.
    pub fn process_mut(&mut self) -> Option<&mut Process> {
        self.process.as_deref_mut()
    }

    /// Enqueue a message to the attached calling process when `target` is its pid.
    pub fn send_to_attached_self(&mut self, target: u64, message: Term) -> bool {
        let Some(process) = self.process.as_deref_mut() else {
            return false;
        };
        if process.pid() != target {
            return false;
        }
        process.mailbox_mut().push_owned(message);
        #[cfg(feature = "telemetry")]
        crate::telemetry::metrics::record_message_sent();
        true
    }

    /// Ensure the calling process has at least `words` nursery words available.
    pub fn ensure_heap_space(&mut self, words: usize) -> Result<(), Term> {
        let Some(process) = self.process.as_deref_mut() else {
            let _ = words;
            return Ok(());
        };
        crate::gc::ensure_space(process, words, self.live_x)
            .map_err(|_| Term::atom(crate::atom::Atom::BADARG))
    }

    /// Run `body` with `terms` registered as GC roots.
    ///
    /// This is the safe way for a BIF to keep heap terms alive across
    /// allocations: any collection triggered inside `body` traces and forwards
    /// the registered roots, and `body` reads their current values through the
    /// [`RootedTerms`] handle. The roots are removed when `body` returns,
    /// on both success and error paths.
    ///
    /// Returns `badarg` when no process is attached, matching the allocation
    /// helpers.
    pub fn with_rooted<R>(
        &mut self,
        terms: &[Term],
        body: impl FnOnce(&mut Self, &mut RootedTerms) -> Result<R, Term>,
    ) -> Result<R, Term> {
        let depth = {
            let process = self
                .process
                .as_deref_mut()
                .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
            let depth = process.native_root_depth();
            for term in terms {
                process.push_native_root(*term);
            }
            depth
        };
        let mut handle = RootedTerms {
            base: depth,
            len: terms.len(),
        };
        let result = body(self, &mut handle);
        if let Some(process) = self.process.as_deref_mut() {
            process.truncate_native_roots(depth);
        }
        result
    }

    /// Append a term to a rooted scope, growing it.
    ///
    /// Only valid for the innermost active `with_rooted` scope; pushing
    /// through an outer scope's handle would corrupt inner indices, so it is
    /// rejected with `badarg`. Use this for loop accumulators whose length is
    /// not known up front.
    pub fn rooted_push(&mut self, handle: &mut RootedTerms, term: Term) -> Result<(), Term> {
        let process = self
            .process
            .as_deref_mut()
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        if handle.base + handle.len != process.native_root_depth() {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        }
        process.push_native_root(term);
        handle.len += 1;
        Ok(())
    }

    /// Number of terms currently held by a rooted scope.
    #[must_use]
    pub fn rooted_len(&self, handle: &RootedTerms) -> usize {
        handle.len
    }

    /// Read the current (post-GC) value of a rooted term.
    ///
    /// Returns `badarg` when the handle is stale or no process is attached;
    /// both indicate a bug in the calling BIF rather than user error.
    pub fn rooted(&self, handle: &RootedTerms, index: usize) -> Result<Term, Term> {
        let process = self
            .process
            .as_deref()
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        if index >= handle.len {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        }
        process
            .native_root(handle.base + index)
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))
    }

    /// Overwrite a rooted slot with a new term, keeping it traced.
    ///
    /// Use this for loop accumulators: re-root the updated value after each
    /// allocation so the next collection forwards it.
    pub fn set_rooted(
        &mut self,
        handle: &RootedTerms,
        index: usize,
        term: Term,
    ) -> Result<(), Term> {
        let process = self
            .process
            .as_deref_mut()
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        if index >= handle.len {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        }
        process.set_native_root(handle.base + index, term);
        Ok(())
    }

    /// Return the spawn facility, if one has been configured.
    #[must_use]
    pub fn spawn_facility(&self) -> Option<&dyn SpawnFacility> {
        self.spawn_facility.as_deref()
    }

    /// Set the spawn facility for process creation BIFs.
    pub fn set_spawn_facility(&mut self, facility: Option<Arc<dyn SpawnFacility>>) {
        self.spawn_facility = facility;
    }

    /// Return the remote spawn facility, if one has been configured.
    #[must_use]
    pub fn remote_spawn_facility(&self) -> Option<&dyn RemoteSpawnFacility> {
        self.remote_spawn_facility.as_deref()
    }

    /// Set the remote spawn facility for node-qualified spawn BIFs.
    pub fn set_remote_spawn_facility(&mut self, facility: Option<Arc<dyn RemoteSpawnFacility>>) {
        self.remote_spawn_facility = facility;
    }

    /// Return the link facility, if one has been configured.
    #[must_use]
    pub fn link_facility(&self) -> Option<&dyn LinkFacility> {
        self.link_facility.as_deref()
    }

    /// Set the link facility for link management BIFs.
    pub fn set_link_facility(&mut self, facility: Option<Arc<dyn LinkFacility>>) {
        self.link_facility = facility;
    }

    /// Return the distribution control facility, if one has been configured.
    #[must_use]
    pub fn distribution_control_facility(&self) -> Option<&dyn DistributionControlFacility> {
        self.distribution_control_facility.as_deref()
    }

    /// Set the distribution control facility for remote link lifecycle BIFs.
    pub fn set_distribution_control_facility(
        &mut self,
        facility: Option<Arc<dyn DistributionControlFacility>>,
    ) {
        self.distribution_control_facility = facility;
    }

    /// Return the global name facility, if one has been configured.
    #[must_use]
    pub fn global_name_facility(&self) -> Option<&dyn GlobalNameFacility> {
        self.global_name_facility.as_deref()
    }

    /// Set the global name facility for `global:*_name` BIFs.
    pub fn set_global_name_facility(&mut self, facility: Option<Arc<dyn GlobalNameFacility>>) {
        self.global_name_facility = facility;
    }

    /// Return the group-leader facility, if one has been configured.
    #[must_use]
    pub fn group_leader_facility(&self) -> Option<&dyn GroupLeaderFacility> {
        self.group_leader_facility.as_deref()
    }

    /// Set the group-leader facility for process metadata BIFs.
    pub fn set_group_leader_facility(&mut self, facility: Option<Arc<dyn GroupLeaderFacility>>) {
        self.group_leader_facility = facility;
    }

    /// Return the supervision facility, if one has been configured.
    #[must_use]
    pub fn supervision_facility(&self) -> Option<&dyn SupervisionFacility> {
        self.supervision_facility.as_deref()
    }

    /// Set the supervision facility for monitor/demonitor/exit BIFs.
    pub fn set_supervision_facility(&mut self, facility: Option<Arc<dyn SupervisionFacility>>) {
        self.supervision_facility = facility;
    }

    /// Return the code-management facility, if one has been configured.
    #[must_use]
    pub fn code_management_facility(&self) -> Option<&dyn CodeManagementFacility> {
        self.code_management_facility.as_deref()
    }

    /// Set the code-management facility for hot-code BIFs.
    pub fn set_code_management_facility(
        &mut self,
        facility: Option<Arc<dyn CodeManagementFacility>>,
    ) {
        self.code_management_facility = facility;
    }

    /// Return the atom table, if one has been configured.
    #[must_use]
    pub fn atom_table(&self) -> Option<&AtomTable> {
        self.atom_table.as_deref()
    }

    /// Return a shared atom table handle, if one has been configured.
    #[must_use]
    pub fn atom_table_arc(&self) -> Option<Arc<AtomTable>> {
        self.atom_table.clone()
    }

    /// Set the atom table for type conversion BIFs.
    pub fn set_atom_table(&mut self, table: Option<Arc<AtomTable>>) {
        self.atom_table = table;
    }

    /// Return the process-info facility, if one has been configured.
    #[must_use]
    pub fn process_info_facility(&self) -> Option<&dyn ProcessInfoFacility> {
        self.process_info_facility.as_deref()
    }

    /// Set the process-info facility for process introspection BIFs.
    pub fn set_process_info_facility(&mut self, facility: Option<Arc<dyn ProcessInfoFacility>>) {
        self.process_info_facility = facility;
    }

    /// Return the registry facility, if one has been configured.
    #[must_use]
    pub fn registry_facility(&self) -> Option<&dyn RegistryFacility> {
        self.registry_facility.as_deref()
    }

    /// Set the registry facility for process name registry BIFs.
    pub fn set_registry_facility(&mut self, facility: Option<Arc<dyn RegistryFacility>>) {
        self.registry_facility = facility;
    }

    /// Schedule a timer via the runtime timer wheel.
    pub fn schedule_timer(
        &mut self,
        delay: Duration,
        target_pid: u64,
        message: Term,
    ) -> Option<TimerRef> {
        let timers = self.timers.as_ref()?;
        Some(
            timers
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .schedule(delay, target_pid, message),
        )
    }

    /// Reserve a timer reference and schedule with a message derived from it.
    pub fn schedule_timer_with_reference<F>(
        &mut self,
        delay: Duration,
        target_pid: u64,
        message: F,
    ) -> Option<TimerRef>
    where
        F: FnOnce(TimerRef) -> Term,
    {
        let timers = self.timers.as_ref()?;
        let mut timers = timers.lock().unwrap_or_else(|error| error.into_inner());
        let reference = timers.reserve_reference();
        timers.schedule_reserved(reference, delay, target_pid, message(reference))
    }

    /// Reserve a timer reference without scheduling it yet.
    pub fn reserve_timer_reference(&mut self) -> Option<TimerRef> {
        let timers = self.timers.as_ref()?;
        Some(
            timers
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .reserve_reference(),
        )
    }

    /// Schedule a message using an already reserved timer reference.
    pub fn schedule_reserved_timer(
        &mut self,
        reference: TimerRef,
        delay: Duration,
        target_pid: u64,
        message: Term,
    ) -> Option<TimerRef> {
        let timers = self.timers.as_ref()?;
        timers
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .schedule_reserved(reference, delay, target_pid, message)
    }

    /// Cancel a timer via the runtime timer wheel.
    pub fn cancel_timer(&mut self, reference: TimerRef) -> Option<Duration> {
        let timers = self.timers.as_ref()?;
        timers
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .cancel(reference)
    }

    /// Allocates a term on the calling process heap.
    ///
    /// Gate 1 only has immediate terms, so this currently returns the term
    /// unchanged. Boxed values can later route through the process heap without
    /// changing the native calling convention.
    pub const fn allocate_term(&mut self, term: Term) -> Term {
        term
    }

    // --- Select facility ---

    /// Return the select facility, if one has been configured.
    #[must_use]
    pub fn select_facility(&self) -> Option<&dyn SelectFacility> {
        self.select_facility.as_deref()
    }

    /// Set the select facility for mailbox scanning BIFs.
    pub fn set_select_facility(&mut self, facility: Option<Arc<dyn SelectFacility>>) {
        self.select_facility = facility;
    }

    // --- System info facility ---

    /// Return the system-info facility, if one has been configured.
    #[must_use]
    pub fn system_info_facility(&self) -> Option<&dyn SystemInfoFacility> {
        self.system_info_facility.as_deref()
    }

    /// Set the system-info facility for VM introspection BIFs.
    pub fn set_system_info_facility(&mut self, facility: Option<Arc<dyn SystemInfoFacility>>) {
        self.system_info_facility = facility;
    }

    // --- ETS facility ---

    /// Return the ETS facility, if one has been configured.
    #[must_use]
    pub fn ets_facility(&self) -> Option<&dyn EtsFacility> {
        self.ets_facility.as_deref()
    }

    /// Set the ETS facility for `ets` module BIFs.
    pub fn set_ets_facility(&mut self, facility: Option<Arc<dyn EtsFacility>>) {
        self.ets_facility = facility;
    }

    // --- PG facility ---

    /// Return the pg facility, if one has been configured.
    #[must_use]
    pub fn pg_facility(&self) -> Option<&dyn PgFacility> {
        self.pg_facility.as_deref()
    }

    /// Set the pg facility for process group BIFs.
    pub fn set_pg_facility(&mut self, facility: Option<Arc<dyn PgFacility>>) {
        self.pg_facility = facility;
    }

    // --- I/O facility ---

    /// Return the async I/O facility, if one has been configured.
    #[must_use]
    pub fn io_facility(&self) -> Option<&dyn IoFacility> {
        self.io_facility.as_deref()
    }

    /// Set the async I/O facility for I/O BIFs.
    pub fn set_io_facility(&mut self, facility: Option<Arc<dyn IoFacility>>) {
        self.io_facility = facility;
    }

    // --- IO message facility ---

    /// Return the IO message facility, if one has been configured.
    #[must_use]
    pub fn io_message_facility(&self) -> Option<&dyn IoMessageFacility> {
        self.io_message_facility.as_deref()
    }

    /// Set the IO message facility for group-leader protocol BIFs.
    pub fn set_io_message_facility(&mut self, facility: Option<Arc<dyn IoMessageFacility>>) {
        self.io_message_facility = facility;
    }

    /// Submit an I/O operation for the attached pid and request suspension.
    pub fn submit_io_and_suspend(&mut self, op: IoOp, mode: ResultMode) -> Result<(), IoError> {
        let pid = self.pid.ok_or(IoError::MissingPid)?;
        if self.io_facility.is_none() {
            return Err(IoError::Unavailable);
        }
        // Register the host-await suspension BEFORE the submission can race a
        // completion: the bridge publishes the result under the call id it
        // finds registered for the pid.
        let _call_id = self.request_await_suspend(None);
        let Some(facility) = self.io_facility.as_ref() else {
            return Err(IoError::Unavailable);
        };
        if let Err(error) = facility.submit_and_suspend_for_pid(pid, op, mode) {
            self.cancel_requested_suspend();
            return Err(error);
        }
        Ok(())
    }

    // --- File I/O facility ---

    /// Return the file I/O facility, if one has been configured.
    #[must_use]
    pub fn file_io_facility(&self) -> Option<&dyn FileIoFacility> {
        self.file_io_facility.as_deref()
    }

    /// Set the file I/O facility for completion-ring backed file BIFs.
    pub fn set_file_io_facility(&mut self, facility: Option<Arc<dyn FileIoFacility>>) {
        self.file_io_facility = facility;
    }

    // --- TCP I/O facility ---

    /// Return the TCP I/O facility, if one has been configured.
    #[must_use]
    pub fn tcp_io_facility(&self) -> Option<&dyn TcpIoFacility> {
        self.tcp_io_facility.as_deref()
    }

    /// Set the TCP I/O facility for active-mode socket BIFs.
    pub fn set_tcp_io_facility(&mut self, facility: Option<Arc<dyn TcpIoFacility>>) {
        self.tcp_io_facility = facility;
    }

    /// Submit a file I/O operation and suspend the calling process until completion.
    pub fn submit_file_io(
        &mut self,
        op: IoOp,
        continuation: FileIoContinuation,
    ) -> Result<u64, Term> {
        self.submit_file_io_with_timeout(op, continuation, None)
    }

    /// Submit a file I/O operation and suspend the calling process until completion or timeout.
    pub fn submit_file_io_with_timeout(
        &mut self,
        op: IoOp,
        continuation: FileIoContinuation,
        timeout_ms: Option<u64>,
    ) -> Result<u64, Term> {
        let pid = self
            .pid
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        if self.file_io_facility.is_none() {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        }
        // Register the host-await suspension BEFORE the submission can race a
        // completion arriving on the ring poller thread.
        let _call_id = self.request_await_suspend(timeout_ms);
        let facility = self
            .file_io_facility
            .as_ref()
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        let op_id = facility.submit_file_io(pid, op, continuation);
        Ok(op_id)
    }

    /// Associate an already-submitted file I/O operation with this process.
    pub fn track_submitted_file_io(
        &mut self,
        op_id: u64,
        continuation: FileIoContinuation,
    ) -> Result<(), Term> {
        let pid = self
            .pid
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        let facility = self
            .file_io_facility
            .as_ref()
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))?;
        facility.track_submitted_file_io(pid, op_id, continuation);
        Ok(())
    }

    /// Take the completion used to resume the currently executing file BIF.
    pub fn take_file_io_completion(&self) -> Option<FileIoCompletion> {
        let pid = self.pid?;
        self.file_io_facility.as_ref()?.take_file_io_completion(pid)
    }

    /// Return the completion ring backing file I/O resources.
    #[must_use]
    pub fn file_completion_ring(&self) -> Option<&dyn CompletionRing> {
        self.file_io_facility
            .as_ref()
            .map(|facility| facility.ring())
    }

    /// Store a value in the attached process dictionary.
    pub fn dict_put(&mut self, key: Term, value: Term) -> Result<Term, Term> {
        let Some(process) = self.process.as_deref_mut() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_put(key, value))
    }

    /// Return the attached process group leader.
    pub fn group_leader(&self) -> Result<Term, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.group_leader())
    }

    /// Return the attached process scheduling priority.
    pub fn priority(&self) -> Result<Priority, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.priority())
    }

    /// Set the attached process scheduling priority and return its old value.
    pub fn set_priority(&mut self, priority: Priority) -> Result<Priority, Term> {
        let Some(process) = self.process.as_deref_mut() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        let old_priority = process.priority();
        process.set_priority(priority);
        Ok(old_priority)
    }

    /// Set the attached process group leader when it matches `pid`.
    pub fn set_attached_group_leader(&mut self, pid: u64, group_leader: Term) -> bool {
        let Some(process) = self.process.as_deref_mut() else {
            return false;
        };
        if process.pid() != pid {
            return false;
        }
        process.set_group_leader(group_leader);
        true
    }

    /// Fetch a value from the attached process dictionary.
    pub fn dict_get(&self, key: Term) -> Result<Term, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_get(key))
    }

    /// Copy all attached process dictionary entries in current vector order.
    pub fn dict_get_all(&self) -> Result<Vec<(Term, Term)>, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_get_all().to_vec())
    }

    /// Count attached process dictionary entries without copying their terms.
    pub fn dict_len(&self) -> Result<usize, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_get_all().len())
    }

    /// Remove a value from the attached process dictionary.
    pub fn dict_erase(&mut self, key: Term) -> Result<Term, Term> {
        let Some(process) = self.process.as_deref_mut() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_erase(key))
    }

    /// Remove and return all attached process dictionary entries.
    pub fn dict_erase_all(&mut self) -> Result<Vec<(Term, Term)>, Term> {
        let Some(process) = self.process.as_deref_mut() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_erase_all())
    }

    /// Copy all dictionary keys whose values exactly match `value`.
    pub fn dict_get_keys(&self, value: Term) -> Result<Vec<Term>, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process.dict_get_keys(value))
    }

    /// Count dictionary keys whose values exactly match `value` without copying terms.
    pub fn dict_count_keys_for_value(&self, value: Term) -> Result<usize, Term> {
        let Some(process) = self.process.as_ref() else {
            return Err(Term::atom(crate::atom::Atom::BADARG));
        };
        Ok(process
            .dict_get_all()
            .iter()
            .filter(|(_, existing_value)| compare::exact_eq(*existing_value, value))
            .count())
    }

    /// Return the configured output sink for `io` module BIFs.
    #[must_use]
    pub fn io_sink(&self) -> &dyn IoSink {
        self.io_sink.as_ref()
    }

    /// Set the output sink for `io` module BIFs.
    pub fn set_io_sink(&mut self, sink: Arc<dyn IoSink>) {
        self.io_sink = sink;
    }

    /// Request runtime shutdown after the current BIF returns.
    pub fn request_shutdown(&mut self) {
        self.shutdown_requested = true;
    }

    /// Take and clear the shutdown request flag.
    pub fn take_shutdown_request(&mut self) -> bool {
        let requested = self.shutdown_requested;
        self.shutdown_requested = false;
        requested
    }

    /// Set the exception class to use if this BIF returns `Err(reason)`.
    pub fn set_exception_class(&mut self, class: ExceptionClass) {
        self.exception_class = class;
    }

    /// Take the requested exception class, resetting subsequent errors to `error`.
    pub fn take_exception_class(&mut self) -> ExceptionClass {
        let class = self.exception_class;
        self.exception_class = ExceptionClass::Error;
        class
    }

    // --- Trampoline ---

    /// Store a trampoline request for the interpreter to execute.
    ///
    /// The interpreter checks for a trampoline after each BIF call. When
    /// present, it sets up the closure call and uses the closure's return
    /// value as the BIF's return value.
    pub fn set_trampoline(&mut self, fun: Term, args: Vec<Term>) {
        self.trampoline = Some(TrampolineRequest {
            fun,
            args,
            continuation: None,
        });
    }

    /// Store a trampoline request with native continuation state.
    pub fn set_continuation_trampoline(
        &mut self,
        fun: Term,
        args: Vec<Term>,
        continuation: NativeContinuation,
    ) {
        self.trampoline = Some(TrampolineRequest {
            fun,
            args,
            continuation: Some(continuation),
        });
    }

    /// Take the trampoline request, clearing it from the context.
    ///
    /// Returns `None` if no trampoline was requested.
    pub fn take_trampoline(&mut self) -> Option<TrampolineRequest> {
        self.trampoline.take()
    }

    /// Check whether a trampoline request is pending.
    #[must_use]
    pub fn has_trampoline(&self) -> bool {
        self.trampoline.is_some()
    }

    // --- Suspend ---

    /// Request a message-wakeable suspension.
    ///
    /// Any message arrival wakes the process and re-executes the native at
    /// the suspended call instruction (the `select` mailbox-scan protocol,
    /// and the embedder marker-await pattern built on
    /// `Scheduler::enqueue_atom_message`). The native must therefore be
    /// re-entrant. A completion published for the returned call id
    /// (`Scheduler::wake_with_result*`) is also delivered — applied into x0
    /// exactly at this suspension, or dropped as stale once the suspension
    /// is superseded, never applied blind at a later park position.
    ///
    /// Natives that submit one-shot host work (a query, a file operation)
    /// must use [`ProcessContext::request_await_suspend`] instead: a
    /// re-execution triggered by an unrelated message would re-submit that
    /// work.
    ///
    /// Returns the allocated suspension call id when a process is attached;
    /// detached (dirty-thread) contexts return `None` and the owning
    /// scheduler thread allocates the id when it applies the request.
    pub fn request_suspend(&mut self, timeout_ms: Option<u64>) -> Option<u64> {
        self.request_suspend_flavor(timeout_ms, true)
    }

    /// Request a result-gated host-await suspension.
    ///
    /// The process parks until the completion identified by the returned
    /// suspension call id is delivered (`Scheduler::wake_with_result_for`,
    /// the pid-resolved `Scheduler::wake_with_result`, or a file-I/O
    /// completion), or until the optional timeout fires and re-executes the
    /// native at the suspended call instruction. Plain message arrivals do
    /// NOT wake the process — re-executing the native would repeat its host
    /// side effect.
    pub fn request_await_suspend(&mut self, timeout_ms: Option<u64>) -> Option<u64> {
        self.request_suspend_flavor(timeout_ms, false)
    }

    fn request_suspend_flavor(
        &mut self,
        timeout_ms: Option<u64>,
        wake_on_message: bool,
    ) -> Option<u64> {
        let call_id = self
            .process
            .as_deref_mut()
            .map(Process::allocate_suspension_call_id);
        self.suspend = Some(SuspendRequest {
            timeout_ms,
            wake_on_message,
            call_id,
        });
        if let (Some(pid), Some(call_id), Some(registrar)) =
            (self.pid, call_id, self.suspension_registrar.as_ref())
        {
            registrar.register_host_await(pid, call_id, wake_on_message);
        }
        call_id
    }

    /// Withdraw a pending suspend request whose native call is unwinding
    /// (e.g. it raised an exception after requesting suspension), so the
    /// published host-await registration cannot strand a stale call id.
    pub fn cancel_requested_suspend(&mut self) {
        let Some(request) = self.suspend.take() else {
            return;
        };
        self.cancel_suspend_request(&request);
    }

    /// Withdraw the host-await registration of an already-taken suspend
    /// request (used when the native's exception path unwinds after the
    /// request was extracted from this context).
    pub fn cancel_suspend_request(&self, request: &SuspendRequest) {
        if let (Some(pid), Some(call_id), Some(registrar)) = (
            self.pid,
            request.call_id,
            self.suspension_registrar.as_ref(),
        ) {
            registrar.cancel_host_await(pid, call_id);
        }
    }

    /// Take the suspend request, clearing it from the context.
    pub fn take_suspend(&mut self) -> Option<SuspendRequest> {
        self.suspend.take()
    }

    /// Set the scheduler-side registrar that publishes host-await call ids.
    pub fn set_suspension_registrar(&mut self, registrar: Option<Arc<dyn SuspensionRegistrar>>) {
        self.suspension_registrar = registrar;
    }

    // --- Exception metadata ---

    /// Set the stacktrace to use if the current BIF returns `Err(reason)`.
    pub fn set_exception_stacktrace(&mut self, trace: Term) {
        self.exception_stacktrace = trace;
    }

    /// Take the pending exception stacktrace, resetting subsequent BIF errors to `[]`.
    pub fn take_exception_stacktrace(&mut self) -> Term {
        let stacktrace = self.exception_stacktrace;
        self.exception_stacktrace = Term::NIL;
        stacktrace
    }

    // --- Heap allocation helpers ---

    fn alloc_words(&mut self, words: usize) -> Result<&mut [u64], Term> {
        self.ensure_heap_space(words)?;
        self.alloc_words_prereserved(words)
    }

    /// Keep detached allocations alive by moving them into an owned term.
    ///
    /// Dirty native calls run without an attached process heap. Terms allocated
    /// in that detached context point into `detached_allocations`, so the dirty
    /// completion path must preserve those allocations until it can copy the
    /// returned term onto the resuming process heap.
    pub fn take_detached_result(&mut self, root: Term) -> Option<crate::ets::OwnedTerm> {
        if self.detached_allocations.is_empty() {
            None
        } else {
            Some(crate::ets::OwnedTerm::from_allocations(
                root,
                std::mem::take(&mut self.detached_allocations),
            ))
        }
    }

    /// Allocate heap words WITHOUT triggering GC. Caller must have already
    /// called `ensure_heap_space` for the total allocation budget. Panics
    /// (via alloc_slice error) if insufficient space remains.
    fn alloc_words_prereserved(&mut self, words: usize) -> Result<&mut [u64], Term> {
        if let Some(process) = self.process.as_deref_mut() {
            return process
                .heap_mut()
                .alloc_slice(words)
                .map_err(|_| Term::atom(crate::atom::Atom::BADARG));
        }

        self.detached_allocations
            .push(vec![0; words].into_boxed_slice());
        self.detached_allocations
            .last_mut()
            .map(|words| words.as_mut())
            .ok_or_else(|| Term::atom(crate::atom::Atom::BADARG))
    }
}

mod alloc;

#[cfg(test)]
mod tests;