beamr 0.19.1

//! Distribution control message framing, SEND/REG_SEND handling, and remote
//! spawn helpers (SPAWN_REQUEST/SPAWN_REPLY).

use std::fmt;

use crate::atom::{Atom, AtomTable};
use crate::distribution::control_link;
use crate::distribution::pg::PgUpdate;
use crate::etf::decode::DecodeError;
use crate::etf::encode::{EncodeError, encode_term};
use crate::native::ProcessContext;
use crate::native::spawn::{SpawnError, SpawnFacility, SpawnOptions};
use crate::process::{ExitReason, Process, RemotePid};
use crate::term::Term;
use crate::term::boxed::{Cons, Tuple};
use crate::term::pid_ref::PidRef;

/// Distribution control opcode for PID-to-PID send.
pub const SEND: i64 = 2;
/// Distribution control opcode for registered-name send.
pub const REG_SEND: i64 = 6;

/// beamr-private distribution control opcode for process-group membership
/// propagation (`Join`/`Leave`).
///
/// The OTP distribution protocol assigns control opcodes 1..=31 (the highest
/// understood by beamr is `SPAWN_REPLY = 31`, see
/// [`control_link::ControlOp::from_opcode`]). `101` is well above that
/// range and is not used by OTP, so it cannot collide with a standard control
/// message. It is deliberately beamr-private: a stock Erlang/OTP node would
/// never emit it, and a beamr node ignores any opcode it does not recognise,
/// so an unupgraded peer simply drops the frame instead of misinterpreting it.
pub const PG_UPDATE: i64 = 101;

/// Discriminant written as the second element of a `PG_UPDATE` control tuple to
/// mark a join.
const PG_JOIN_TAG: i64 = 1;
/// Discriminant written as the second element of a `PG_UPDATE` control tuple to
/// mark a leave.
const PG_LEAVE_TAG: i64 = 2;

/// Error raised when a remote send cannot be completed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DistributionSendError {
    /// The target node has no usable distribution connection.
    NoConnection,
    /// The target PID or message cannot be encoded for distribution.
    Encode,
}

impl fmt::Display for DistributionSendError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoConnection => formatter.write_str("noconnection"),
            Self::Encode => formatter.write_str("distribution encode failed"),
        }
    }
}

/// Facility used by opcodes and BIFs to send a message to a remote PID.
pub trait DistributionSendFacility: Send + Sync {
    /// Encode and send `message` to `target` on its remote node.
    fn send_remote(&self, target: Term, message: Term) -> Result<(), DistributionSendError>;
}

/// Scheduler-safe delivery target for incoming decoded control messages.
pub trait ControlDelivery: Send + Sync {
    /// Decode `payload_etf` for `target_pid` and enqueue it in the target mailbox.
    fn deliver_payload(&self, target_pid: u64, payload_etf: &[u8]) -> bool;
}

/// Registry lookup used by incoming REG_SEND controls.
pub trait ControlRegistry: Send + Sync {
    /// Resolve a registered atom name to a local pid.
    fn whereis(&self, name: Atom) -> Option<u64>;
}

/// Decoded distribution control message.
///
/// Fields are extracted values rather than raw Terms because the decode
/// process heap is temporary — boxed Terms would become dangling after return.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ControlMessage {
    /// `{2, Cookie, ToPid}` — stores extracted pid number.
    Send { to_pid: u64 },
    /// `{6, FromPid, Cookie, ToName}` — stores extracted name atom.
    RegSend { to_name: Atom },
    /// `{101, 1, Scope, Group, MemberExternalPid}` — a remote process-group join.
    ///
    /// `node`/`pid_number`/`serial` are extracted from the member PID, which is
    /// always encoded as an external PID carrying the originating node's name so
    /// that the receiver records a fully-qualified `RemoteMember` rather than a
    /// node-less local PID.
    PgJoin {
        /// Scope atom the member joined.
        scope: Atom,
        /// Group atom the member joined.
        group: Atom,
        /// Originating node atom (from the member's external PID).
        node: Atom,
        /// Member PID number on the originating node.
        pid_number: u64,
        /// Member PID serial on the originating node.
        serial: u64,
    },
    /// `{101, 2, Scope, Group, MemberExternalPid}` — a remote process-group leave.
    PgLeave {
        /// Scope atom the member left.
        scope: Atom,
        /// Group atom the member left.
        group: Atom,
        /// Originating node atom (from the member's external PID).
        node: Atom,
        /// Member PID number on the originating node.
        pid_number: u64,
        /// Member PID serial on the originating node.
        serial: u64,
    },
    /// `{1, FromExtPid, ToExtPid}` — establish a remote link on local `to_pid`.
    Link {
        /// Remote endpoint requesting the link.
        from: RemotePid,
        /// Local target PID number (extracted from the addressed `to` pid).
        to_pid: u64,
    },
    /// `{4, FromExtPid, ToExtPid}` — remove a remote link from local `to_pid`.
    Unlink {
        /// Remote endpoint removing the link.
        from: RemotePid,
        /// Local target PID number (extracted from the addressed `to` pid).
        to_pid: u64,
    },
    /// `{3, FromExtPid, ToExtPid, ReasonAtom}` — link-exit signal (op 3).
    Exit {
        /// Remote endpoint that died while linked to `to_pid`.
        from: RemotePid,
        /// Local target PID number (extracted from the addressed `to` pid).
        to_pid: u64,
        /// Terminal exit reason carried on the wire.
        reason: ExitReason,
    },
    /// `{8, FromExtPid, ToExtPid, ReasonAtom}` — `exit/2` signal (op 8).
    Exit2 {
        /// Remote caller of `exit/2`.
        from: RemotePid,
        /// Local target PID number (extracted from the addressed `to` pid).
        to_pid: u64,
        /// Raw exit reason carried on the wire (MAY be `kill`).
        reason: ExitReason,
    },
}

/// Scheduler-side sink for inbound process-group membership controls.
///
/// Implemented over the scheduler's `PgRegistry` so that a decoded `PG_UPDATE`
/// frame applies directly to the local registry's remote-member view without a
/// mailbox round-trip.
pub trait PgDelivery: Send + Sync {
    /// Apply a remote join to the local registry.
    fn apply_pg_join(&self, scope: Atom, group: Atom, node: Atom, pid_number: u64, serial: u64);
    /// Apply a remote leave to the local registry.
    fn apply_pg_leave(&self, scope: Atom, group: Atom, node: Atom, pid_number: u64, serial: u64);
}

/// Errors while decoding or handling a distribution control frame.
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum ControlError {
    /// The frame prefix or lengths were invalid.
    InvalidFrame,
    /// ETF decoding failed.
    Decode(DecodeError),
    /// Control tuple shape did not match any understood control operation.
    InvalidControl,
    /// Frame's target pid names another node (R6).
    MisAddressed,
}

