beamr 0.16.2

//! Supervision integration for the scheduler — exit signal propagation
//! through links and DOWN message delivery through monitors.
//!
//! Extracted from `mod.rs` to keep per-file line counts within the project
//! constraint (500 lines).

use std::{collections::VecDeque, sync::Arc};

use crate::atom::Atom;
use crate::distribution::control::{
    ControlDelivery, ControlRegistry, DistributionSendError, DistributionSendFacility,
    encode_send_frame,
};
use crate::distribution::remote_link::{DistributionControlFacility, RemoteLinkError};
use crate::ets::{EtsError, EtsTable, EtsTableId, EtsTableMetadata};
use crate::io::{CompletionRing, IoOp};
use crate::namespace::NamespaceId;
use crate::native::CapabilitySet;
use crate::native::ets_bifs::EtsFacility;
use crate::native::io_message::IoMessageFacility;
use crate::native::links::{LinkError, LinkFacility};
use crate::native::spawn::{
    SpawnError, SpawnFacility, SpawnMonitorResult, SpawnOptions, SpawnOptionsResult,
};
use crate::native::supervision::{MonitorResult, SupervisionError, SupervisionFacility};
use crate::native::{FileIoCompletion, FileIoContinuation, FileIoFacility};
use crate::process::heap::DEFAULT_HEAP_SIZE;
use crate::process::{ExitReason, Priority, Process, ProcessStatus, RemotePid};
use crate::scheduler::process_slot::PendingExitSource;
use crate::supervision::link;
use crate::supervision::monitor;
use crate::term::Term;
use crate::term::boxed;
use crate::term::pid_ref::PidRef;

use super::execution::{cleanup_exited_process, wake_process};
use super::spawning::SpawnRequest;
use super::{
    ProcessSlot, ScheduledProcess, SharedState, dist_control_out, lock_or_recover,
    namespace_registry,
};

/// Propagate exit signals through links and deliver DOWN messages through
/// monitors when a process exits. Uses a worklist pattern to handle cascade
/// deaths iteratively rather than recursively.
pub(super) fn propagate_exit(shared: &SharedState, pid: u64, reason: ExitReason) {
    // Collect linked PIDs from the exiting process.
    let linked_pids = take_links_from(shared, pid);
    let remote_links = take_remote_links_from(shared, pid);
    let terminal_reason = link::terminal_reason(reason);

    // Deliver DOWN messages to all monitors of this process.
    deliver_down_messages(shared, pid, reason);

    // Mark dead in link_set for future link_pid() calls.
    {
        let mut ls = lock_or_recover(&shared.link_set);
        ls.process_exited_tombstone(pid, terminal_reason);
    }

    // Send remote EXIT controls for cross-node links.
    for remote in remote_links {
        send_remote_exit(shared, pid, remote, terminal_reason);
    }

    // Process link cascade with worklist pattern.
    // The signal sent through links is always `terminal_reason`: Kill becomes
    // Killed, matching BEAM semantics where only a direct exit signal is
    // untrappable — propagation through links always uses the terminal reason.
    let mut worklist: VecDeque<(u64, u64, ExitReason)> = linked_pids
        .into_iter()
        .map(|linked_pid| (pid, linked_pid, terminal_reason))
        .collect();
    while let Some((source_pid, target_pid, signal_reason)) = worklist.pop_front() {
        let cascade = process_exit_signal(shared, source_pid, target_pid, signal_reason);
        worklist.extend(cascade);
    }
}

pub(super) fn register_distribution_control_handler(shared: &Arc<SharedState>) {
    // Capture a `Weak`, NOT an `Arc`: the handler is stored inside
    // `ConnectionManagerInner.control_frame_handler`, and `SharedState` owns that
    // `ConnectionManager`. A strong capture would form the permanent cycle
    // `SharedState -> distribution_connections -> control_frame_handler ->
    // Arc<SharedState>`, leaking every scheduler's `SharedState` forever. Mirror
    // the connection-down hook in `scheduler/mod.rs` (the pg-purge closure): hold
    // a `Weak` and `upgrade()` per inbound frame, dropping the frame if the
    // scheduler has already gone.
    // No manager to register on when distribution is Disabled (spec §3.6):
    // inbound control frames cannot arrive without a connection.
    let Some(dist) = shared.distribution() else {
        return;
    };
    let shared_for_handler = Arc::downgrade(shared);
    dist.connections()
        .register_control_frame_handler_with_origin(move |origin, control, payload| {
            if control.is_empty() && payload.is_empty() {
                // Keepalive: the all-zero 8-byte header reaches the handler as
                // an empty control + payload. It is liveness traffic (already
                // consumed by the read loop's inbound-activity refresh), not a
                // control frame — decoding it would fail and count every
                // healthy heartbeat as a dropped "dispatch" frame in telemetry.
                return;
            }
            let Some(shared) = shared_for_handler.upgrade() else {
                // Scheduler dropped: there is nothing left to deliver to.
                return;
            };
            let facility = SchedulerDistributionSendFacility {
                shared: Arc::clone(&shared),
            };
            let sinks = crate::distribution::control_link::ControlSinks {
                delivery: &facility,
                registry: Some(&facility),
                pg: Some(&facility),
                links: Some(&facility),
                // The authenticated handshake name that keys the connection
                // table: a link control whose `from.node` differs is a forgery
                // and is dropped inside `dispatch_frame`.
                origin_node: Some(origin),
                local_node: Some(shared.local_node.name),
            };
            if let Err(_error) = crate::distribution::control_link::dispatch_frame(
                control,
                payload,
                &shared.atom_table,
                &sinks,
            ) {
                // A malformed, misaddressed, or otherwise rejected frame is
                // dropped without killing the read loop (scenario 9) — but
                // never invisibly.
                #[cfg(feature = "telemetry")]
                crate::telemetry::metrics::record_control_frame_dropped("dispatch");
            }
        });
}

pub(super) fn deliver_ets_transfer(
    shared: &SharedState,
    recipient_pid: u64,
    table_id: EtsTableId,
    from_pid: u64,
    data: Term,
    atom_table: &crate::atom::AtomTable,
) -> bool {
    let Some(entry) = shared.process_bodies.get(&recipient_pid) else {
        return false;
    };
    let transfer_atom = atom_table.intern("ETS-TRANSFER");
    let mut slot = lock_or_recover(&entry);
    let delivered = match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(process)) => {
            let Some(message) =
                build_ets_transfer_message(process, transfer_atom, table_id, from_pid, data)
            else {
                return false;
            };
            process.mailbox_mut().push_owned(message);
            true
        }
        ProcessSlot::Executing(metadata) => {
            let Ok(data) = crate::ets::copy_term_to_ets(data) else {
                return false;
            };
            metadata.pending_ets_transfer_messages.push(
                super::process_slot::PendingEtsTransferMessage {
                    table_id,
                    from_pid,
                    data,
                },
            );
            true
        }
        ProcessSlot::Absent => false,
    };
    drop(slot);
    if delivered {
        wake_process(shared, recipient_pid);
    }
    delivered
}

pub(super) fn build_ets_transfer_message(
    process: &mut Process,
    transfer_atom: Atom,
    table_id: EtsTableId,
    from_pid: u64,
    data: Term,
) -> Option<Term> {
    let table = Term::try_small_int(i64::try_from(table_id).ok()?)?;
    let from = Term::try_pid(from_pid)?;
    let data = crate::ets::copy_term_to_heap(data, process.heap_mut()).ok()?;
    let words = process.heap_mut().alloc_slice(5).ok()?;
    boxed::write_tuple(words, &[Term::atom(transfer_atom), table, from, data])
}

/// Real `GroupLeaderFacility` backed by the scheduler's shared state.
pub(super) struct SchedulerGroupLeaderFacility {
    pub(super) shared: Arc<SharedState>,
}

impl crate::native::GroupLeaderFacility for SchedulerGroupLeaderFacility {
    fn set_group_leader(
        &self,
        pid: u64,
        leader: Term,
    ) -> Result<(), crate::native::group_leader::GroupLeaderError> {
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return Err(crate::native::group_leader::GroupLeaderError::NoProc);
        };
        let Some(entry) = self.shared.process_bodies.get(&pid) else {
            return Err(crate::native::group_leader::GroupLeaderError::NoProc);
        };
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(ScheduledProcess(process)) => {
                process.set_group_leader(leader);
                Ok(())
            }
            ProcessSlot::Executing(metadata) => {
                metadata.group_leader = leader;
                Ok(())
            }
            ProcessSlot::Absent => Err(crate::native::group_leader::GroupLeaderError::NoProc),
        }
    }

    fn group_leader(
        &self,
        pid: u64,
    ) -> Result<Term, crate::native::group_leader::GroupLeaderError> {
        let Some(entry) = self.shared.process_bodies.get(&pid) else {
            return Err(crate::native::group_leader::GroupLeaderError::NoProc);
        };
        let slot = lock_or_recover(&entry);
        match &*slot {
            ProcessSlot::Present(ScheduledProcess(process)) => Ok(process.group_leader()),
            ProcessSlot::Executing(metadata) => Ok(metadata.group_leader),
            ProcessSlot::Absent => Err(crate::native::group_leader::GroupLeaderError::NoProc),
        }
    }
}

/// Take the link set from an exiting process. The process body may already
/// have been removed, absent, or executing, so handle each slot explicitly.
pub(super) fn take_links_from(shared: &SharedState, pid: u64) -> Vec<u64> {
    if let Some(entry) = shared.process_bodies.get(&pid) {
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(ScheduledProcess(process)) => {
                return process.take_links();
            }
            ProcessSlot::Executing(metadata) => return metadata.links.clone(),
            ProcessSlot::Absent => {}
        }
    }
    Vec::new()
}

pub(super) fn take_remote_links_from(shared: &SharedState, pid: u64) -> Vec<RemotePid> {
    if let Some(entry) = shared.process_bodies.get(&pid) {
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(ScheduledProcess(process)) => {
                return process.take_remote_links();
            }
            ProcessSlot::Executing(metadata) => {
                return std::mem::take(&mut metadata.remote_links);
            }
            ProcessSlot::Absent => {}
        }
    }
    Vec::new()
}

/// Record a remote link on a local process. `false` means dead or absent
/// target ONLY: a duplicate link is idempotent success (OTP parity, ruling 2),
/// so callers may key a noproc reply / `BadTarget` on the return value. The
/// status check happens under the slot lock — no TOCTOU. An
/// Executing-but-tombstoned target is deliberately linked (ruling 3): at
/// store-back the tombstone cleanup runs `propagate_exit`, which sends the
/// linker a wire EXIT with the true terminal reason — self-healing.
pub(super) fn establish_remote_link(
    shared: &SharedState,
    local_pid: u64,
    remote: RemotePid,
) -> bool {
    let Some(entry) = shared.process_bodies.get(&local_pid) else {
        return false;
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(process)) => {
            if matches!(process.status(), ProcessStatus::Exited(_)) {
                return false;
            }
            let _ = process.add_remote_link(remote); // duplicate = idempotent success
            true
        }
        ProcessSlot::Executing(metadata) => {
            metadata.add_remote_link(remote);
            true
        }
        ProcessSlot::Absent => false,
    }
}