impl From<DecodeError> for ControlError {
    fn from(error: DecodeError) -> Self {
        Self::Decode(error)
    }
}

/// Encode a framed SEND control and payload.
pub fn encode_send_frame(
    cookie: Term,
    to_pid: Term,
    message: Term,
    atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
    let mut process = Process::new(0, 32);
    let mut context = ProcessContext::new();
    context.attach_process(&mut process, 0);
    let control = context
        .alloc_tuple(&[Term::small_int(SEND), cookie, to_pid])
        .map_err(|_| EncodeError::UnsupportedTerm)?;
    encode_frame(control, message, atom_table)
}

/// Encode a framed REG_SEND control and payload.
pub fn encode_reg_send_frame(
    from_pid: Term,
    cookie: Term,
    to_name: Atom,
    message: Term,
    atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
    let mut process = Process::new(0, 32);
    let mut context = ProcessContext::new();
    context.attach_process(&mut process, 0);
    let control = context
        .alloc_tuple(&[
            Term::small_int(REG_SEND),
            from_pid,
            cookie,
            Term::atom(to_name),
        ])
        .map_err(|_| EncodeError::UnsupportedTerm)?;
    encode_frame(control, message, atom_table)
}

/// Encode a framed `PG_UPDATE` control with an empty payload.
///
/// The control tuple is `{101, Tag, Scope, Group, MemberExternalPid}` where
/// `Tag` is `1` for a join or `2` for a leave. The member is always encoded as
/// an **external** PID carrying `local_node` as its node component: a plain
/// local PID would decode on the receiver with `node = None` and corrupt the
/// recorded `RemoteMember`. The payload is `[]` (`NIL`) — `PG_UPDATE` carries
/// no message body.
pub fn encode_pg_update_frame(
    update: PgUpdate,
    local_node: Atom,
    atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
    let (tag, scope, group, pid) = match update {
        PgUpdate::Join { scope, group, pid } => (PG_JOIN_TAG, scope, group, pid),
        PgUpdate::Leave { scope, group, pid } => (PG_LEAVE_TAG, scope, group, pid),
    };
    let mut process = Process::new(0, 64);
    let mut context = ProcessContext::new();
    context.attach_process(&mut process, 0);
    // Serial 0: local immediate PIDs have no serial component, and the wire
    // identity for a locally-originated member is (local_node, pid_number, 0).
    let member = context
        .alloc_external_pid(local_node, pid, 0)
        .map_err(|_| EncodeError::UnsupportedTerm)?;
    let control = context
        .alloc_tuple(&[
            Term::small_int(PG_UPDATE),
            Term::small_int(tag),
            Term::atom(scope),
            Term::atom(group),
            member,
        ])
        .map_err(|_| EncodeError::UnsupportedTerm)?;
    encode_frame(control, Term::NIL, atom_table)
}

pub(crate) fn encode_frame(
    control: Term,
    message: Term,
    atom_table: &AtomTable,
) -> Result<Vec<u8>, EncodeError> {
    let control_etf = encode_term(control, atom_table)?;
    let payload_etf = encode_term(message, atom_table)?;
    let control_len = u32::try_from(control_etf.len()).map_err(|_| EncodeError::UnsupportedTerm)?;
    let payload_len = u32::try_from(payload_etf.len()).map_err(|_| EncodeError::UnsupportedTerm)?;
    let mut frame = Vec::with_capacity(8 + control_etf.len() + payload_etf.len());
    frame.extend_from_slice(&control_len.to_be_bytes());
    frame.extend_from_slice(&payload_len.to_be_bytes());
    frame.extend_from_slice(&control_etf);
    frame.extend_from_slice(&payload_etf);
    Ok(frame)
}

/// Split a frame produced by [`encode_send_frame`] or [`encode_reg_send_frame`].
pub fn split_frame(frame: &[u8]) -> Result<(&[u8], &[u8]), ControlError> {
    let header = frame.get(..8).ok_or(ControlError::InvalidFrame)?;
    let control_len = u32::from_be_bytes([header[0], header[1], header[2], header[3]]) as usize;
    let payload_len = u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize;
    let control_start = 8_usize;
    let payload_start = control_start
        .checked_add(control_len)
        .ok_or(ControlError::InvalidFrame)?;
    let end = payload_start
        .checked_add(payload_len)
        .ok_or(ControlError::InvalidFrame)?;
    if end != frame.len() {
        return Err(ControlError::InvalidFrame);
    }
    let control = frame
        .get(control_start..payload_start)
        .ok_or(ControlError::InvalidFrame)?;
    let payload = frame
        .get(payload_start..end)
        .ok_or(ControlError::InvalidFrame)?;
    Ok((control, payload))
}

/// Decode a control ETF term.
///
/// Compatibility wrapper: decodes without a local-node identity, so no
/// [`ControlError::MisAddressed`] validation applies. Full addressed decode
/// lives in [`control_link::decode_control_addressed`].
pub fn decode_control(
    control_etf: &[u8],
    atom_table: &AtomTable,
) -> Result<ControlMessage, ControlError> {
    control_link::decode_control_addressed(control_etf, atom_table, None)
}

/// Decode a `{101, Tag, Scope, Group, MemberExternalPid}` control tuple.
pub(crate) fn decode_pg_update(tuple: &Tuple) -> Result<ControlMessage, ControlError> {
    let tag = tuple
        .get(1)
        .and_then(Term::as_small_int)
        .ok_or(ControlError::InvalidControl)?;
    let scope = tuple
        .get(2)
        .and_then(Term::as_atom)
        .ok_or(ControlError::InvalidControl)?;
    let group = tuple
        .get(3)
        .and_then(Term::as_atom)
        .ok_or(ControlError::InvalidControl)?;
    let member = PidRef::new(tuple.get(4).ok_or(ControlError::InvalidControl)?)
        .ok_or(ControlError::InvalidControl)?;
    // The member MUST be an external PID carrying the originating node. A local
    // PID (node = None) means the sender failed to encode the node and the
    // resulting RemoteMember would be unattributable, so reject it.
    let node = member.node().ok_or(ControlError::InvalidControl)?;
    let pid_number = member.pid_number();
    let serial = member.serial();
    match tag {
        PG_JOIN_TAG => Ok(ControlMessage::PgJoin {
            scope,
            group,
            node,
            pid_number,
            serial,
        }),
        PG_LEAVE_TAG => Ok(ControlMessage::PgLeave {
            scope,
            group,
            node,
            pid_number,
            serial,
        }),
        _ => Err(ControlError::InvalidControl),
    }
}

/// Handle an incoming frame by decoding the control term and delivering the payload.
///
/// Compatibility wrapper over [`control_link::dispatch_frame`] with no link
/// sink and no origin/local-node identity. Behavior delta from the pre-codec
/// version: LINK/UNLINK/EXIT/EXIT2 frames (opcodes 1/3/4/8) now decode and
/// return `Ok(false)` (dropped, no sink) instead of
/// `Err(`[`ControlError::InvalidControl`]`)`.
pub fn handle_frame(
    control_etf: &[u8],
    payload_etf: &[u8],
    atom_table: &AtomTable,
    delivery: &dyn ControlDelivery,
    registry: Option<&dyn ControlRegistry>,
    pg: Option<&dyn PgDelivery>,
) -> Result<bool, ControlError> {
    control_link::dispatch_frame(
        control_etf,
        payload_etf,
        atom_table,
        &control_link::ControlSinks {
            delivery,
            registry,
            pg,
            links: None,
            origin_node: None,
            local_node: None,
        },
    )
}

// ── SPAWN_REQUEST / SPAWN_REPLY ─────────────────────────────────────────────

/// Distribution control opcode for SPAWN_REQUEST.
pub const SPAWN_REQUEST: i64 = 29;
/// Distribution control opcode for SPAWN_REPLY.
pub const SPAWN_REPLY: i64 = 31;

/// Module/function/arguments entry point carried by SPAWN_REQUEST.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RemoteMfa {
    /// Target module atom.
    pub module: Atom,
    /// Target function atom.
    pub function: Atom,
    /// Arguments list.
    pub args: Vec<Term>,
}

/// Parsed SPAWN_REQUEST control message.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SpawnRequest {
    /// Unique request identifier correlating request/reply.
    pub request_id: u64,
    /// The sender PID term.
    pub from: Term,
    /// The group leader PID term.
    pub group_leader: Term,
    /// Module/function/arguments entry point.
    pub mfa: RemoteMfa,
    /// Spawn options (link, monitor).
    pub options: SpawnOptions,
}

/// Parsed SPAWN_REPLY control message.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct SpawnReply {
    /// Request id from the original SPAWN_REQUEST.
    pub request_id: u64,
    /// The newly spawned PID term.
    pub pid: Term,
}

/// Error returned while parsing a distribution control term for spawn operations.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ControlDecodeError {
    /// The control term was not a tuple.
    NotTuple,
    /// The control tuple had the wrong number of elements.
    BadArity,
    /// The opcode was unknown or did not match the expected operation.
    UnknownOp,
    /// The request id was not a non-negative integer.
    BadRequestId,
    /// The MFA entry point was malformed.
    BadMfa,
    /// The spawn option list was malformed.
    BadOptions,
    /// The PID element was not a valid PID term.
    BadPid,
}

/// Error returned by a local SPAWN_REQUEST handler.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum SpawnRequestError {
    /// The control tuple could not be decoded.
    Decode(ControlDecodeError),
    /// No caller PID was set on the process context.
    MissingCallerPid,
    /// The local spawn facility returned an error.
    Spawn(SpawnError),
    /// The spawned PID is out of u32 wire range or allocation failed.
    PidOutOfRange,
}

/// Decode a `{29, ReqId, From, GroupLeader, {M,F,A}, OptList}` control term.
pub fn decode_spawn_request(
    term: Term,
    context: &ProcessContext<'_>,
) -> Result<SpawnRequest, ControlDecodeError> {
    let tuple = Tuple::new(term).ok_or(ControlDecodeError::NotTuple)?;
    if tuple.arity() != 6 {
        return Err(ControlDecodeError::BadArity);
    }
    let op = tuple
        .get(0)
        .and_then(|t| t.as_small_int())
        .ok_or(ControlDecodeError::UnknownOp)?;
    if op != SPAWN_REQUEST {
        return Err(ControlDecodeError::UnknownOp);
    }
    let request_id = parse_non_negative_u64(tuple.get(1).ok_or(ControlDecodeError::BadRequestId)?)
        .ok_or(ControlDecodeError::BadRequestId)?;
    let from = tuple.get(2).ok_or(ControlDecodeError::BadArity)?;
    let group_leader = tuple.get(3).ok_or(ControlDecodeError::BadArity)?;
    let mfa = parse_mfa(tuple.get(4).ok_or(ControlDecodeError::BadMfa)?)?;
    let options =
        parse_remote_spawn_options(tuple.get(5).ok_or(ControlDecodeError::BadOptions)?, context)?;

    Ok(SpawnRequest {
        request_id,
        from,
        group_leader,
        mfa,
        options,
    })
}

/// Decode a `{31, ReqId, Pid}` control term.
pub fn decode_spawn_reply(term: Term) -> Result<SpawnReply, ControlDecodeError> {
    let tuple = Tuple::new(term).ok_or(ControlDecodeError::NotTuple)?;
    if tuple.arity() != 3 {
        return Err(ControlDecodeError::BadArity);
    }
    let op = tuple
        .get(0)
        .and_then(|t| t.as_small_int())
        .ok_or(ControlDecodeError::UnknownOp)?;
    if op != SPAWN_REPLY {
        return Err(ControlDecodeError::UnknownOp);
    }
    let request_id = parse_non_negative_u64(tuple.get(1).ok_or(ControlDecodeError::BadRequestId)?)
        .ok_or(ControlDecodeError::BadRequestId)?;
    let pid = tuple.get(2).ok_or(ControlDecodeError::BadPid)?;
    if PidRef::new(pid).is_none() {
        return Err(ControlDecodeError::BadPid);
    }
    Ok(SpawnReply { request_id, pid })
}

/// Allocate a SPAWN_REQUEST control tuple on `context`'s process heap.
pub fn alloc_spawn_request(
    context: &mut ProcessContext<'_>,
    request: &SpawnRequest,
) -> Result<Term, Term> {
    let args = context.alloc_list(&request.mfa.args)?;
    let mfa = context.alloc_tuple(&[
        Term::atom(request.mfa.module),
        Term::atom(request.mfa.function),
        args,
    ])?;
    // AR-1 site 1. `mfa` is a single boxed tuple with a scope-shaped lifetime,
    // not a run, so it takes `with_rooted` directly rather than the accumulator.
    // `args` above is NOT a crossing: it is an argument to the `alloc_tuple`
    // that could collect, and the sinks root their arguments before reserving.
    context.with_rooted(&[mfa], |context, roots| {
        let opt_list = spawn_options_to_list(context, request.options.clone())?;
        let op = Term::try_small_int(SPAWN_REQUEST).ok_or_else(badarg)?;
        let req_id = Term::try_small_int(i64::try_from(request.request_id).map_err(|_| badarg())?)
            .ok_or_else(badarg)?;
        // Re-read AFTER spawn_options_to_list: it ends in alloc_list, which can
        // collect and forward `mfa`, and the pre-fix body used the stale copy.
        let mfa = context.rooted(roots, 0)?;
        context.alloc_tuple(&[
            op,
            req_id,
            request.from,
            request.group_leader,
            mfa,
            opt_list,
        ])
    })
}