pub(super) fn remove_remote_link(shared: &SharedState, local_pid: u64, remote: RemotePid) -> bool {
    let Some(entry) = shared.process_bodies.get(&local_pid) else {
        return false;
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(process)) => process.remove_remote_link(remote),
        ProcessSlot::Executing(metadata) => metadata.remove_remote_link(remote),
        ProcessSlot::Absent => false,
    }
}

// The remote-supervision appliers live in `remote_supervision.rs` (moved out
// to keep this file inside the per-file line budget); re-exported here so the
// connection-event subscriber (`connection_lifecycle::handle_connection_event`)
// and existing tests keep their `supervision_integration::` paths.
pub(crate) use super::remote_supervision::{connection_down, process_remote_exit_signal};

fn send_remote_exit(shared: &SharedState, caller_pid: u64, target: RemotePid, reason: ExitReason) {
    dist_control_out::send_exit_linked(shared, caller_pid, target, reason);
}

/// Deliver a single exit signal to a linked process. Returns any cascade
/// entries (source_pid, linked_pid, reason) for processes that must also die.
fn process_exit_signal(
    shared: &SharedState,
    source_pid: u64,
    target_pid: u64,
    reason: ExitReason,
) -> Vec<(u64, u64, ExitReason)> {
    let Some(entry) = shared.process_bodies.get(&target_pid) else {
        return Vec::new();
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(target)) => {
            // Already exited? Nothing to do.
            if matches!(target.status(), ProcessStatus::Exited(_)) {
                return Vec::new();
            }

            // Remove the reverse link.
            target.remove_link(source_pid);

            let should_die =
                reason == ExitReason::Kill || (reason != ExitReason::Normal && !target.trap_exit());

            if should_die {
                // Kill signal bypasses trap_exit and propagates as 'killed'.
                let propagated_reason = link::terminal_reason(reason);
                if reason == ExitReason::Kill {
                    target.set_trap_exit(false);
                }

                // Collect this process's links for cascade before terminating
                // — BOTH halves: local links feed the worklist, remote links
                // each get a wire EXIT below (dropping them with the body
                // would silently sever the peer's half-link).
                let cascade_links: Vec<u64> = target
                    .take_links()
                    .into_iter()
                    .filter(|linked_pid| *linked_pid != source_pid)
                    .collect();
                let remote_links = target.take_remote_links();

                target.terminate(propagated_reason);

                // Record tombstone and remove resources owned by the terminated process.
                shared.insert_exit_tombstone(target_pid, propagated_reason);
                let _deleted_tables = shared.transfer_or_delete_tables_owned_by(target_pid);
                {
                    let mut ls = lock_or_recover(&shared.link_set);
                    ls.process_exited_tombstone(target_pid, propagated_reason);
                }

                // Deliver DOWN messages for monitors on the cascaded process.
                // Must drop the slot lock first to avoid deadlock.
                drop(slot);
                drop(entry);
                deliver_down_messages(shared, target_pid, propagated_reason);

                // Send the terminal EXIT to every remote half-link, exactly
                // as full propagation would (`propagate_exit`).
                for remote in remote_links {
                    send_remote_exit(shared, target_pid, remote, propagated_reason);
                }

                // Finalize exactly like a direct kill — table, BODY, owned
                // fd resources, wait set, timers, suspensions, pg, metrics.
                // Propagation is NOT re-run: this arm already took the
                // target's links for the cascade worklist and delivered its
                // DOWNs above. (The previous hand-written subset here left
                // the body, fds, pg memberships, and metric state stranded
                // on every cascade kill of a stored process.)
                super::execution::finalize_exited_process(shared, target_pid, propagated_reason);

                cascade_links
                    .into_iter()
                    .map(|linked_pid| (target_pid, linked_pid, propagated_reason))
                    .collect()
            } else if target.trap_exit() {
                // Process traps exits: deliver {EXIT, SourcePid, Reason} as message.
                link::enqueue_exit_message_pub(target, source_pid, reason);

                // Wake the process if it was waiting for a message.
                let target_pid_copy = target_pid;
                drop(slot);
                drop(entry);
                wake_process(shared, target_pid_copy);

                Vec::new()
            } else {
                // Normal exit to non-trapping process: no action needed.
                Vec::new()
            }
        }
        ProcessSlot::Executing(metadata) => {
            metadata.remove_link(source_pid);
            let should_die =
                reason == ExitReason::Kill || (reason != ExitReason::Normal && !metadata.trap_exit);

            if should_die {
                let propagated_reason = link::terminal_reason(reason);
                if reason == ExitReason::Kill {
                    metadata.trap_exit = false;
                }
                let cascade_links: Vec<u64> = metadata
                    .links
                    .iter()
                    .copied()
                    .filter(|linked_pid| *linked_pid != source_pid)
                    .collect();
                shared.insert_exit_tombstone(target_pid, propagated_reason);
                let _deleted_tables = shared.transfer_or_delete_tables_owned_by(target_pid);
                {
                    let mut ls = lock_or_recover(&shared.link_set);
                    ls.process_exited_tombstone(target_pid, propagated_reason);
                }
                drop(slot);
                drop(entry);
                deliver_down_messages(shared, target_pid, propagated_reason);

                cascade_links
                    .into_iter()
                    .map(|linked_pid| (target_pid, linked_pid, propagated_reason))
                    .collect()
            } else if metadata.trap_exit {
                // Process traps exits: queue {EXIT, SourcePid, Reason} for
                // delivery when the slice completes. This mirrors the Present
                // arm's `else if target.trap_exit()` and MUST include NORMAL
                // exits (OTP delivers {'EXIT', Pid, normal} for a normal
                // linked exit to a trapping process). `should_die` has already
                // peeled off Kill and abnormal-non-trapping cases, so reaching
                // here means a trapping target for any non-kill reason.
                metadata
                    .pending_exit_messages
                    .push((PendingExitSource::Local(source_pid), reason));
                drop(slot);
                drop(entry);
                wake_process(shared, target_pid);
                Vec::new()
            } else {
                Vec::new()
            }
        }
        ProcessSlot::Absent => Vec::new(),
    }
}

/// Deliver DOWN messages to all watchers of `target_pid`.
fn deliver_down_messages(shared: &SharedState, target_pid: u64, reason: ExitReason) {
    // Collect monitor info under monitor_set lock, then release.
    let watcher_info: Vec<(u64, u64)> = {
        let mut ms = lock_or_recover(&shared.monitor_set);
        ms.collect_watchers_and_remove(target_pid, reason)
    };

    for (watcher_pid, reference) in watcher_info {
        deliver_single_down_and_wake(shared, watcher_pid, reference, target_pid, reason);
    }
}

/// Deliver a single DOWN message to a watcher process. Returns true if
/// the message was successfully enqueued.
fn deliver_single_down(
    shared: &SharedState,
    watcher_pid: u64,
    reference: u64,
    target_pid: u64,
    reason: ExitReason,
) -> bool {
    let Some(entry) = shared.process_bodies.get(&watcher_pid) else {
        return false;
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(watcher)) => {
            if matches!(watcher.status(), ProcessStatus::Exited(_)) {
                return false;
            }

            watcher.remove_monitor(reference);
            monitor::enqueue_down_message_pub(watcher, reference, target_pid, reason);
            true
        }
        ProcessSlot::Executing(metadata) => {
            metadata.remove_monitor(reference);
            metadata
                .pending_down_messages
                .push((reference, target_pid, reason));
            true
        }
        ProcessSlot::Absent => false,
    }
}

/// Admit one DOWN and perform the wake only after the mailbox or executing-slot
/// metadata contains it. Returns whether a live watcher accepted the message.
fn deliver_single_down_and_wake(
    shared: &SharedState,
    watcher_pid: u64,
    reference: u64,
    target_pid: u64,
    reason: ExitReason,
) -> bool {
    let delivered = deliver_single_down(shared, watcher_pid, reference, target_pid, reason);
    if delivered {
        // For Present this closes the enqueue-to-park race. For Executing the
        // wake is normally a no-op because the watcher is not registered in
        // the wait set yet; store-back merges pending_down_messages before the
        // Wait arm's post-registration mailbox recheck, which self-wakes it.
        wake_process(shared, watcher_pid);
    }
    delivered
}

/// Build the `NativeServices` bundle for a scheduler time slice.
pub(super) fn build_native_services(
    shared: &Arc<SharedState>,
    namespace_id: NamespaceId,
) -> crate::interpreter::NativeServices {
    let spawn: Arc<dyn SpawnFacility> = Arc::new(SchedulerSpawnFacility {
        shared: Arc::clone(shared),
        namespace_id,
    });
    let link: Arc<dyn crate::native::links::LinkFacility> = Arc::new(SchedulerLinkFacility {
        shared: Arc::clone(shared),
    });
    let group_leader: Arc<dyn crate::native::GroupLeaderFacility> =
        Arc::new(SchedulerGroupLeaderFacility {
            shared: Arc::clone(shared),
        });
    let supervision: Arc<dyn crate::native::supervision::SupervisionFacility> =
        Arc::new(SchedulerSupervisionFacility {
            shared: Arc::clone(shared),
        });
    let process_info: Arc<dyn crate::native::ProcessInfoFacility> =
        Arc::new(SchedulerProcessInfoFacility {
            shared: Arc::clone(shared),
        });
    let code_management: Arc<dyn crate::native::CodeManagementFacility> =
        Arc::new(super::module_management::SchedulerCodeManagementFacility {
            shared: Arc::clone(shared),
        });
    let system_info: Arc<dyn crate::native::SystemInfoFacility> =
        Arc::new(SchedulerSystemInfoFacility {
            shared: Arc::clone(shared),
        });
    let ets_facility: Arc<dyn crate::native::EtsFacility> = Arc::new(SchedulerEtsFacility {
        shared: Arc::clone(shared),
    });
    let pg_facility: Arc<dyn crate::distribution::pg::PgFacility> =
        Arc::clone(&shared.pg_registry) as _;
    // The file facility exists only when the file ring is present (spec §3.3).
    // A Disabled file ring yields no facility, so `submit_file_io` refuses at
    // the BIF surface (badarg) BEFORE any suspension is registered — the
    // refusal-first discipline, held here by simply not offering the facility.
    // §3.3's embedder-typed refusal (Q-B's ServiceUnavailable half) lands with
    // live-Disabled construction in commit 5: the only Disabled file ring
    // today is replay, which never runs file natives live.
    let file_io_facility: Option<Arc<dyn FileIoFacility>> =
        shared.file_io_ring.service().map(|ring| {
            Arc::new(SchedulerFileIoFacility {
                shared: Arc::clone(shared),
                ring: Arc::clone(ring.ring()),
            }) as Arc<dyn FileIoFacility>
        });
    // Distribution-facing facilities are offered ONLY when the bundle is Owned
    // (spec §3.6): with distribution Disabled the net-kernel, remote-send, and
    // control surfaces are absent, so the BIFs take their standalone-node arms
    // (node ⇒ nonode@nohost, nodes ⇒ [], connect/disconnect ⇒ false, remote
    // send ⇒ noconnection) instead of touching an absent runtime.
    let distribution_owned = shared.distribution().is_some();
    let net_kernel = shared
        .distribution()
        .map(|dist| Arc::clone(dist.net_kernel()));
    let distribution_send: Option<Arc<dyn DistributionSendFacility>> =
        distribution_owned.then(|| {
            Arc::new(SchedulerDistributionSendFacility {
                shared: Arc::clone(shared),
            }) as Arc<dyn DistributionSendFacility>
        });
    let distribution_control_facility: Option<Arc<dyn DistributionControlFacility>> =
        distribution_owned.then(|| {
            Arc::new(SchedulerDistributionControlFacility {
                shared: Arc::clone(shared),
            }) as Arc<dyn DistributionControlFacility>
        });
    let local_send: Arc<dyn crate::native::local_send::LocalSendFacility> =
        Arc::new(SchedulerLocalSendFacility {
            shared: Arc::clone(shared),
        });
    crate::interpreter::NativeServices {
        atom_table: Some(Arc::clone(&shared.atom_table)),
        local_node: Some(shared.local_node),
        net_kernel,
        distribution_send,
        local_send: Some(local_send),
        ets_facility: Some(ets_facility),
        pg_facility: Some(pg_facility),
        timers: Some(Arc::clone(&shared.timers)),
        spawn_facility: Some(spawn),
        link_facility: Some(link),
        distribution_control_facility,
        group_leader_facility: Some(group_leader),
        supervision_facility: Some(supervision),
        process_info_facility: Some(process_info),
        io_sink: Some(Arc::clone(&lock_or_recover(&shared.output_sink))),
        code_management_facility: Some(code_management),
        system_info_facility: Some(system_info),
        io_facility: if shared.replay_mode {
            None
        } else {
            shared.io_facility.clone()
        },
        io_message_facility: Some(Arc::new(SchedulerIoMessageFacility {
            shared: Arc::clone(shared),
        })),
        teardown_admission_facility: Some(Arc::new(SchedulerTeardownAdmissionFacility {
            shared: Arc::clone(shared),
        })),
        #[cfg(feature = "readiness")]
        readiness_facility: shared.readiness_consumer.as_ref().map(|_| {
            Arc::new(SchedulerReadinessFacility {
                shared: Arc::clone(shared),
            }) as Arc<dyn crate::native::ReadinessFacility>
        }),
        // Both surfaces ride the file ring, so both are absent when it is
        // Disabled (spec §3.3) — `file_io_facility` is already gated on the
        // ring above; the TCP surface follows the same condition.
        file_io_facility,
        tcp_io_facility: shared.file_io_ring.service().is_some().then(|| {
            Arc::new(SchedulerTcpIoFacility {
                shared: Arc::clone(shared),
            }) as Arc<dyn crate::native::TcpIoFacility>
        }),
        jit_cache: Some(Arc::clone(&shared.jit_cache)),
        // The profiling handle composes AWAY under replay (reduction counts
        // are validated there and JIT charges differ, commit-5 precedent) and
        // wherever the dirty-CPU service is Disabled: absence at the call
        // edges is the disable mechanism, never a runtime flag check.
        jit_profiling: (!shared.replay_mode && shared.dirty_cpu.service().is_some()).then(|| {
            Arc::new(crate::jit::JitProfilingServices {
                profiler: Arc::clone(&shared.jit_profiler),
                submitter: Arc::new(SchedulerJitSubmissionFacility {
                    shared: Arc::clone(shared),
                }),
            })
        }),
        replay_driver: shared.replay_driver.clone(),
        bif_registry: Some(Arc::clone(&shared.bif_registry)),
        nif_private_data: shared.nif_private_data.clone(),
        suspension_registrar: Some(Arc::new(
            crate::scheduler::suspension::SchedulerSuspensionRegistrar {
                shared: Arc::clone(shared),
            },
        )),
        ..crate::interpreter::NativeServices::default()
    }
}

// ── Facility implementations ────────────────────────────────────────────────

struct SchedulerJitSubmissionFacility {
    shared: Arc<SharedState>,
}

impl crate::jit::JitSubmissionFacility for SchedulerJitSubmissionFacility {
    fn submit(
        &self,
        request: crate::jit::CompilationRequest,
    ) -> Result<(), super::dirty::DirtySubmitError> {
        let job = crate::jit::CompilationJob::new(
            request,
            Arc::clone(&self.shared.jit_compiler),
            Arc::clone(&self.shared.jit_profiler),
            Arc::clone(&self.shared.jit_cache),
        );
        crate::jit::try_submit_jit_compilation(&self.shared.dirty_cpu, job)
    }
}

struct SchedulerTeardownAdmissionFacility {
    shared: Arc<SharedState>,
}

impl crate::native::TeardownAdmissionFacility for SchedulerTeardownAdmissionFacility {
    fn try_reserve(&self) -> Option<Box<dyn Send>> {
        self.shared
            .try_reserve_teardown_admission()
            .map(|admission| Box::new(admission) as Box<dyn Send>)
    }
}

#[cfg(feature = "readiness")]
struct SchedulerReadinessFacility {
    shared: Arc<SharedState>,
}

#[cfg(feature = "readiness")]
impl crate::native::ReadinessFacility for SchedulerReadinessFacility {
    fn register(
        &self,
        fd: std::os::fd::RawFd,
        interest: crate::scheduler::Interest,
        pid: u64,
        marker: crate::atom::Atom,
    ) -> Result<crate::scheduler::ReadinessToken, crate::scheduler::ReadinessError> {
        self.shared.readiness_register(fd, interest, pid, marker)
    }

    fn rearm(
        &self,
        token: &crate::scheduler::ReadinessToken,
        interest: crate::scheduler::Interest,
    ) -> Result<(), crate::scheduler::ReadinessError> {
        self.shared.readiness_rearm(token, interest)
    }
}

pub(super) struct SchedulerDistributionSendFacility {
    pub(super) shared: Arc<SharedState>,
}

impl DistributionSendFacility for SchedulerDistributionSendFacility {
    fn send_remote(&self, target: Term, message: Term) -> Result<(), DistributionSendError> {
        let pid = PidRef::new(target).ok_or(DistributionSendError::Encode)?;
        let node = pid.node().ok_or(DistributionSendError::Encode)?;
        let node_name = self
            .shared
            .atom_table
            .resolve(node)
            .ok_or(DistributionSendError::NoConnection)?
            .to_owned();
        let frame = encode_send_frame(
            Term::atom(Atom::OK),
            target,
            message,
            &self.shared.atom_table,
        )
        .map_err(|_| DistributionSendError::Encode)?;
        // This facility is offered only when the bundle is Owned, but route the
        // manager access through it so absence is `noconnection`, never a panic.
        let dist = self
            .shared
            .distribution()
            .ok_or(DistributionSendError::NoConnection)?;
        block_on_distribution_send(dist.connections(), node, &node_name, &frame)
    }
}

pub(super) fn block_on_distribution_send(
    manager: &crate::distribution::connection::ConnectionManager,
    node: Atom,
    node_name: &str,
    frame: &[u8],
) -> Result<(), DistributionSendError> {
    block_on_distribution_send_with_write_deadline(
        manager,
        node,
        node_name,
        frame,
        crate::distribution::sender::WRITE_TIMEOUT,
    )
}

pub(super) fn block_on_distribution_send_with_write_deadline(
    manager: &crate::distribution::connection::ConnectionManager,
    node: Atom,
    node_name: &str,
    frame: &[u8],
    write_deadline: std::time::Duration,
) -> Result<(), DistributionSendError> {
    let manager = manager.clone();
    let node_name = node_name.to_owned();
    let frame = frame.to_vec();
    let future = async move {
        let connection = match manager.get_connection(node) {
            Some(connection) => connection,
            None => manager
                .connect(&node_name)
                .await
                .map_err(|_| DistributionSendError::NoConnection)?,
        };
        // The write is BOUNDED with the same deadline the sender drain uses:
        // this path blocks a scheduler thread synchronously, and a peer that
        // is TCP-connected but never reads would otherwise park it
        // indefinitely — no other machinery can end the wait (the heartbeat's
        // own writes queue behind this same writer mutex, and inbound
        // keepalives keep inbound liveness fresh while outbound is wedged).
        // On elapse the connection is retired through the same
        // mark_down_write_timeout path the drain uses, whose socket shutdown
        // also unblocks the erroring write future being dropped here.
        match tokio::time::timeout(write_deadline, connection.write_raw(&frame)).await {
            Ok(Ok(())) => Ok(()),
            Ok(Err(_)) => Err(DistributionSendError::NoConnection),
            Err(_elapsed) => {
                connection.mark_down_write_timeout();
                Err(DistributionSendError::NoConnection)
            }
        }
    };
    if let Ok(handle) = tokio::runtime::Handle::try_current() {
        if matches!(
            handle.runtime_flavor(),
            tokio::runtime::RuntimeFlavor::MultiThread
        ) {
            tokio::task::block_in_place(|| handle.block_on(future))
        } else {
            std::thread::spawn(move || {
                tokio::runtime::Builder::new_current_thread()
                    .enable_io()
                    .enable_time()
                    .build()
                    .map_err(|_| DistributionSendError::NoConnection)?
                    .block_on(future)
            })
            .join()
            .map_err(|_| DistributionSendError::NoConnection)?
        }
    } else {
        tokio::runtime::Builder::new_current_thread()
            .enable_io()
            .enable_time()
            .build()
            .map_err(|_| DistributionSendError::NoConnection)?
            .block_on(future)
    }
}

impl ControlDelivery for SchedulerDistributionSendFacility {
    fn deliver_payload(&self, target_pid: u64, payload_etf: &[u8]) -> bool {
        let Some(entry) = self.shared.process_bodies.get(&target_pid) else {
            return false;
        };
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(process) => {
                let mut context = crate::native::ProcessContext::new();
                context.attach_process(&mut process.0, 0);
                let Ok(message) = crate::etf::decode::decode_term(
                    payload_etf,
                    &mut context,
                    &self.shared.atom_table,
                ) else {
                    return false;
                };
                process.0.mailbox_mut().push_owned(message);
            }
            ProcessSlot::Executing(metadata) => {
                metadata
                    .pending_distribution_payloads
                    .push(payload_etf.to_vec());
            }
            ProcessSlot::Absent => return false,
        }
        drop(slot);
        drop(entry);
        wake_process(&self.shared, target_pid);
        true
    }
}

impl ControlRegistry for SchedulerDistributionSendFacility {
    fn whereis(&self, name: Atom) -> Option<u64> {
        self.shared.process_registry.get(&name).map(|entry| *entry)
    }
}