/// Allocate a SPAWN_REPLY control tuple on `context`'s process heap.
pub fn alloc_spawn_reply(
    context: &mut ProcessContext<'_>,
    request_id: u64,
    pid: Term,
) -> Result<Term, Term> {
    let op = Term::try_small_int(SPAWN_REPLY).ok_or_else(badarg)?;
    let req_id =
        Term::try_small_int(i64::try_from(request_id).map_err(|_| badarg())?).ok_or_else(badarg)?;
    context.alloc_tuple(&[op, req_id, pid])
}

/// Handle a decoded SPAWN_REQUEST by spawning locally with link/monitor options
/// applied atomically.
///
/// The current scheduler spawn API is local-caller based; until remote
/// link/monitor metadata is represented in the scheduler, this uses the supplied
/// local service caller PID as the atomic-options owner rather than pretending
/// the external `From` PID is local.
pub fn handle_spawn_request(
    term: Term,
    context: &mut ProcessContext<'_>,
    spawn_facility: &dyn SpawnFacility,
) -> Result<Term, SpawnRequestError> {
    let request = decode_spawn_request(term, context).map_err(SpawnRequestError::Decode)?;
    let caller_pid = context.pid().ok_or(SpawnRequestError::MissingCallerPid)?;
    let result = spawn_facility
        .spawn_with_options(
            caller_pid,
            request.mfa.module,
            request.mfa.function,
            request.mfa.args,
            request.options,
        )
        .map_err(SpawnRequestError::Spawn)?;
    let pid_term = spawn_reply_pid(context, result.pid)?;
    alloc_spawn_reply(context, request.request_id, pid_term)
        .map_err(|_| SpawnRequestError::PidOutOfRange)
}

fn spawn_reply_pid(context: &mut ProcessContext<'_>, pid: u64) -> Result<Term, SpawnRequestError> {
    if let Some(local_node) = context.local_node() {
        let pid_number = u32::try_from(pid).map_err(|_| SpawnRequestError::PidOutOfRange)?;
        return context
            .alloc_external_pid(local_node.name, u64::from(pid_number), 0)
            .map_err(|_| SpawnRequestError::PidOutOfRange);
    }

    Term::try_pid(pid).ok_or(SpawnRequestError::PidOutOfRange)
}

fn parse_mfa(term: Term) -> Result<RemoteMfa, ControlDecodeError> {
    let tuple = Tuple::new(term).ok_or(ControlDecodeError::BadMfa)?;
    if tuple.arity() != 3 {
        return Err(ControlDecodeError::BadMfa);
    }
    let module = tuple
        .get(0)
        .and_then(|t| t.as_atom())
        .ok_or(ControlDecodeError::BadMfa)?;
    let function = tuple
        .get(1)
        .and_then(|t| t.as_atom())
        .ok_or(ControlDecodeError::BadMfa)?;
    let args = spawn_list_to_vec(tuple.get(2).ok_or(ControlDecodeError::BadMfa)?)
        .ok_or(ControlDecodeError::BadMfa)?;
    Ok(RemoteMfa {
        module,
        function,
        args,
    })
}

fn parse_remote_spawn_options(
    term: Term,
    context: &ProcessContext<'_>,
) -> Result<SpawnOptions, ControlDecodeError> {
    let atom_table = context.atom_table().ok_or(ControlDecodeError::BadOptions)?;
    let link_atom = atom_table.intern("link");
    let monitor_atom = atom_table.intern("monitor");
    let mut options = SpawnOptions::default();
    for option in spawn_list_to_vec(term).ok_or(ControlDecodeError::BadOptions)? {
        if option.as_atom() == Some(link_atom) {
            options.link = true;
        } else if option.as_atom() == Some(monitor_atom) {
            options.monitor = true;
        } else {
            return Err(ControlDecodeError::BadOptions);
        }
    }
    Ok(options)
}

fn spawn_options_to_list(
    context: &mut ProcessContext<'_>,
    options: SpawnOptions,
) -> Result<Term, Term> {
    let atom_table = context.atom_table().ok_or_else(badarg)?;
    let mut elements = Vec::new();
    if options.link {
        elements.push(Term::atom(atom_table.intern("link")));
    }
    if options.monitor {
        elements.push(Term::atom(atom_table.intern("monitor")));
    }
    context.alloc_list(&elements)
}

fn spawn_list_to_vec(term: Term) -> Option<Vec<Term>> {
    let mut elements = Vec::new();
    let mut current = term;
    loop {
        if current.is_nil() {
            return Some(elements);
        }
        let cons = Cons::new(current)?;
        elements.push(cons.head());
        current = cons.tail();
    }
}

fn parse_non_negative_u64(term: Term) -> Option<u64> {
    let value = term.as_small_int()?;
    if value < 0 {
        return None;
    }
    u64::try_from(value).ok()
}