impl crate::distribution::control::PgDelivery for SchedulerDistributionSendFacility {
    fn apply_pg_join(&self, scope: Atom, group: Atom, node: Atom, pid_number: u64, serial: u64) {
        self.shared
            .pg_registry
            .apply_remote_join(scope, group, node, pid_number, serial);
    }

    fn apply_pg_leave(&self, scope: Atom, group: Atom, node: Atom, pid_number: u64, serial: u64) {
        self.shared
            .pg_registry
            .apply_remote_leave(scope, group, node, pid_number, serial);
    }
}

/// Scheduler-side implementation of [`LocalSendFacility`]: delivers a local
/// message to a target process body held in `process_bodies`, mirroring the
/// I/O delivery template (lock slot → Present/Executing/Absent → push before
/// wake) and the distribution-payload deferral for the Executing case.
struct SchedulerLocalSendFacility {
    shared: Arc<SharedState>,
}

impl crate::native::local_send::LocalSendFacility for SchedulerLocalSendFacility {
    fn send_local(
        &self,
        request: crate::native::local_send::LocalSendRequest<'_>,
    ) -> Result<(), crate::native::local_send::LocalSendError> {
        // Self-send from the BYTECODE path is handled in-hand by
        // `messaging::send` against the sender's own body and never reaches the
        // facility. A NATIVE self-send (NATIVE-001 R7), by contrast, is routed
        // here deliberately: the sender's slot is `Executing` during its slice,
        // so the message lands in `pending_local_messages` and is merged into
        // the mailbox at store-back (visible on the next slice) — exactly the
        // existing Executing-slot deferral, with no native-specific code path.
        let Some(entry) = self.shared.process_bodies.get(&request.target_pid) else {
            // Absent/dead pid: silent drop, matching BEAM semantics.
            return Ok(());
        };
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(process) => {
                // The Present branch is the ONLY place replay can occur for a
                // cross-process send: on a single replay thread no other pid can
                // be Executing, so the receiver is always Present (or Absent).
                let previous_receiver_clock = process.0.logical_clock();
                let receiver_clock = process.0.observe_message_clock(request.sender_clock);
                if let Some(driver) = request.replay_driver {
                    let mut guard = match driver.lock() {
                        Ok(guard) => guard,
                        Err(error) => error.into_inner(),
                    };
                    let recorded = match guard.next_message_delivery(
                        crate::replay::RecordedDeliveryKind::Message,
                        Some(request.sender_pid),
                        request.target_pid,
                        request.message,
                    ) {
                        Ok(recorded) => recorded,
                        Err(error) => {
                            process.0.set_logical_clock(previous_receiver_clock);
                            return Err(crate::native::local_send::LocalSendError::ReplayMismatch(
                                error.to_string(),
                            ));
                        }
                    };
                    if recorded.sender_clock != request.sender_clock
                        || recorded.receiver_clock != receiver_clock
                    {
                        process.0.set_logical_clock(previous_receiver_clock);
                        return Err(crate::native::local_send::LocalSendError::ReplayMismatch(
                            format!(
                                "message delivery clock mismatch: expected sender/receiver clocks ({}, {}), recorded ({}, {})",
                                request.sender_clock,
                                receiver_clock,
                                recorded.sender_clock,
                                recorded.receiver_clock
                            ),
                        ));
                    }
                }
                #[cfg(feature = "telemetry")]
                {
                    if process
                        .0
                        .mailbox()
                        .sender()
                        .send_traced(
                            request.sender_pid,
                            request.target_pid,
                            request.message,
                            process.0.heap_mut(),
                        )
                        .is_err()
                    {
                        // BEAM `!` cannot fail, so we still return Ok and drop the
                        // message — but a copy-into-mailbox failure (e.g. the
                        // receiver's young heap is full; beamr's bump allocator
                        // returns HeapFull rather than GCing) should never be
                        // silent. Surface it via telemetry. NOT a debug_assert:
                        // HeapFull is a legitimate runtime condition, not an
                        // invariant violation, so it must not crash debug builds.
                        crate::telemetry::metrics::record_message_dropped("mailbox_present");
                        return Ok(());
                    }
                }
                #[cfg(not(feature = "telemetry"))]
                {
                    if process
                        .0
                        .mailbox()
                        .sender()
                        .send(request.message, process.0.heap_mut())
                        .is_err()
                    {
                        // See the telemetry branch above: keep `!` infallible and
                        // drop on copy failure (HeapFull). No telemetry symbol in a
                        // non-telemetry build; no debug_assert because HeapFull is a
                        // legitimate runtime condition. Matches the I/O delivery
                        // template's silent-drop posture.
                        return Ok(());
                    }
                }
                // The scheduler's `wake_process` (below) requeues a parked
                // receiver and the slice machinery flips its status on resume —
                // exactly as the I/O delivery template does. The facility must
                // NOT force the status transition itself (doing so leaves a
                // parked process in an inconsistent Running state).
            }
            ProcessSlot::Executing(metadata) => {
                // Live-mode-only path: the receiver is mid-slice on another
                // thread, so we cannot touch its heap. ETF-encode the message
                // here so it survives the heap crossing, then decode it onto the
                // receiver heap at store-back (see scheduler/execution/core.rs).
                //
                // Clock note: delivering to an Executing receiver does NOT advance
                // the receiver's Lamport clock here. That is intentional and
                // consistent with the I/O and distribution deferred-delivery paths
                // (pending_io_messages / pending_distribution_payloads), which also
                // defer the receiver-side clock work. It is replay-safe because
                // replay is single-threaded: a receiver is never Executing during
                // replay, so this arm is only ever taken in live mode and never
                // contributes to the recorded delivery ordering.
                let payload =
                    match crate::etf::encode::encode_term(request.message, &self.shared.atom_table)
                    {
                        Ok(payload) => payload,
                        Err(_error) => {
                            // BEAM `!` cannot fail, so we keep returning Ok and drop
                            // the message rather than erroring the send. Surface the
                            // drop via telemetry so a codec gap is never an invisible
                            // message loss and is countable in prod.
                            //
                            // KNOWN LIMITATION (follow-up): `encode_term` rejects
                            // free-variable closures (`num_free != 0`), so a closure
                            // capturing variables sent to an *Executing* receiver is
                            // dropped here, even though the Present/in-hand path
                            // (`copy_term`) would deliver it faithfully. This is a
                            // pre-existing ETF asymmetry, not introduced by this fix.
                            // It is NOT a debug_assert: a free-var-closure send is
                            // legitimate user code, not an invariant violation, so it
                            // must not crash debug builds. Closing the gap requires
                            // symmetric closure ETF (or a heap-fragment copy for the
                            // Executing case) — tracked separately. References, the
                            // common actor case, DO round-trip (see etf/decode.rs).
                            #[cfg(feature = "telemetry")]
                            crate::telemetry::metrics::record_message_dropped("etf_encode");
                            return Ok(());
                        }
                    };
                metadata.pending_local_messages.push(payload);
            }
            ProcessSlot::Absent => return Ok(()),
        }
        drop(slot);
        drop(entry);
        wake_process(&self.shared, request.target_pid);
        Ok(())
    }
}

struct SchedulerIoMessageFacility {
    shared: Arc<SharedState>,
}

impl IoMessageFacility for SchedulerIoMessageFacility {
    fn send_message(&self, sender_pid: u64, target_pid: u64, message: Term) -> bool {
        let _ = sender_pid;
        // Teardown-admission (spec §4 step 3): no delivery into a scheduler
        // that is being (or has been) torn down; held across the delivery so
        // an admitted send lands before shutdown returns.
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return false;
        };
        let Some(entry) = self.shared.process_bodies.get(&target_pid) else {
            return false;
        };
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(process) => {
                process.0.mailbox_mut().push_owned(message);
            }
            ProcessSlot::Executing(metadata) => {
                metadata
                    .pending_io_messages
                    .push(super::PendingMailboxMessage::TargetOwned(message));
            }
            ProcessSlot::Absent => return false,
        }
        drop(slot);
        drop(entry);
        wake_process(&self.shared, target_pid);
        true
    }
}

struct SchedulerFileIoFacility {
    shared: Arc<SharedState>,
    /// The file ring captured at build time. The facility is constructed only
    /// when the file ring is present (spec §3.3), so this is the live Owned or
    /// Shared ring — never a Disabled slot.
    ring: Arc<dyn CompletionRing>,
}

struct SchedulerEtsFacility {
    shared: Arc<SharedState>,
}

impl EtsFacility for SchedulerEtsFacility {
    fn create_table(&self, metadata: EtsTableMetadata) -> Result<EtsTableId, EtsError> {
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return Err(EtsError::Badarg);
        };
        self.shared.ets_registry.try_create_table(metadata)
    }

    fn lookup_table(&self, id: EtsTableId) -> Option<Arc<dyn EtsTable>> {
        self.shared.ets_registry.lookup_table(id)
    }

    fn lookup_named_table(&self, name: Atom) -> Option<Arc<dyn EtsTable>> {
        self.shared.ets_registry.lookup_named_table(name)
    }

    fn lookup_table_by_name(&self, name: Atom) -> Option<EtsTableId> {
        self.shared.ets_registry.lookup_table_by_name(name)
    }

    fn delete_table(&self, id: EtsTableId) -> bool {
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return false;
        };
        self.shared.ets_registry.delete_table(id)
    }

    fn give_away_table(
        &self,
        table_id: EtsTableId,
        new_owner: u64,
        from_pid: u64,
        gift_data: Term,
        atom_table: &crate::atom::AtomTable,
    ) -> Result<(), EtsError> {
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return Err(EtsError::Badarg);
        };
        if !deliver_ets_transfer(
            &self.shared,
            new_owner,
            table_id,
            from_pid,
            gift_data,
            atom_table,
        ) {
            return Err(EtsError::Badarg);
        }
        if self
            .shared
            .ets_registry
            .transfer_table_owner(table_id, new_owner)
        {
            Ok(())
        } else {
            Err(EtsError::Badarg)
        }
    }
}

impl FileIoFacility for SchedulerFileIoFacility {
    fn submit_file_io(&self, pid: u64, op: IoOp, continuation: FileIoContinuation) -> u64 {
        let op_id = self.ring.submit(op);
        self.track_submitted_file_io(pid, op_id, continuation);
        op_id
    }

    fn track_submitted_file_io(&self, pid: u64, op_id: u64, continuation: FileIoContinuation) {
        if let Some((_, completion)) = self.shared.file_io_orphans.remove(&op_id) {
            self.shared.file_io_results.insert(
                pid,
                FileIoCompletion {
                    op_id,
                    continuation,
                    completion,
                },
            );
            super::execution::wake_process(&self.shared, pid);
        } else {
            self.shared
                .file_io_pending
                .insert(op_id, (pid, continuation));
        }
    }

    fn take_file_io_completion(&self, pid: u64) -> Option<FileIoCompletion> {
        self.shared
            .file_io_results
            .remove(&pid)
            .map(|(_, result)| result)
    }

    fn cancel_pending_file_io_for_pid(&self, pid: u64) {
        let op_ids: Vec<u64> = self
            .shared
            .file_io_pending
            .iter()
            .filter_map(|entry| (entry.value().0 == pid).then_some(*entry.key()))
            .collect();
        for op_id in op_ids {
            if self.shared.file_io_pending.remove(&op_id).is_some() {
                self.shared.file_io_canceled.insert(op_id);
            }
        }
        self.shared.file_io_results.remove(&pid);
    }

    fn ring(&self) -> &dyn CompletionRing {
        self.ring.as_ref()
    }
}

struct SchedulerTcpIoFacility {
    shared: Arc<SharedState>,
}

impl crate::native::TcpIoFacility for SchedulerTcpIoFacility {
    fn submit_active_tcp_read(
        &self,
        socket: Arc<crate::io::resource::FdInner>,
        buf_len: usize,
    ) -> Option<u64> {
        let op_id = self.shared.submit_file_ring_op(IoOp::Read {
            fd: socket.fd(),
            buf_len,
            offset: u64::MAX,
        })?;
        self.shared.file_io_pending.insert(
            op_id,
            (
                socket.controlling_process(),
                crate::native::FileIoContinuation::TcpActiveRecv { fd: socket },
            ),
        );
        Some(op_id)
    }
}

/// Real `ProcessInfoFacility` backed by the scheduler's shared state.
pub(super) struct SchedulerProcessInfoFacility {
    pub(super) shared: Arc<SharedState>,
}

impl crate::native::ProcessInfoFacility for SchedulerProcessInfoFacility {
    fn process_info(
        &self,
        pid: u64,
        item: crate::native::ProcessInfoItem,
    ) -> Option<crate::native::ProcessInfoValue> {
        self.shared.process_info(pid, item)
    }
}

/// Real `SpawnFacility` backed by the scheduler's shared state.
/// Test-only control for the admitted-spawn/shutdown interleaving pin:
/// holds the `Arc::as_ptr` of the ONE `SharedState` whose spawns should park
/// at the admission gate (0 = disabled), so a parallel test's spawn on a
/// different scheduler can neither park nor satisfy the acknowledgment.
#[cfg(test)]
pub(super) static SPAWN_HOLD_TARGET: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
pub(super) static SPAWN_HELD_AT_GATE: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

pub(super) struct SchedulerSpawnFacility {
    pub(super) shared: Arc<SharedState>,
    pub(super) namespace_id: NamespaceId,
}

pub(super) struct SchedulerSystemInfoFacility {
    pub(super) shared: Arc<SharedState>,
}

impl crate::native::SystemInfoFacility for SchedulerSystemInfoFacility {
    fn scheduler_count(&self) -> usize {
        self.shared.scheduler_count()
    }

    fn process_count(&self) -> usize {
        self.shared.process_count()
    }

    fn atom_count(&self) -> usize {
        self.shared.atom_count()
    }

    fn atom_limit(&self) -> usize {
        self.shared.atom_table.limit()
    }

    fn memory_summary(&self) -> crate::native::system_info_bifs::MemorySummary {
        self.shared.memory_summary()
    }
}

impl SpawnFacility for SchedulerSpawnFacility {
    fn spawn(
        &self,
        caller_pid: u64,
        module: Atom,
        function: Atom,
        args: Vec<Term>,
        link_to: Option<u64>,
    ) -> Result<u64, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        self.admission_test_gate();
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = self.caller_capabilities(caller_pid);
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let arity = u8::try_from(args.len()).map_err(|_| SpawnError::UnresolvedMfa)?;
        let entry = registry
            .lookup_mfa(module, function, arity)
            .map_err(|_| SpawnError::UnresolvedMfa)?;
        let ip = entry
            .module
            .label_ip(entry.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);

        let mut child = super::spawning::build_process(super::spawning::SpawnRequest {
            pid: child_pid,
            module: entry.module.name,
            module_version: Arc::clone(&entry.module),
            instruction_pointer: ip,
            args,
            namespace_id,
            group_leader,
            capabilities,
            priority: Priority::Normal,
            heap_size: DEFAULT_HEAP_SIZE,
            parent_pid: caller_pid,
            function,
            arity,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        });

        if let Some(parent_pid) = link_to {
            let child_linked = child.add_link(parent_pid);
            let parent_linked = add_link_to_slot(&self.shared, parent_pid, child_pid);
            if child_linked && parent_linked {
                #[cfg(feature = "telemetry")]
                crate::telemetry::lifecycle::record_process_linked(parent_pid, child_pid);
            }
        }

        self.shared.process_bodies.insert(
            child_pid,
            std::sync::Mutex::new(ProcessSlot::Present(ScheduledProcess(child))),
        );

        // Enqueue to a scheduler thread's inject queue by notifying.
        self.shared
            .spawn_counter
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        // Put on process table directly (already done above), and wake a thread.
        // NOTE: The child is already in process_bodies; the scheduler loop will
        // find it when it picks up the PID. We need to put the PID in a run queue.
        // Since we don't have direct access to inject queues from here, we put
        // the pid into the wait set as woken so it gets picked up.
        {
            let mut ws = lock_or_recover(&self.shared.wait_set);
            ws.woken.push((child_pid, 0));
        }
        self.shared.wake_condvar.notify_all();

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            entry.module.name,
            function,
            arity,
        );

        Ok(child_pid)
    }

    fn spawn_native(
        &self,
        caller_pid: u64,
        factory: crate::native::native_process::NativeHandlerFactory,
        link_to: Option<u64>,
    ) -> Result<u64, SpawnError> {
        // Existing spelling: spawn a native process that does NOT trap exits.
        // Delegates to the flag-carrying inherent method with `trap_exit=false`
        // so behaviour is byte-identical to before.
        self.spawn_native_flagged(caller_pid, factory, link_to, false)
    }

    fn spawn_monitor(
        &self,
        caller_pid: u64,
        module: Atom,
        function: Atom,
        args: Vec<Term>,
    ) -> Result<SpawnMonitorResult, SpawnError> {
        self.spawn_mfa_with_monitor(caller_pid, module, function, args)
    }

    fn spawn_lambda(
        &self,
        caller_pid: u64,
        module: Atom,
        lambda_index: u32,
        link_to: Option<u64>,
    ) -> Result<u64, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = self.caller_capabilities(caller_pid);
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let loaded = registry.lookup(module).ok_or(SpawnError::UnresolvedMfa)?;
        let lambda = loaded
            .lambdas
            .get(lambda_index as usize)
            .ok_or(SpawnError::UnresolvedMfa)?;
        let ip = loaded
            .label_ip(lambda.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);

        let mut child = super::spawning::build_process(super::spawning::SpawnRequest {
            pid: child_pid,
            module: loaded.name,
            module_version: Arc::clone(&loaded),
            instruction_pointer: ip,
            args: Vec::new(),
            namespace_id,
            group_leader,
            capabilities,
            priority: Priority::Normal,
            heap_size: DEFAULT_HEAP_SIZE,
            parent_pid: caller_pid,
            function: Atom::NIL,
            arity: 0,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        });

        if let Some(parent_pid) = link_to {
            let child_linked = child.add_link(parent_pid);
            let parent_linked = add_link_to_slot(&self.shared, parent_pid, child_pid);
            if child_linked && parent_linked {
                #[cfg(feature = "telemetry")]
                crate::telemetry::lifecycle::record_process_linked(parent_pid, child_pid);
            }
        }

        self.shared.process_bodies.insert(
            child_pid,
            std::sync::Mutex::new(ProcessSlot::Present(ScheduledProcess(child))),
        );

        self.shared
            .spawn_counter
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        {
            let mut ws = lock_or_recover(&self.shared.wait_set);
            ws.woken.push((child_pid, 0));
        }
        self.shared.wake_condvar.notify_all();

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            loaded.name,
            Atom::NIL,
            0,
        );

        Ok(child_pid)
    }

    fn spawn_lambda_monitor(
        &self,
        caller_pid: u64,
        module: Atom,
        lambda_index: u32,
    ) -> Result<SpawnMonitorResult, SpawnError> {
        self.spawn_lambda_with_monitor(caller_pid, module, lambda_index)
    }

    fn spawn_with_options(
        &self,
        caller_pid: u64,
        module: Atom,
        function: Atom,
        args: Vec<Term>,
        options: SpawnOptions,
    ) -> Result<SpawnOptionsResult, SpawnError> {
        self.spawn_mfa_with_options(caller_pid, module, function, args, options)
    }

    fn spawn_lambda_with_options(
        &self,
        caller_pid: u64,
        module: Atom,
        lambda_index: u32,
        options: SpawnOptions,
    ) -> Result<SpawnOptionsResult, SpawnError> {
        self.spawn_lambda_with_options_impl(caller_pid, module, lambda_index, options)
    }

    fn spawn_closure_link(&self, caller_pid: u64, closure_term: Term) -> Result<u64, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        self.spawn_closure_linked(caller_pid, closure_term)
            .map_err(|_| SpawnError::UnresolvedMfa)
    }
}

impl SchedulerSpawnFacility {
    /// Teardown-admission gate (spec §4 step 3), shared by EVERY spawn entry
    /// point: an in-flight dirty native on an embedder-owned SHARED pool holds
    /// this facility past the owner's shutdown — no spawn path may allocate a
    /// pid or touch process state for a dead scheduler. Returns an RAII token
    /// the caller HOLDS across its whole mutation: admission is linearized
    /// with the shutdown drain (one lock), so an admitted spawn completes
    /// BEFORE shutdown returns and a later one refuses — a snapshot check
    /// would admit a delayed spawn that mutates after shutdown has returned.
    fn admit_or_refuse(&self) -> Result<crate::scheduler::TeardownAdmission, SpawnError> {
        self.shared
            .try_reserve_teardown_admission()
            .ok_or(SpawnError::SchedulerTearingDown)
    }

    /// Test-only barrier immediately after spawn admission: lets a test hold
    /// an ADMITTED spawn while it starts shutdown, proving the drain waits.
    /// Instance-scoped: parks ONLY spawns on the SharedState the test armed,
    /// so parallel tests neither stall here nor satisfy the acknowledgment.
    #[cfg(test)]
    fn admission_test_gate(&self) {
        let me = std::sync::Arc::as_ptr(&self.shared) as usize;
        if SPAWN_HOLD_TARGET.load(std::sync::atomic::Ordering::Acquire) == me {
            SPAWN_HELD_AT_GATE.store(true, std::sync::atomic::Ordering::Release);
            while SPAWN_HOLD_TARGET.load(std::sync::atomic::Ordering::Acquire) == me {
                std::thread::sleep(std::time::Duration::from_millis(2));
            }
        }
    }

    #[cfg(not(test))]
    fn admission_test_gate(&self) {}