fn badarg() -> Term {
    Term::atom(Atom::BADARG)
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::Mutex;

    use super::*;
    use crate::etf::decode::decode_term;

    struct TestDelivery {
        messages: Mutex<HashMap<u64, Vec<Term>>>,
        atom_table: AtomTable,
    }

    impl TestDelivery {
        fn new() -> Self {
            Self {
                messages: Mutex::new(HashMap::new()),
                atom_table: AtomTable::with_common_atoms(),
            }
        }
    }

    impl ControlDelivery for TestDelivery {
        fn deliver_payload(&self, target_pid: u64, payload_etf: &[u8]) -> bool {
            let mut process = Process::new(target_pid, 64);
            let mut context = ProcessContext::new();
            context.attach_process(&mut process, 0);
            let Ok(message) = decode_term(payload_etf, &mut context, &self.atom_table) else {
                return false;
            };
            self.messages
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .entry(target_pid)
                .or_default()
                .push(message);
            true
        }
    }

    struct TestRegistry(Atom, u64);

    impl ControlRegistry for TestRegistry {
        fn whereis(&self, name: Atom) -> Option<u64> {
            (name == self.0).then_some(self.1)
        }
    }

    #[test]
    fn send_control_delivers_payload_to_pid() {
        let atom_table = AtomTable::with_common_atoms();
        let frame = encode_send_frame(
            Term::atom(Atom::OK),
            Term::pid(7),
            Term::atom(Atom::OK),
            &atom_table,
        )
        .expect("frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");
        let delivery = TestDelivery::new();

        assert_eq!(
            handle_frame(control, payload, &atom_table, &delivery, None, None),
            Ok(true)
        );
        let messages = delivery
            .messages
            .lock()
            .unwrap_or_else(|error| error.into_inner());
        assert_eq!(
            messages.get(&7).and_then(|values| values.first()).copied(),
            Some(Term::atom(Atom::OK))
        );
    }

    #[test]
    fn reg_send_control_resolves_name_before_delivery() {
        let atom_table = AtomTable::with_common_atoms();
        let name = atom_table.intern("receiver");
        let frame = encode_reg_send_frame(
            Term::pid(1),
            Term::atom(Atom::OK),
            name,
            Term::atom(Atom::OK),
            &atom_table,
        )
        .expect("frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");
        let delivery = TestDelivery::new();
        let registry = TestRegistry(name, 9);

        assert_eq!(
            handle_frame(
                control,
                payload,
                &atom_table,
                &delivery,
                Some(&registry),
                None
            ),
            Ok(true)
        );
        let messages = delivery
            .messages
            .lock()
            .unwrap_or_else(|error| error.into_inner());
        assert_eq!(
            messages.get(&9).and_then(|values| values.first()).copied(),
            Some(Term::atom(Atom::OK))
        );
    }

    // ── PG_UPDATE tests ─────────────────────────────────────────────────

    #[test]
    fn pg_update_opcode_is_outside_otp_control_range() {
        // The chosen opcode must not collide with any standard OTP control op
        // (the table tops out at SPAWN_REPLY = 31), so `from_opcode` rejects it.
        const _: () = assert!(PG_UPDATE > SPAWN_REPLY);
        assert!(crate::distribution::control_link::ControlOp::from_opcode(PG_UPDATE).is_none());
    }

    #[test]
    fn pg_update_join_round_trips_preserving_node_pid_serial() {
        let atom_table = AtomTable::with_common_atoms();
        let local_node = atom_table.intern("node-a@host");
        let scope = atom_table.intern("pg");
        let group = atom_table.intern("workers");

        let frame = encode_pg_update_frame(
            PgUpdate::Join {
                scope,
                group,
                pid: 4242,
            },
            local_node,
            &atom_table,
        )
        .expect("pg join frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");
        assert!(
            payload
                == encode_term(Term::NIL, &atom_table)
                    .expect("nil encodes")
                    .as_slice()
        );

        let decoded = decode_control(control, &atom_table).expect("control decodes");
        assert_eq!(
            decoded,
            ControlMessage::PgJoin {
                scope,
                group,
                node: local_node,
                pid_number: 4242,
                serial: 0,
            }
        );
    }

    #[test]
    fn pg_update_leave_round_trips_preserving_node_pid_serial() {
        let atom_table = AtomTable::with_common_atoms();
        let local_node = atom_table.intern("node-a@host");
        let scope = atom_table.intern("pg");
        let group = atom_table.intern("workers");

        let frame = encode_pg_update_frame(
            PgUpdate::Leave {
                scope,
                group,
                pid: 7,
            },
            local_node,
            &atom_table,
        )
        .expect("pg leave frame encodes");
        let (control, _payload) = split_frame(&frame).expect("frame splits");

        let decoded = decode_control(control, &atom_table).expect("control decodes");
        assert_eq!(
            decoded,
            ControlMessage::PgLeave {
                scope,
                group,
                node: local_node,
                pid_number: 7,
                serial: 0,
            }
        );
    }

    #[test]
    fn handle_frame_routes_pg_join_to_delivery() {
        let atom_table = AtomTable::with_common_atoms();
        let local_node = atom_table.intern("node-a@host");
        let scope = atom_table.intern("pg");
        let group = atom_table.intern("workers");

        let frame = encode_pg_update_frame(
            PgUpdate::Join {
                scope,
                group,
                pid: 11,
            },
            local_node,
            &atom_table,
        )
        .expect("pg join frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");

        let delivery = TestDelivery::new();
        let pg = RecordingPgDelivery::default();
        assert_eq!(
            handle_frame(control, payload, &atom_table, &delivery, None, Some(&pg)),
            Ok(true)
        );
        let recorded = pg.joins.lock().unwrap_or_else(|error| error.into_inner());
        assert_eq!(recorded.as_slice(), &[(scope, group, local_node, 11, 0)]);
    }

    #[test]
    fn handle_frame_routes_pg_leave_to_delivery() {
        let atom_table = AtomTable::with_common_atoms();
        let local_node = atom_table.intern("node-a@host");
        let scope = atom_table.intern("pg");
        let group = atom_table.intern("workers");

        let frame = encode_pg_update_frame(
            PgUpdate::Leave {
                scope,
                group,
                pid: 13,
            },
            local_node,
            &atom_table,
        )
        .expect("pg leave frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");

        let delivery = TestDelivery::new();
        let pg = RecordingPgDelivery::default();
        assert_eq!(
            handle_frame(control, payload, &atom_table, &delivery, None, Some(&pg)),
            Ok(true)
        );
        let recorded = pg.leaves.lock().unwrap_or_else(|error| error.into_inner());
        assert_eq!(recorded.as_slice(), &[(scope, group, local_node, 13, 0)]);
    }

    #[test]
    fn pg_control_without_pg_sink_is_dropped() {
        let atom_table = AtomTable::with_common_atoms();
        let local_node = atom_table.intern("node-a@host");
        let scope = atom_table.intern("pg");
        let group = atom_table.intern("workers");

        let frame = encode_pg_update_frame(
            PgUpdate::Join {
                scope,
                group,
                pid: 11,
            },
            local_node,
            &atom_table,
        )
        .expect("pg join frame encodes");
        let (control, payload) = split_frame(&frame).expect("frame splits");

        let delivery = TestDelivery::new();
        assert_eq!(
            handle_frame(control, payload, &atom_table, &delivery, None, None),
            Ok(false)
        );
    }

    /// `(scope, group, node, pid_number, serial)` recorded per pg delivery.
    type PgRecord = (Atom, Atom, Atom, u64, u64);

    #[derive(Default)]
    struct RecordingPgDelivery {
        joins: Mutex<Vec<PgRecord>>,
        leaves: Mutex<Vec<PgRecord>>,
    }

    impl PgDelivery for RecordingPgDelivery {
        fn apply_pg_join(
            &self,
            scope: Atom,
            group: Atom,
            node: Atom,
            pid_number: u64,
            serial: u64,
        ) {
            self.joins
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .push((scope, group, node, pid_number, serial));
        }

        fn apply_pg_leave(
            &self,
            scope: Atom,
            group: Atom,
            node: Atom,
            pid_number: u64,
            serial: u64,
        ) {
            self.leaves
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .push((scope, group, node, pid_number, serial));
        }
    }

    // ── SPAWN_REQUEST / SPAWN_REPLY tests ───────────────────────────────

    use crate::distribution::Node;
    use crate::native::spawn::{SpawnMonitorResult, SpawnOptionsResult};
    use crate::term::boxed::{write_cons, write_external_pid, write_tuple};

    type SpawnRecord = (u64, Atom, Atom, Vec<Term>, SpawnOptions);

    struct MockSpawnFacility {
        next_pid: u64,
        records: Mutex<Vec<SpawnRecord>>,
    }

    impl MockSpawnFacility {
        fn new(next_pid: u64) -> Self {
            Self {
                next_pid,
                records: Mutex::new(Vec::new()),
            }
        }
    }

    impl SpawnFacility for MockSpawnFacility {
        fn spawn(
            &self,
            _caller_pid: u64,
            _module: Atom,
            _function: Atom,
            _args: Vec<Term>,
            _link_to: Option<u64>,
        ) -> Result<u64, SpawnError> {
            Ok(self.next_pid)
        }

        fn spawn_monitor(
            &self,
            _caller_pid: u64,
            _module: Atom,
            _function: Atom,
            _args: Vec<Term>,
        ) -> Result<SpawnMonitorResult, SpawnError> {
            Ok(SpawnMonitorResult {
                pid: self.next_pid,
                reference: 0,
            })
        }

        fn spawn_lambda(
            &self,
            _caller_pid: u64,
            _module: Atom,
            _lambda_index: u32,
            _link_to: Option<u64>,
        ) -> Result<u64, SpawnError> {
            Ok(self.next_pid)
        }

        fn spawn_lambda_monitor(
            &self,
            _caller_pid: u64,
            _module: Atom,
            _lambda_index: u32,
        ) -> Result<SpawnMonitorResult, SpawnError> {
            Ok(SpawnMonitorResult {
                pid: self.next_pid,
                reference: 0,
            })
        }

        fn spawn_with_options(
            &self,
            caller_pid: u64,
            module: Atom,
            function: Atom,
            args: Vec<Term>,
            options: SpawnOptions,
        ) -> Result<SpawnOptionsResult, SpawnError> {
            let monitor = options.monitor;
            self.records
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .push((caller_pid, module, function, args, options));
            Ok(SpawnOptionsResult {
                pid: self.next_pid,
                reference: monitor.then_some(1),
            })
        }

        fn spawn_lambda_with_options(
            &self,
            _caller_pid: u64,
            _module: Atom,
            _lambda_index: u32,
            _options: SpawnOptions,
        ) -> Result<SpawnOptionsResult, SpawnError> {
            Ok(SpawnOptionsResult {
                pid: self.next_pid,
                reference: None,
            })
        }
    }

    #[test]
    fn decodes_spawn_request_with_link_monitor_options() {
        let atoms = std::sync::Arc::new(AtomTable::with_common_atoms());
        let module = atoms.intern("sample");
        let function = atoms.intern("run");
        let link = atoms.intern("link");
        let monitor = atoms.intern("monitor");
        let mut process = Process::new(1, 128);
        let mut context = ProcessContext::new();
        context.set_atom_table(Some(atoms));
        context.attach_process(&mut process, 0);

        let mut arg_list_heap = [0_u64; 2];
        let arg_list =
            write_cons(&mut arg_list_heap, Term::small_int(7), Term::NIL).expect("arg list fits");
        let mut mfa_heap = [0_u64; 4];
        let mfa = write_tuple(
            &mut mfa_heap,
            &[Term::atom(module), Term::atom(function), arg_list],
        )
        .expect("mfa tuple fits");
        let mut opt2_heap = [0_u64; 2];
        let opt_tail = write_cons(&mut opt2_heap, Term::atom(monitor), Term::NIL)
            .expect("monitor option fits");
        let mut opt1_heap = [0_u64; 2];
        let opt_list =
            write_cons(&mut opt1_heap, Term::atom(link), opt_tail).expect("link option fits");
        let mut from_heap = [0_u64; 4];
        let from = write_external_pid(&mut from_heap, module, 99, 0).expect("from pid fits");
        let mut gl_heap = [0_u64; 4];
        let group_leader =
            write_external_pid(&mut gl_heap, module, 1, 0).expect("group leader fits");
        let mut request_heap = [0_u64; 7];
        let request_term = write_tuple(
            &mut request_heap,
            &[
                Term::small_int(29),
                Term::small_int(42),
                from,
                group_leader,
                mfa,
                opt_list,
            ],
        )
        .expect("request tuple fits");

        let request = decode_spawn_request(request_term, &context).expect("spawn request decodes");

        assert_eq!(request.request_id, 42);
        assert_eq!(request.from, from);
        assert_eq!(request.group_leader, group_leader);
        assert_eq!(request.mfa.module, module);
        assert_eq!(request.mfa.function, function);
        assert_eq!(request.mfa.args, vec![Term::small_int(7)]);
        assert!(request.options.link);
        assert!(request.options.monitor);
        assert_eq!(request.options.priority, None);
        assert_eq!(request.options.min_heap_size, None);
    }

    #[test]
    fn handle_spawn_request_creates_local_process_and_reply() {
        let atoms = std::sync::Arc::new(AtomTable::with_common_atoms());
        let module = atoms.intern("sample");
        let function = atoms.intern("run");
        let link = atoms.intern("link");
        let local_node_name = atoms.intern("local@host");
        let mut process = Process::new(100, 128);
        let mut context = ProcessContext::new();
        context.set_pid(Some(100));
        context.set_atom_table(Some(atoms));
        context.set_local_node(Some(Node::new(local_node_name, 0)));
        context.attach_process(&mut process, 0);

        let mut mfa_heap = [0_u64; 4];
        let mfa = write_tuple(
            &mut mfa_heap,
            &[Term::atom(module), Term::atom(function), Term::NIL],
        )
        .expect("mfa tuple fits");
        let mut opt_heap = [0_u64; 2];
        let opt_list =
            write_cons(&mut opt_heap, Term::atom(link), Term::NIL).expect("option list fits");
        let mut request_heap = [0_u64; 7];
        let request = write_tuple(
            &mut request_heap,
            &[
                Term::small_int(29),
                Term::small_int(5),
                Term::pid(100),
                Term::pid(100),
                mfa,
                opt_list,
            ],
        )
        .expect("request tuple fits");
        let facility = MockSpawnFacility::new(321);

        let reply = handle_spawn_request(request, &mut context, &facility).expect("spawn handled");
        let decoded = decode_spawn_reply(reply).expect("reply decodes");

        assert_eq!(decoded.request_id, 5);
        let pid = PidRef::new(decoded.pid).expect("reply pid");
        assert!(!pid.is_local());
        assert_eq!(pid.node(), Some(local_node_name));
        assert_eq!(pid.pid_number(), 321);
        let records = facility
            .records
            .lock()
            .unwrap_or_else(|error| error.into_inner());
        assert_eq!(records.len(), 1);
        assert_eq!(records[0].0, 100);
        assert_eq!(records[0].1, module);
        assert_eq!(records[0].2, function);
        assert!(records[0].4.link);
        assert!(!records[0].4.monitor);
    }

    #[test]
    fn alloc_spawn_reply_encodes_op_31() {
        let atoms = std::sync::Arc::new(AtomTable::with_common_atoms());
        let mut process = Process::new(1, 128);
        let mut context = ProcessContext::new();
        context.set_atom_table(Some(atoms));
        context.attach_process(&mut process, 0);

        let reply = alloc_spawn_reply(&mut context, 77, Term::pid(9)).expect("reply allocated");
        let tuple = Tuple::new(reply).expect("reply tuple");

        assert_eq!(tuple.arity(), 3);
        assert_eq!(tuple.get(0), Some(Term::small_int(31)));
        assert_eq!(tuple.get(1), Some(Term::small_int(77)));
        assert_eq!(tuple.get(2), Some(Term::pid(9)));
    }
}