    /// Build and make-runnable a native process, setting its `trap_exit` flag
    /// BEFORE it is published to `process_bodies` (i.e. before its first slice
    /// can run). Shared by [`SpawnFacility::spawn_native`] (`trap_exit=false`,
    /// the historical spelling) and the flag-before-runnable entry points
    /// [`Scheduler::spawn_native_trap_exit`](crate::scheduler::Scheduler) and
    /// [`Scheduler::spawn_native_link_trap_exit`](crate::scheduler::Scheduler).
    ///
    /// This is the native mirror of bytecode `spawn_trap_exit`
    /// (`scheduler/spawning.rs`): the flag is set on the `Process` while it is
    /// still owned locally, so it is observably live from the process's birth —
    /// there is no window in which the process exists but is not yet trapping.
    fn spawn_native_flagged(
        &self,
        caller_pid: u64,
        factory: crate::native::native_process::NativeHandlerFactory,
        link_to: Option<u64>,
        trap_exit: bool,
    ) -> Result<u64, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        // Mirror `spawn`, but build a native `Process` (no bytecode setup, no
        // instruction pointer) carrying the handler the factory produces.
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = self.caller_capabilities(caller_pid);

        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);

        let mut child = Process::with_capabilities(child_pid, DEFAULT_HEAP_SIZE, capabilities);
        child.set_group_leader(group_leader);
        child.set_namespace_id(namespace_id);
        child.set_priority(Priority::Normal);
        // Flag-before-runnable: set trap_exit while the process is still owned
        // locally, so it is live before the slot is published below.
        child.set_trap_exit(trap_exit);
        child.set_native_body(crate::native::native_process::NativeBody::new(factory));

        if let Some(parent_pid) = link_to {
            let child_linked = child.add_link(parent_pid);
            let parent_linked = add_link_to_slot(&self.shared, parent_pid, child_pid);
            if child_linked && parent_linked {
                #[cfg(feature = "telemetry")]
                crate::telemetry::lifecycle::record_process_linked(parent_pid, child_pid);
            }
        }

        self.shared.process_bodies.insert(
            child_pid,
            std::sync::Mutex::new(ProcessSlot::Present(ScheduledProcess(child))),
        );

        self.shared
            .spawn_counter
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        {
            let mut ws = lock_or_recover(&self.shared.wait_set);
            ws.woken.push((child_pid, 0));
        }
        self.shared.wake_condvar.notify_all();

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            Atom::NIL,
            Atom::NIL,
            0,
        );

        Ok(child_pid)
    }

    /// Native spawn with `trap_exit` set before the process is runnable.
    ///
    /// The scheduler-facing entry point behind
    /// [`Scheduler::spawn_native_trap_exit`](crate::scheduler::Scheduler) and
    /// its linked sibling: `caller_pid` seeds namespace/group-leader exactly as
    /// [`SpawnFacility::spawn_native`] does, and `link_to` links the child
    /// atomically at spawn.
    pub(in crate::scheduler) fn spawn_native_trap_exit(
        &self,
        caller_pid: u64,
        factory: crate::native::native_process::NativeHandlerFactory,
        link_to: Option<u64>,
    ) -> Result<u64, SpawnError> {
        self.spawn_native_flagged(caller_pid, factory, link_to, true)
    }

    fn spawn_mfa_with_monitor(
        &self,
        caller_pid: u64,
        module: Atom,
        function: Atom,
        args: Vec<Term>,
    ) -> Result<SpawnMonitorResult, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = self.caller_capabilities(caller_pid);
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let arity = u8::try_from(args.len()).map_err(|_| SpawnError::UnresolvedMfa)?;
        let entry = registry
            .lookup_mfa(module, function, arity)
            .map_err(|_| SpawnError::UnresolvedMfa)?;
        let ip = entry
            .module
            .label_ip(entry.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);

        let child = super::spawning::build_process(super::spawning::SpawnRequest {
            pid: child_pid,
            module: entry.module.name,
            module_version: Arc::clone(&entry.module),
            instruction_pointer: ip,
            args,
            namespace_id,
            group_leader,
            capabilities,
            priority: Priority::Normal,
            heap_size: DEFAULT_HEAP_SIZE,
            parent_pid: caller_pid,
            function,
            arity,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        });

        let result = self.register_monitor_insert_and_wake(caller_pid, child_pid, child);
        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            entry.module.name,
            function,
            arity,
        );
        Ok(result)
    }

    fn spawn_mfa_with_options(
        &self,
        caller_pid: u64,
        module: Atom,
        function: Atom,
        args: Vec<Term>,
        options: SpawnOptions,
    ) -> Result<SpawnOptionsResult, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = options
            .capabilities
            .clone()
            .unwrap_or_else(|| self.caller_capabilities(caller_pid));
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let arity = u8::try_from(args.len()).map_err(|_| SpawnError::UnresolvedMfa)?;
        let entry = registry
            .lookup_mfa(module, function, arity)
            .map_err(|_| SpawnError::UnresolvedMfa)?;
        let ip = entry
            .module
            .label_ip(entry.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let request = SpawnRequest {
            pid: self.next_child_pid(),
            module: entry.module.name,
            module_version: Arc::clone(&entry.module),
            instruction_pointer: ip,
            args,
            namespace_id,
            group_leader,
            capabilities,
            priority: options.priority.unwrap_or(Priority::Normal),
            heap_size: options.min_heap_size.unwrap_or(DEFAULT_HEAP_SIZE),
            parent_pid: caller_pid,
            function,
            arity,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        };
        Ok(self.insert_options_child(caller_pid, request, options))
    }

    fn spawn_lambda_with_options_impl(
        &self,
        caller_pid: u64,
        module: Atom,
        lambda_index: u32,
        options: SpawnOptions,
    ) -> Result<SpawnOptionsResult, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = options
            .capabilities
            .clone()
            .unwrap_or_else(|| self.caller_capabilities(caller_pid));
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let loaded = registry.lookup(module).ok_or(SpawnError::UnresolvedMfa)?;
        let lambda = loaded
            .lambdas
            .get(lambda_index as usize)
            .ok_or(SpawnError::UnresolvedMfa)?;
        let ip = loaded
            .label_ip(lambda.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let request = SpawnRequest {
            pid: self.next_child_pid(),
            module: loaded.name,
            module_version: Arc::clone(&loaded),
            instruction_pointer: ip,
            args: Vec::new(),
            namespace_id,
            group_leader,
            capabilities,
            priority: options.priority.unwrap_or(Priority::Normal),
            heap_size: options.min_heap_size.unwrap_or(DEFAULT_HEAP_SIZE),
            parent_pid: caller_pid,
            function: Atom::NIL,
            arity: 0,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        };
        Ok(self.insert_options_child(caller_pid, request, options))
    }

    fn next_child_pid(&self) -> u64 {
        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);
        child_pid
    }

    fn insert_options_child(
        &self,
        caller_pid: u64,
        request: SpawnRequest,
        options: SpawnOptions,
    ) -> SpawnOptionsResult {
        let child_pid = request.pid;
        #[cfg(feature = "telemetry")]
        let parent_pid = request.parent_pid;
        #[cfg(feature = "telemetry")]
        let module = request.module;
        #[cfg(feature = "telemetry")]
        let function = request.function;
        #[cfg(feature = "telemetry")]
        let arity = request.arity;
        let mut child = super::spawning::build_process(request);
        if options.link {
            let child_linked = child.add_link(caller_pid);
            let caller_linked = add_link_to_slot(&self.shared, caller_pid, child_pid);
            if child_linked && caller_linked {
                #[cfg(feature = "telemetry")]
                crate::telemetry::lifecycle::record_process_linked(caller_pid, child_pid);
            }
        }
        if options.monitor {
            let result = self.register_monitor_insert_and_wake(caller_pid, child_pid, child);
            #[cfg(feature = "telemetry")]
            crate::telemetry::lifecycle::record_process_spawned(
                &self.shared.atom_table,
                child_pid,
                parent_pid,
                module,
                function,
                arity,
            );
            SpawnOptionsResult {
                pid: result.pid,
                reference: Some(result.reference),
            }
        } else {
            self.insert_and_wake(child_pid, child);
            #[cfg(feature = "telemetry")]
            crate::telemetry::lifecycle::record_process_spawned(
                &self.shared.atom_table,
                child_pid,
                parent_pid,
                module,
                function,
                arity,
            );
            SpawnOptionsResult {
                pid: child_pid,
                reference: None,
            }
        }
    }

    fn spawn_lambda_with_monitor(
        &self,
        caller_pid: u64,
        module: Atom,
        lambda_index: u32,
    ) -> Result<SpawnMonitorResult, SpawnError> {
        let _admission = self.admit_or_refuse()?;
        let namespace_id = self.caller_namespace(caller_pid);
        let group_leader = self.caller_group_leader(caller_pid);
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let loaded = registry.lookup(module).ok_or(SpawnError::UnresolvedMfa)?;
        let lambda = loaded
            .lambdas
            .get(lambda_index as usize)
            .ok_or(SpawnError::UnresolvedMfa)?;
        let ip = loaded
            .label_ip(lambda.label)
            .map_err(|_| SpawnError::UnresolvedMfa)?;

        let child_pid = self
            .shared
            .next_pid
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.shared.process_table.spawn_with_pid(child_pid);

        let child = super::spawning::build_process(super::spawning::SpawnRequest {
            pid: child_pid,
            module: loaded.name,
            module_version: Arc::clone(&loaded),
            instruction_pointer: ip,
            args: Vec::new(),
            namespace_id,
            group_leader,
            capabilities: self.caller_capabilities(caller_pid),
            priority: Priority::Normal,
            heap_size: DEFAULT_HEAP_SIZE,
            parent_pid: caller_pid,
            function: Atom::NIL,
            arity: 0,
            #[cfg(feature = "telemetry")]
            trace_context: None,
        });

        let result = self.register_monitor_insert_and_wake(caller_pid, child_pid, child);
        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            loaded.name,
            Atom::NIL,
            0,
        );
        Ok(result)
    }

    fn register_monitor_insert_and_wake(
        &self,
        caller_pid: u64,
        child_pid: u64,
        mut child: crate::process::Process,
    ) -> SpawnMonitorResult {
        let reference = {
            let mut ms = lock_or_recover(&self.shared.monitor_set);
            let reference = ms.allocate_reference_pub();
            let mon = crate::process::Monitor::new(reference, caller_pid, child_pid);
            ms.register_monitor(reference, mon, child_pid);
            child.add_monitor(mon);
            drop(ms);
            add_monitor_to_slot(&self.shared, caller_pid, mon);
            reference
        };

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_monitored(caller_pid, child_pid, reference);

        self.insert_and_wake(child_pid, child);

        SpawnMonitorResult {
            pid: child_pid,
            reference,
        }
    }

    fn insert_and_wake(&self, child_pid: u64, child: Process) {
        self.shared.process_bodies.insert(
            child_pid,
            std::sync::Mutex::new(ProcessSlot::Present(ScheduledProcess(child))),
        );
        self.shared
            .spawn_counter
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        {
            let mut ws = lock_or_recover(&self.shared.wait_set);
            ws.woken.push((child_pid, 0));
        }
        self.shared.wake_condvar.notify_all();
    }

    fn caller_namespace(&self, caller_pid: u64) -> NamespaceId {
        if let Some(parent_entry) = self.shared.process_bodies.get(&caller_pid) {
            let parent_slot = lock_or_recover(&parent_entry);
            match &*parent_slot {
                ProcessSlot::Present(ScheduledProcess(parent)) => return parent.namespace_id(),
                ProcessSlot::Executing(metadata) => return metadata.namespace_id,
                ProcessSlot::Absent => {}
            }
        }
        self.namespace_id
    }

    fn caller_group_leader(&self, caller_pid: u64) -> Term {
        if let Some(parent_entry) = self.shared.process_bodies.get(&caller_pid) {
            let parent_slot = lock_or_recover(&parent_entry);
            match &*parent_slot {
                ProcessSlot::Present(ScheduledProcess(parent)) => return parent.group_leader(),
                ProcessSlot::Executing(metadata) => return metadata.group_leader,
                ProcessSlot::Absent => {}
            }
        }
        // No live caller: this is a synthetic/root spawn (e.g. the public
        // `spawn_native` uses caller pid 0). Seed it exactly like a top-level
        // bytecode spawn — process 0 when the standard ring is Owned, the
        // no-such-pid sentinel when Disabled (spec §3.4) — never the absent
        // caller's own pid, which would resurrect the self-leader hang shape.
        match Term::try_pid(self.shared.standard_io_pid) {
            Some(pid_term) => pid_term,
            None => Term::NIL,
        }
    }

    fn caller_capabilities(&self, caller_pid: u64) -> CapabilitySet {
        if let Some(parent_entry) = self.shared.process_bodies.get(&caller_pid) {
            let parent_slot = lock_or_recover(&parent_entry);
            match &*parent_slot {
                ProcessSlot::Present(ScheduledProcess(parent)) => {
                    return parent.capabilities().clone();
                }
                ProcessSlot::Executing(metadata) => return metadata.capabilities.clone(),
                ProcessSlot::Absent => {}
            }
        }
        CapabilitySet::all()
    }

    /// Spawn a linked child that runs a zero-arity closure (thunk), deep-copying
    /// the closure's environment into the child's own heap.
    ///
    /// Unlike the plain spawn paths — whose `args: Vec<Term>` are written into
    /// x-registers WITHOUT a heap copy, so callers must keep any backing heap
    /// alive — this primitive owns the copy: every free variable is copied via
    /// the mailbox copy machinery into the child heap before the child becomes
    /// runnable, so the caller's heap may be collected, mutated, or freed the
    /// moment this returns. On [`crate::process::heap::HeapFull`] the child is
    /// rebuilt with a doubled heap up to [`CLOSURE_SPAWN_MAX_HEAP_WORDS`].
    ///
    /// The closure target resolves through the caller's namespace registry the
    /// same way `call_fun` resolves it (generation match, unique-id fallback,
    /// old-generation fallback); export funs (`fun m:f/0`) resolve through the
    /// module export table. Funs backed by a native (BIF/NIF) entry are not
    /// spawnable — there is no bytecode entry IP for them.
    pub(in crate::scheduler) fn spawn_closure_linked(
        &self,
        caller_pid: u64,
        closure_term: Term,
    ) -> Result<u64, crate::error::ExecError> {
        use crate::error::ExecError;
        use crate::term::boxed::Closure;

        let closure = Closure::new(closure_term).ok_or(ExecError::Badfun { term: closure_term })?;
        if closure.arity() != 0 {
            return Err(ExecError::Badarity {
                fun: closure_term,
                args: Vec::new(),
            });
        }
        if closure.num_free() > 256 {
            return Err(ExecError::InvalidOperand("closure free variables"));
        }
        let namespace_id = self.caller_namespace(caller_pid);
        let registry = namespace_registry(&self.shared, namespace_id)
            .unwrap_or_else(|| Arc::clone(&self.shared.module_registry));
        let target = resolve_thunk_target(&registry, closure, closure_term)?;
        let child_pid = self.next_child_pid();
        let mut child =
            self.build_thunk_child(caller_pid, namespace_id, child_pid, closure, &target)?;

        // Link atomically at spawn time: the child object gets the link before
        // it is inserted, so there is no unlinked window. Mirrors `spawn`'s
        // link handling (a caller that died in the meantime simply yields an
        // unlinked child that exits on its own).
        let child_linked = child.add_link(caller_pid);
        let caller_linked = add_link_to_slot(&self.shared, caller_pid, child_pid);
        if child_linked && caller_linked {
            #[cfg(feature = "telemetry")]
            crate::telemetry::lifecycle::record_process_linked(caller_pid, child_pid);
        }

        self.insert_and_wake(child_pid, child);

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_spawned(
            &self.shared.atom_table,
            child_pid,
            caller_pid,
            target.module.name,
            Atom::NIL,
            0,
        );

        Ok(child_pid)
    }

    /// Build the thunk child process: entry at the resolved lambda/export IP,
    /// free variables deep-copied into the child heap and loaded into
    /// `x0..num_free-1` — exactly where `call_fun` places a zero-arity
    /// closure's environment.
    fn build_thunk_child(
        &self,
        caller_pid: u64,
        namespace_id: NamespaceId,
        child_pid: u64,
        closure: crate::term::boxed::Closure,
        target: &crate::interpreter::opcodes::closures::ResolvedClosureTarget,
    ) -> Result<Process, crate::error::ExecError> {
        use crate::error::ExecError;

        let instruction_pointer = target.module.label_ip(target.label)?;
        let group_leader = self.caller_group_leader(caller_pid);
        let capabilities = self.caller_capabilities(caller_pid);
        let mut heap_size = DEFAULT_HEAP_SIZE;
        loop {
            let mut child = super::spawning::build_process(SpawnRequest {
                pid: child_pid,
                module: target.module_name,
                module_version: Arc::clone(&target.module),
                instruction_pointer,
                args: Vec::new(),
                namespace_id,
                group_leader,
                capabilities: capabilities.clone(),
                priority: Priority::Normal,
                heap_size,
                parent_pid: caller_pid,
                function: Atom::NIL,
                arity: 0,
                #[cfg(feature = "telemetry")]
                trace_context: None,
            });
            match copy_closure_env(closure, &mut child) {
                Ok(()) => return Ok(child),
                Err(crate::mailbox::SendError::HeapFull(full)) => {
                    if heap_size >= CLOSURE_SPAWN_MAX_HEAP_WORDS {
                        return Err(ExecError::from(full));
                    }
                    heap_size = heap_size
                        .saturating_mul(2)
                        .min(CLOSURE_SPAWN_MAX_HEAP_WORDS);
                }
                Err(crate::mailbox::SendError::InvalidBoxedTerm) => {
                    return Err(ExecError::InvalidOperand("closure free variable"));
                }
            }
        }
    }
}