#[cfg(test)]
mod ar1_row4_site1_tests {
    // ⛔ DEFECT-ASSERTING TESTS — READ THIS BEFORE TRUSTING A GREEN.
    //
    // These pin the MEASURED CORRUPT SURFACE of AR-1 row 4 at f993280. They do
    // NOT assert correct behaviour, so a green here means "the defect is still
    // present, exactly as measured" — never "this site is safe".
    //
    // ⇒ THEY GO RED WHEN AR-1 IS FIXED, AND THAT IS THE POINT. The fix lane
    // INVERTS them to assert correctness rather than deleting them; the pinned
    // counts below are the surface the fix has to move.

    use std::sync::Arc;

    use super::{RemoteMfa, SpawnRequest, alloc_spawn_request};
    use crate::atom::AtomTable;
    use crate::native::ProcessContext;
    use crate::native::spawn::SpawnOptions;
    use crate::process::Process;
    use crate::term::Term;
    use crate::term::boxed::{Cons, Tuple};

    const HEAP: usize = 512;

    /// Which body the cell drives.
    #[derive(Clone, Copy, PartialEq, Eq, Debug)]
    enum Arm {
        Fixed,
        UnrootedReplica,
    }

    /// ⛔⛔ THE SYNTHETIC POSITIVE — `alloc_spawn_request`'s body EXACTLY AS IT
    /// WAS BEFORE THE FIX, and it must stay that way.
    ///
    /// The shipped body now roots `mfa`, so it can no longer witness the defect
    /// it was written to catch. This carries the pre-fix ordering verbatim: the
    /// boxed `mfa` tuple bound BEFORE `spawn_options_to_list` and used AFTER it,
    /// with nothing rooting it across the call.
    /// ⛔ Do NOT migrate it onto `with_rooted`.
    fn alloc_spawn_request_unrooted_replica(
        context: &mut ProcessContext<'_>,
        request: &SpawnRequest,
    ) -> Result<Term, Term> {
        let args = context.alloc_list(&request.mfa.args)?;
        let mfa = context.alloc_tuple(&[
            Term::atom(request.mfa.module),
            Term::atom(request.mfa.function),
            args,
        ])?;
        let opt_list = super::spawn_options_to_list(context, request.options.clone())?;
        let op = Term::try_small_int(super::SPAWN_REQUEST).ok_or_else(super::badarg)?;
        let req_id =
            Term::try_small_int(i64::try_from(request.request_id).map_err(|_| super::badarg())?)
                .ok_or_else(super::badarg)?;
        context.alloc_tuple(&[
            op,
            req_id,
            request.from,
            request.group_leader,
            mfa,
            opt_list,
        ])
    }

    /// One cell. `options_on` selects the window arm and `arm` selects the body;
    /// everything else is identical. Returns `(available_at_entry, outcome)`.
    fn spawn_request_round_trip(
        args: usize,
        options_on: bool,
        arm: Arm,
    ) -> (usize, Result<(), String>) {
        let table = Arc::new(AtomTable::with_common_atoms());
        let module = table.intern("ar1_site1_module");
        let function = table.intern("ar1_site1_function");

        let mut process = Process::new(1, HEAP);
        let mut context = ProcessContext::new();
        context.set_atom_table(Some(Arc::clone(&table)));
        context.attach_process(&mut process, 0);

        // Arguments are SMALL INTS — immediates. Deliberate: an immediate
        // argument cannot itself go stale, so any corruption observed below is
        // attributable to the `mfa` carrier and to nothing the arguments did.
        // Their only role here is as a RULER, two heap words apiece.
        let arg_terms: Vec<Term> = (0..args)
            .map(|index| Term::small_int(i64::try_from(index % 1000).unwrap_or(0)))
            .collect();

        let available = context.process_heap().map(|h| h.available()).unwrap_or(0);

        let request = SpawnRequest {
            request_id: 7,
            from: Term::small_int(11),
            group_leader: Term::small_int(13),
            mfa: RemoteMfa {
                module,
                function,
                args: arg_terms.clone(),
            },
            options: SpawnOptions {
                link: options_on,
                monitor: options_on,
                ..SpawnOptions::default()
            },
        };

        let outcome = (|| -> Result<(), String> {
            let term = match arm {
                Arm::Fixed => alloc_spawn_request(&mut context, &request),
                Arm::UnrootedReplica => {
                    alloc_spawn_request_unrooted_replica(&mut context, &request)
                }
            }
            .map_err(|_| "alloc_spawn_request returned an error term".to_string())?;

            let outer = Tuple::new(term).ok_or_else(|| "result is not a tuple".to_string())?;
            if outer.arity() != 6 {
                return Err(format!("result arity {} not 6", outer.arity()));
            }

            // THE CARRIER: slot 4 is `mfa`, allocated BEFORE the window and used
            // AFTER it. If it went stale, this is where it shows.
            let mfa_slot = outer.get(4).ok_or_else(|| "no mfa slot".to_string())?;
            let mfa = Tuple::new(mfa_slot)
                .ok_or_else(|| "mfa is not a tuple — carrier `mfa` went stale".to_string())?;
            if mfa.arity() != 3 {
                return Err(format!(
                    "mfa arity {} not 3 — carrier `mfa` went stale",
                    mfa.arity()
                ));
            }
            if mfa.get(0) != Some(Term::atom(module)) {
                return Err("mfa module atom differs — carrier `mfa` went stale".to_string());
            }
            if mfa.get(1) != Some(Term::atom(function)) {
                return Err("mfa function atom differs — carrier `mfa` went stale".to_string());
            }

            // The argument list hangs off the carrier, so walking it is a second,
            // independent read of the same corruption. Iterative and hard-capped
            // for the usual reason: a stale carrier can alias a cell into a cycle.
            let mut seen = 0usize;
            let mut tail = mfa.get(2).ok_or_else(|| "no args slot".to_string())?;
            let cap = args * 2 + 16;
            while !tail.is_nil() {
                if seen > cap {
                    return Err(format!(
                        "args did not terminate within {cap} cells — cyclic tail, carrier `mfa` went stale"
                    ));
                }
                let cons = Cons::new(tail).ok_or_else(|| {
                    format!("arg {seen}: tail is not a cons — carrier `mfa` went stale")
                })?;
                let want = *arg_terms
                    .get(seen)
                    .ok_or_else(|| format!("arg {seen}: more args recovered than were put"))?;
                if cons.head() != want {
                    return Err(format!(
                        "arg {seen}: value differs from the one put — carrier `mfa` went stale"
                    ));
                }
                seen += 1;
                tail = cons.tail();
            }
            if seen != args {
                return Err(format!("recovered {seen} args, put {args}"));
            }
            Ok(())
        })();

        (available, outcome)
    }