/// Initial-heap cap for a closure-spawned child (words). The environment copy
/// retries with a doubled heap on `HeapFull`; this bounds the doubling so a
/// pathological environment fails the spawn with a typed error instead of
/// exhausting memory. 2^26 words = 512 MiB.
const CLOSURE_SPAWN_MAX_HEAP_WORDS: usize = 1 << 26;

/// Resolve a thunk closure to the module/label its child process starts at.
///
/// Local closures resolve exactly as `call_fun` does (generation match with
/// unique-id validation, then unique-id search of the current and old module
/// generations). Export funs resolve through the export table at arity 0.
fn resolve_thunk_target(
    registry: &Arc<crate::module::ModuleRegistry>,
    closure: crate::term::boxed::Closure,
    closure_term: Term,
) -> Result<crate::interpreter::opcodes::closures::ResolvedClosureTarget, crate::error::ExecError> {
    use crate::error::ExecError;
    use crate::interpreter::opcodes::closures::{ResolvedClosureTarget, resolve_closure_target};

    let module_atom = closure
        .module()
        .ok_or(ExecError::Badfun { term: closure_term })?;
    if closure.is_export() {
        let function = closure
            .export_function()
            .ok_or(ExecError::Badfun { term: closure_term })?;
        let entry = registry.lookup_mfa(module_atom, function, 0)?;
        return Ok(ResolvedClosureTarget {
            module_name: entry.module.name,
            label: entry.label,
            module: entry.module,
        });
    }
    let current = registry
        .lookup(module_atom)
        .ok_or(ExecError::Badfun { term: closure_term })?;
    resolve_closure_target(closure, current.as_ref(), Some(registry), closure_term)
}

/// Deep-copy every free variable of `closure` into `child`'s heap and load the
/// copies into `x0..num_free-1` (a zero-arity closure's environment registers).
fn copy_closure_env(
    closure: crate::term::boxed::Closure,
    child: &mut Process,
) -> Result<(), crate::mailbox::SendError> {
    for index in 0..closure.num_free() {
        let free_var = closure
            .free_var(index)
            .ok_or(crate::mailbox::SendError::InvalidBoxedTerm)?;
        let copied = crate::mailbox::copy_term(free_var, child.heap_mut())?;
        let register =
            u16::try_from(index).map_err(|_| crate::mailbox::SendError::InvalidBoxedTerm)?;
        child.set_x_reg(register, copied);
    }
    Ok(())
}

fn add_monitor_to_slot(shared: &SharedState, pid: u64, monitor: crate::process::Monitor) -> bool {
    let Some(entry) = shared.process_bodies.get(&pid) else {
        return false;
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(process)) => {
            process.add_monitor(monitor);
            true
        }
        ProcessSlot::Executing(metadata) => {
            metadata.add_monitor(monitor);
            true
        }
        ProcessSlot::Absent => false,
    }
}

fn add_link_to_slot(shared: &SharedState, pid: u64, linked_pid: u64) -> bool {
    let Some(entry) = shared.process_bodies.get(&pid) else {
        return false;
    };
    let mut slot = lock_or_recover(&entry);
    match &mut *slot {
        ProcessSlot::Present(ScheduledProcess(process)) => {
            process.add_link(linked_pid);
            true
        }
        ProcessSlot::Executing(metadata) => {
            metadata.add_link(linked_pid, pid);
            true
        }
        ProcessSlot::Absent => false,
    }
}

fn slot_has_link(shared: &SharedState, pid: u64, linked_pid: u64) -> bool {
    let Some(entry) = shared.process_bodies.get(&pid) else {
        return false;
    };
    let slot = lock_or_recover(&entry);
    match &*slot {
        ProcessSlot::Present(ScheduledProcess(process)) => process.links().contains(&linked_pid),
        ProcessSlot::Executing(metadata) => metadata.links.contains(&linked_pid),
        ProcessSlot::Absent => false,
    }
}

fn remove_link_from_slot(shared: &SharedState, pid: u64, linked_pid: u64) {
    if let Some(entry) = shared.process_bodies.get(&pid) {
        let mut slot = lock_or_recover(&entry);
        match &mut *slot {
            ProcessSlot::Present(ScheduledProcess(process)) => {
                process.remove_link(linked_pid);
            }
            ProcessSlot::Executing(metadata) => metadata.remove_link(linked_pid),
            ProcessSlot::Absent => {}
        }
    }
}

/// Real `LinkFacility` backed by the scheduler's shared state.
pub(super) struct SchedulerLinkFacility {
    pub(super) shared: Arc<SharedState>,
}

impl LinkFacility for SchedulerLinkFacility {
    fn link(&self, caller_pid: u64, target_pid: u64) -> Result<(), LinkError> {
        // Teardown-admission (spec §4 step 3), held across the link mutation.
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return Err(LinkError::NoProc);
        };
        if caller_pid == target_pid {
            return Ok(());
        }

        // Check if target is already dead.
        if self.shared.exit_tombstones.contains_key(&target_pid) {
            return Err(LinkError::NoProc);
        }

        // Check target exists in process table.
        if self.shared.process_table.get(target_pid).is_none() {
            return Err(LinkError::NoProc);
        }

        let already_linked = slot_has_link(&self.shared, caller_pid, target_pid);

        // Add link to caller.
        if !add_link_to_slot(&self.shared, caller_pid, target_pid) {
            return Err(LinkError::NoCaller);
        }

        // Add link to target.
        let target_linked = add_link_to_slot(&self.shared, target_pid, caller_pid);

        if !already_linked && target_linked {
            #[cfg(feature = "telemetry")]
            crate::telemetry::lifecycle::record_process_linked(caller_pid, target_pid);
        }

        Ok(())
    }

    fn unlink(&self, caller_pid: u64, target_pid: u64) -> Result<(), LinkError> {
        if caller_pid == target_pid {
            return Ok(());
        }

        remove_link_from_slot(&self.shared, caller_pid, target_pid);

        remove_link_from_slot(&self.shared, target_pid, caller_pid);

        Ok(())
    }

    fn set_trap_exit(&self, caller_pid: u64, value: bool) -> Result<bool, LinkError> {
        let Some(entry) = self.shared.process_bodies.get(&caller_pid) else {
            return Err(LinkError::NoCaller);
        };
        let mut slot = lock_or_recover(&entry);
        let ProcessSlot::Present(ScheduledProcess(process)) = &mut *slot else {
            return Err(LinkError::NoCaller);
        };
        let old = process.trap_exit();
        process.set_trap_exit(value);
        Ok(old)
    }
}

/// Real `DistributionControlFacility` backed by scheduler remote-link metadata.
pub(super) struct SchedulerDistributionControlFacility {
    pub(super) shared: Arc<SharedState>,
}

impl DistributionControlFacility for SchedulerDistributionControlFacility {
    fn link_remote(&self, caller_pid: u64, target: RemotePid) -> Result<(), RemoteLinkError> {
        // Store and send the serial-0 identity: the peer's decode drops the
        // `to` serial and it mints every later EXIT/UNLINK `from` as
        // (node, pid_number, serial 0), so a link stored with a nonzero
        // embedder-supplied serial could never be severed by the wire EXIT
        // equality gate (DC-4) — the death signal would be silently lost
        // until node-down.
        let target = RemotePid {
            serial: 0,
            ..target
        };
        if self.shared.process_table.get(caller_pid).is_none() {
            return Err(RemoteLinkError::BadTarget);
        }
        if !establish_remote_link(&self.shared, caller_pid, target) {
            return Err(RemoteLinkError::BadTarget);
        }
        // ESTABLISH-THEN-SEND order is load-bearing: if the enqueue overflows
        // and the inline down-hook fires on this thread, `connection_down`
        // must observe the just-established link to convert it to
        // noconnection. The connection precondition surfaces as `send_link`'s
        // `NoConnection` (and a pid outside the wire's u32 range as
        // `BadTarget`); on those arms the just-established local half-link is
        // unwound — an unconnected LINK would otherwise be immortal (no
        // connection ⇒ no down event ⇒ no cleanup, ever). If the inline hook
        // already consumed the link, the unwind is a no-op.
        if let Err(error) = dist_control_out::send_link(&self.shared, caller_pid, target) {
            let _ = remove_remote_link(&self.shared, caller_pid, target);
            return Err(error);
        }
        Ok(())
    }

    fn unlink_remote(&self, caller_pid: u64, target: RemotePid) -> Result<(), RemoteLinkError> {
        // Serial-0 identity, mirroring `link_remote`: the stored half-link is
        // always serial 0, so a nonzero embedder serial must not miss it.
        let target = RemotePid {
            serial: 0,
            ..target
        };
        let _ = remove_remote_link(&self.shared, caller_pid, target);
        dist_control_out::send_unlink(&self.shared, caller_pid, target);
        Ok(())
    }

    /// The EXIT2 (`exit/2`) path: best-effort fire-and-forget (ruling 7) —
    /// delivered iff the pinned connection stays up; always `Ok(())`, exactly
    /// as OTP's `exit/2` returns `true` even when undeliverable.
    fn exit_remote(
        &self,
        caller_pid: u64,
        target: RemotePid,
        reason: ExitReason,
    ) -> Result<(), RemoteLinkError> {
        dist_control_out::send_exit2(&self.shared, caller_pid, target, reason);
        Ok(())
    }
}

/// Real `SupervisionFacility` backed by the scheduler's shared state.
pub(super) struct SchedulerSupervisionFacility {
    pub(super) shared: Arc<SharedState>,
}

impl SupervisionFacility for SchedulerSupervisionFacility {
    fn monitor(&self, caller_pid: u64, target_pid: u64) -> Result<MonitorResult, SupervisionError> {
        let mut ms = lock_or_recover(&self.shared.monitor_set);

        // Check if target is already dead.
        if let Some(reason) = self.shared.exit_tombstones.get(&target_pid) {
            let reference = ms.allocate_reference_pub();
            drop(ms);

            // Use the normal-exit admission and wake path. In particular, an
            // executing caller receives the DOWN through pending metadata.
            let immediate_down = deliver_single_down_and_wake(
                &self.shared,
                caller_pid,
                reference,
                target_pid,
                reason,
            );

            return Ok(MonitorResult {
                reference,
                immediate_down,
            });
        }

        // Both processes must exist.
        if self.shared.process_table.get(target_pid).is_none() {
            return Err(SupervisionError::NoProc);
        }

        // Allocate reference and register monitor in monitor_set.
        let reference = ms.allocate_reference_pub();
        let mon = crate::process::Monitor::new(reference, caller_pid, target_pid);
        ms.register_monitor(reference, mon, target_pid);
        drop(ms);

        // Add monitor to caller process.
        if let Some(entry) = self.shared.process_bodies.get(&caller_pid) {
            let mut slot = lock_or_recover(&entry);
            if let ProcessSlot::Present(ScheduledProcess(p)) = &mut *slot {
                p.add_monitor(mon);
            }
        }

        // Add monitor to target process.
        if let Some(entry) = self.shared.process_bodies.get(&target_pid) {
            let mut slot = lock_or_recover(&entry);
            if let ProcessSlot::Present(ScheduledProcess(p)) = &mut *slot {
                p.add_monitor(mon);
            }
        }

        #[cfg(feature = "telemetry")]
        crate::telemetry::lifecycle::record_process_monitored(caller_pid, target_pid, reference);

        Ok(MonitorResult {
            reference,
            immediate_down: false,
        })
    }

    fn demonitor(&self, caller_pid: u64, reference: u64) -> Result<(), SupervisionError> {
        let mut ms = lock_or_recover(&self.shared.monitor_set);

        // Get the monitor info before removing.
        let monitor = ms.get_monitor(reference);
        if let Some(monitor) = monitor {
            // Remove from both processes.
            if let Some(entry) = self.shared.process_bodies.get(&caller_pid) {
                let mut slot = lock_or_recover(&entry);
                if let ProcessSlot::Present(ScheduledProcess(process)) = &mut *slot {
                    process.remove_monitor(reference);
                }
            }
            if let Some(entry) = self.shared.process_bodies.get(&monitor.target()) {
                let mut slot = lock_or_recover(&entry);
                if let ProcessSlot::Present(ScheduledProcess(process)) = &mut *slot {
                    process.remove_monitor(reference);
                }
            }
            ms.remove_monitor(reference);
        }

        Ok(())
    }

    fn exit_signal(
        &self,
        _caller_pid: u64,
        target_pid: u64,
        reason: ExitReason,
    ) -> Result<(), SupervisionError> {
        let Some(_admission) = self.shared.try_reserve_teardown_admission() else {
            return Err(SupervisionError::NoProc);
        };
        // Deliver exit signal to target process.
        if let Some(entry) = self.shared.process_bodies.get(&target_pid) {
            let mut slot = lock_or_recover(&entry);
            match &mut *slot {
                ProcessSlot::Present(ScheduledProcess(target)) => {
                    if matches!(target.status(), ProcessStatus::Exited(_)) {
                        return Ok(());
                    }

                    let should_die = reason == ExitReason::Kill
                        || (reason != ExitReason::Normal && !target.trap_exit());

                    if should_die {
                        let terminal = link::terminal_reason(reason);
                        target.terminate(terminal);
                        drop(slot);
                        drop(entry);
                        cleanup_exited_process(&self.shared, target_pid, terminal);
                    } else if target.trap_exit() {
                        link::enqueue_exit_message_pub(target, _caller_pid, reason);
                        drop(slot);
                        drop(entry);
                        wake_process(&self.shared, target_pid);
                    }
                }
                ProcessSlot::Executing(metadata) => {
                    let should_die = reason == ExitReason::Kill
                        || (reason != ExitReason::Normal && !metadata.trap_exit);
                    if should_die {
                        let terminal = link::terminal_reason(reason);
                        shared_exit_tombstone(&self.shared, target_pid, terminal);
                    } else if metadata.trap_exit {
                        // Process traps exits: queue {EXIT, CallerPid, Reason} for
                        // delivery when the slice completes. This mirrors the Present
                        // arm's `else if target.trap_exit()` and MUST include NORMAL
                        // exits (OTP delivers `{'EXIT', Pid, normal}` for `erlang:exit/2`
                        // with reason `normal` to a trapping process). `should_die` has
                        // already peeled off Kill and abnormal-non-trapping cases, so
                        // reaching here means a trapping target for any non-kill reason.
                        metadata
                            .pending_exit_messages
                            .push((PendingExitSource::Local(_caller_pid), reason));
                        drop(slot);
                        drop(entry);
                        wake_process(&self.shared, target_pid);
                    }
                }
                ProcessSlot::Absent => {}
            }
        }
        Ok(())
    }
}

pub(super) fn shared_exit_tombstone(shared: &SharedState, pid: u64, reason: ExitReason) {
    shared.insert_exit_tombstone(pid, reason);
    let _deleted_tables = shared.transfer_or_delete_tables_owned_by(pid);
    let mut ls = lock_or_recover(&shared.link_set);
    ls.process_exited_tombstone(pid, reason);
}