    /// One sweep of the whole argument range against one body. Returns the ON
    /// and OFF red rows plus the clean/refused counts and the predicted-window
    /// headroom range, so the arithmetic can be checked against the data.
    #[allow(clippy::type_complexity)]
    fn sweep(
        arm: Arm,
    ) -> (
        Vec<String>,
        usize,
        usize,
        Vec<String>,
        usize,
        usize,
        i64,
        i64,
    ) {
        let mut on_red = Vec::new();
        let mut on_ok = 0usize;
        let mut on_refused = 0usize;
        let mut off_red = Vec::new();
        let mut off_ok = 0usize;
        let mut off_refused = 0usize;
        let mut headroom_min = i64::MAX;
        let mut headroom_max = i64::MIN;

        // ⛔ THE RANGE IS DERIVED FROM THE ARITHMETIC, NOT CHOSEN FOR ROUNDNESS,
        // AND THE FIRST VERSION OF THIS SWEEP GOT IT WRONG. It ran 0..=200,
        // which walks the window headroom down only to 512 - 400 - 4 = 108 —
        // nowhere near the [0, 4) target — and returned 201 clean cells that
        // meant NOTHING. To reach the band the ruler needs 2N in (504, 508], so
        // N must pass through 251..=254. Running to 300 clears it with margin
        // and then continues past it, which also demonstrates the far side.
        for args in 0..=300usize {
            for options_on in [true, false] {
                let (available, result) = spawn_request_round_trip(args, options_on, arm);
                let predicted = (available as i64) - 2 * (args as i64) - 4;
                if options_on {
                    headroom_min = headroom_min.min(predicted);
                    headroom_max = headroom_max.max(predicted);
                }
                match result {
                    Ok(()) => {
                        if options_on {
                            on_ok += 1;
                        } else {
                            off_ok += 1;
                        }
                    }
                    Err(reason) if reason.contains("returned an error term") => {
                        if options_on {
                            on_refused += 1;
                        } else {
                            off_refused += 1;
                        }
                    }
                    Err(reason) => {
                        let row = format!(
                            "args {args:>4} available {available:>4} predicted-window-headroom \
                             {predicted:>5} : {reason}"
                        );
                        eprintln!(
                            "site 1 [{arm:?}] RED [options {}] {row}",
                            if options_on { "ON" } else { "OFF" }
                        );
                        if options_on {
                            on_red.push(row);
                        } else {
                            off_red.push(row);
                        }
                    }
                }
            }
        }

        eprintln!(
            "site 1 [{arm:?}] arm ON : {} red, {on_ok} clean, {on_refused} refused",
            on_red.len()
        );
        eprintln!(
            "site 1 [{arm:?}] arm OFF: {} red, {off_ok} clean, {off_refused} refused",
            off_red.len()
        );

        (
            on_red,
            on_ok,
            on_refused,
            off_red,
            off_ok,
            off_refused,
            headroom_min,
            headroom_max,
        )
    }

    /// AR-1 row 4, site 1 (`mfa`) — ✅ INVERTED.
    ///
    /// The control arm is asserted FIRST and keeps every claim the pre-fix probe
    /// made: the band is narrow (<=2 cells, because a <=4-word window stepped by
    /// a 2-word ruler admits at most two), the OFF arm is a STRUCTURAL negative
    /// control (both options false makes `spawn_options_to_list` request 0 words,
    /// so it cannot collect), and some ON cell must be clean or the reader is
    /// broken rather than the site defective. Only then is the fixed body
    /// measured, over the identical range.
    #[test]
    fn ar1_site1_alloc_spawn_request_band() {
        // ⛔⛔ POSITIVE CONTROL FIRST, and it licenses everything below it.
        let (
            on_red,
            on_ok,
            _on_refused,
            off_red,
            _off_ok,
            _off_refused,
            headroom_min,
            headroom_max,
        ) = sweep(Arm::UnrootedReplica);

        eprintln!(
            "site 1 predicted-window-headroom range over the ON sweep: \
             [{headroom_min}, {headroom_max}]"
        );

        assert!(
            on_ok > 0,
            "control: some ON cell must be clean, or the reader is broken rather than the site \
             defective"
        );

        // ⛔ THE STRUCTURAL NEGATIVE CONTROL. Arm OFF differs from arm ON in
        // exactly one respect: `spawn_options_to_list` requests 0 words instead
        // of 4, so it cannot collect. If this fires, the exposure window is NOT
        // `spawn_options_to_list` and the whole attribution above is wrong.
        assert!(
            off_red.is_empty(),
            "ATTRIBUTION BROKEN: arm OFF corrupted {} cells, but with both options false \
             `spawn_options_to_list` requests 0 words and cannot collect. The window is not where \
             this probe says it is — re-derive it.\n{}",
            off_red.len(),
            off_red.join("\n")
        );

        assert!(
            !on_red.is_empty(),
            "POSITIVE CONTROL DEAD: the unrooted replica no longer corrupts the carrier anywhere \
             in the range. The pressure regime is gone, so the fixed arm's success below would \
             mean nothing."
        );

        // ⭐ THE BAND'S NARROWNESS IS THE FINDING, so it is asserted rather than
        // merely reported. A wide band would mean the exposure is not the bounded
        // allocation this probe claims it is.
        assert!(
            on_red.len() <= 2,
            "site 1: {} red cells, but the window is a BOUNDED <=4-word allocation stepped by a \
             2-word ruler, so at most 2 cells can land in it. A wider band refutes the \
             structural argument — re-derive it, do not widen this bound.\n{}",
            on_red.len(),
            on_red.join("\n")
        );

        // ✅ THE CLAIM. Same range, same regime, through the rooted body.
        let (fixed_on_red, fixed_on_ok, _, fixed_off_red, _, _, _, _) = sweep(Arm::Fixed);

        assert!(
            fixed_on_red.is_empty() && fixed_off_red.is_empty(),
            "site 1 is NOT rooted: {} ON and {} OFF cells still lost the carrier, while the \
             replica corrupted {} in the same run.\n{}\n{}",
            fixed_on_red.len(),
            fixed_off_red.len(),
            on_red.len(),
            fixed_on_red.join("\n"),
            fixed_off_red.join("\n")
        );
        assert!(
            fixed_on_ok > 0,
            "site 1: the fixed arm produced no clean ON cell at all — that is a dead reader, not \
             a defended site"
        );
    }
}