Skip to main content

blue_lang_proc/
node.rs

1//! The node: blue processes reachable from blue (BLUE-GAPS G11).
2//!
3//! Until this module, everything in this crate was reachable only from Rust:
4//! a blue program could not spawn, send or receive, and a NATS subscriber had
5//! to poll from the main program. A [`Node`] is the process runtime one
6//! interpreter's program runs against, and [`install`] binds its words:
7//!
8//! ```text
9//! spawn(f[, args][, opts])    spawn_link(f[, args][, opts])
10//! spawn_server(handler, state[, opts])                   opts: {scheduling: :normal | :idle}
11//! send(pid, msg)              receive([pattern][, timeout_ms][, on_timeout])
12//! self()   alive?(pid)        link(pid)  unlink(pid)     monitor(pid)  demonitor(ref)
13//! exit(pid, reason)           trap_exit(flag)
14//! supervisor(strategy, children[, max_restarts])         children(sup)
15//! defmigrate(target, new_fn[, transform])
16//! def_hash(f)
17//! evict(pid, wire, peer)      accept(wire, timeout_ms)   (with a Carrier: comm-scavenger)
18//! ```
19//!
20//! ## The shapes, adopted rather than invented
21//!
22//! - **Copy on send, share nothing** (`theory/BLUE.md` §V.8, BLUE-EXCLUSIVITY
23//!   §II.3): a message is deep-copied, and a spawned function is turned back
24//!   into code and recompiled in the child ([`crate::code`]), never shared.
25//! - **Erlang's process vocabulary**: links propagate abnormal exits unless the
26//!   receiver traps them (then they arrive as `[:exit, pid, reason]`), monitors
27//!   deliver `[:down, ref, pid, reason]`, `:kill` cannot be trapped, a send to a
28//!   dead process is dropped.
29//! - **The supervisor strategies are [`Strategy`]'s**, with intensity checked on
30//!   the child that exited and the mailbox emptied on restart, exactly as the
31//!   Rust-level [`crate::Supervisor`] does.
32//! - **The interleaving is replayable** (BLUE-EXCLUSIVITY §II.13): one thread,
33//!   procs walked in spawn order, a fixed reduction quantum. What a process
34//!   could not replay from its program alone — the messages it received, the
35//!   host values it read, the pids it was handed — is recorded on its tape, and
36//!   a crash writes the tape out ([`crate::trace`]). An `:idle` process is the
37//!   one exception to the one thread: its slices run on a low-priority thread
38//!   of its own while the node runs the rest ([`IdleLane::Apart`]), so where
39//!   they fall among the others is the OS's choice and not replayable; its
40//!   own crash still replays from its tape. It still acts only while the root
41//!   waits: a word it calls between waits holds until the next one. [`IdleLane::Held`] keeps the whole
42//!   interleaving replayable by waiting for each idle slice.
43//!
44//! ## The root is the program; children are VM processes
45//!
46//! The program a door evaluates is process [`ROOT`] (pid 0). It runs on
47//! whichever executor the door chose and is never preempted: it gives the
48//! other processes the CPU by waiting — `receive`, `defmigrate` — and its
49//! `receive` drives the scheduler until it has a message, its timeout passes,
50//! or nothing can ever send (a deadlock, reported rather than hung on). The
51//! program ends when the root does, as an escript does; a process the root
52//! never waited for does not run to completion.
53//!
54//! Every other process runs on the bytecode VM, which is the executor that can
55//! preempt (a quantum) and park (`receive` on an empty mailbox costs nothing).
56//!
57//! ## What a host must provide
58//!
59//! [`NodeConfig::child`] builds a fresh interpreter for a child. The node adds
60//! its own words to it, and rebinds every host name (`blue_lang_waku`'s host
61//! capabilities) to a recorder, so a child's reads of the clock, the network
62//! or the filesystem land on its tape.
63
64use std::collections::{BTreeMap, BTreeSet, VecDeque};
65use std::path::PathBuf;
66use std::sync::{mpsc, Arc, Condvar, LazyLock, Mutex, MutexGuard, OnceLock, Weak};
67use std::time::{Duration, Instant};
68
69use tatara_lisp::{Span, Spanned};
70use tatara_lisp_eval::ffi::Arity;
71use tatara_lisp_eval::vm::run::CompiledClosure;
72use tatara_lisp_eval::vm::{
73    compile_program, Budget, Chunk, FnRef, ParkError, Vm, DEFAULT_MAX_DEPTH,
74};
75use tatara_lisp_eval::{EvalError, Interpreter, MapKey, Value};
76
77use blue_lang_waku::Capability;
78
79use crate::code::{self, Code, Entry};
80use crate::image::{self, Account, Arrivals, Continuation, Image, Supervision};
81use crate::mailbox::copy_data;
82use crate::pattern;
83use crate::receipt::{
84    CarriedIo, Frame, HostIo, Identities, Places, Receipt, ReceiptSink, Roots, StateRoot, Subject,
85};
86use crate::scheduling::{self, IdleLane, Scheduling};
87use crate::trace::{Carried, CarriedAccount, Journal, SliceRecord, TapeEntry, Trace};
88use crate::violation::{breach, defined_key, Violation};
89use crate::{provided_name, raised, ExitReason, Pid, Proc, ProcState, Slice, Strategy};
90
91/// The program itself.
92pub const ROOT: Pid = Pid(0);
93
94/// Reductions per slice. BEAM uses 4 000; a blue reduction is one VM
95/// instruction, smaller than a BEAM reduction (a function call).
96pub const DEFAULT_QUANTUM: usize = 1_000;
97
98/// Tape entries one incarnation may record before its trace is marked
99/// truncated. A replay needs the whole tape, so a truncated trace says so
100/// rather than replaying a suffix.
101pub const DEFAULT_MAX_TAPE: usize = 100_000;
102
103/// Messages a held process's mailbox journal keeps of each kind (taken,
104/// still waiting), when the host does not say.
105pub const DEFAULT_JOURNAL: usize = 8;
106
107/// Slices the interleaving record keeps.
108const SCHEDULE_RING: usize = 1_024;
109
110/// Crash traces a node keeps in memory.
111const KEPT_TRACES: usize = 16;
112
113/// Instructions a migration's definitions and transform may run.
114const MIGRATION_FUEL: usize = 10_000_000;
115
116/// Supervisor restarts of one child before it gives up, when the program does
117/// not say.
118pub const DEFAULT_MAX_RESTARTS: usize = 3;
119
120/// Every word [`install`] binds that a blue program can spell.
121pub const PROCESS_WORDS: &[&str] = &[
122    "spawn",
123    "spawn_link",
124    "spawn_server",
125    "send",
126    "receive",
127    "self",
128    "alive?",
129    "link",
130    "unlink",
131    "monitor",
132    "demonitor",
133    "exit",
134    "trap_exit",
135    "supervisor",
136    "children",
137    "defmigrate",
138    "def_hash",
139];
140
141pub const CARRIED_WORDS: &[&str] = &["evict", "accept"];
142
143pub const EVICTED: &str = "evicted";
144
145/// The word a server's loop waits in. Not spellable, so not in
146/// [`PROCESS_WORDS`].
147pub const SERVER_NEXT: &str = "%server_next";
148
149/// A server's loop: take the next message at a boundary, hand it and the state
150/// to the handler, keep what it returns.
151///
152/// The state lives in a process-private global rather than in a VM frame, so a
153/// migration can read and replace it while the process is parked between
154/// messages.
155static SERVER_LOOP: LazyLock<Vec<Spanned>> = LazyLock::new(|| {
156    tatara_lisp::read_spanned(
157        "(define (%serve) (%serve_step (%server_next %state)))
158         (define (%serve_step m) (set! %state (%handler m %state)) (%serve))",
159    )
160    .expect("the server loop reads")
161});
162
163static TRACE_DIR: OnceLock<Option<PathBuf>> = OnceLock::new();
164
165/// Where nodes write crash traces. Only the first call counts, like
166/// `blue_lang_runtime::set_execution_bounds`; a node built before any call
167/// keeps its traces in memory only.
168pub fn set_trace_dir(dir: Option<PathBuf>) {
169    let _ = TRACE_DIR.set(dir);
170}
171
172static JOURNAL: OnceLock<usize> = OnceLock::new();
173
174/// How many messages of each kind a held process's mailbox journal keeps;
175/// `0` keeps none. Only the first call counts, like [`set_trace_dir`]; a
176/// node built before any call keeps [`DEFAULT_JOURNAL`].
177pub fn set_journal_bound(bound: usize) {
178    let _ = JOURNAL.set(bound);
179}
180
181/// How a node reads time.
182#[derive(Clone, Copy, Debug, PartialEq, Eq)]
183pub enum ClockKind {
184    /// Milliseconds since the node started; an idle node sleeps until the
185    /// next deadline.
186    Real,
187    /// Time moves only when every process is waiting: the node jumps to the
188    /// next deadline. Deterministic, and the only clock a host with no clock
189    /// (wasm) can offer.
190    Virtual,
191}
192
193impl ClockKind {
194    #[must_use]
195    pub fn label(self) -> &'static str {
196        match self {
197            ClockKind::Real => "real",
198            ClockKind::Virtual => "virtual",
199        }
200    }
201}
202
203struct Clock {
204    kind: ClockKind,
205    origin: Option<Instant>,
206    virtual_ms: u64,
207}
208
209impl Clock {
210    fn new(kind: ClockKind) -> Self {
211        Self {
212            kind,
213            origin: None,
214            virtual_ms: 0,
215        }
216    }
217
218    fn now(&mut self) -> u64 {
219        match self.kind {
220            ClockKind::Real => u64::try_from(
221                self.origin
222                    .get_or_insert_with(Instant::now)
223                    .elapsed()
224                    .as_millis(),
225            )
226            .unwrap_or(u64::MAX),
227            ClockKind::Virtual => self.virtual_ms,
228        }
229    }
230
231    /// Move to `t`; the real clock answers how long to sleep.
232    fn advance_to(&mut self, t: u64) -> Option<Duration> {
233        match self.kind {
234            ClockKind::Real => {
235                let now = self.now();
236                (t > now).then(|| Duration::from_millis(t - now))
237            }
238            ClockKind::Virtual => {
239                self.virtual_ms = self.virtual_ms.max(t);
240                None
241            }
242        }
243    }
244}
245
246/// A fresh interpreter for a child, before the node adds its words.
247pub type ChildFactory = Arc<dyn Fn() -> Interpreter<()> + Send + Sync>;
248
249/// The host's allocator meter, through which a node holds each process to the
250/// live-byte bound of the evaluation that spawned it (`max_alloc_bytes`).
251///
252/// A process is charged what its own slices leave live and the messages
253/// delivered to it, never what its sender or its spawner holds: a message's
254/// copy belongs to the receiver, and a process's interpreter and code belong
255/// to the node. Past its bound a process exits
256/// [`ExitReason::memory_exhausted`] at the end of the slice that passed it.
257#[derive(Clone, Copy, Debug)]
258pub struct Meter {
259    /// The bound of the account running now: `None` when it has none, or the
260    /// host counts nothing.
261    pub bound: fn() -> Option<u64>,
262    /// Run the closure charged to the account whose live bytes the `i64`
263    /// carries, held to the bound; answers whether it passed the bound.
264    pub charged: fn(&mut i64, Option<u64>, &mut dyn FnMut()) -> bool,
265    /// Charge bytes to the account running now.
266    pub charge: fn(i64),
267}
268
269#[derive(Clone, Copy, Debug)]
270pub struct Carrier {
271    pub room: usize,
272    pub queue: fn(&Value, &Value, &[String], bool) -> Result<usize, String>,
273    pub take: fn(&Value, &str, Duration, bool) -> Result<Vec<String>, String>,
274}
275
276/// What a node runs its processes under.
277#[derive(Clone)]
278pub struct NodeConfig {
279    /// Each child's budget: fuel and depth bound an incarnation, the quantum is
280    /// the slice.
281    pub budget: Budget,
282    pub clock: ClockKind,
283    pub max_tape: usize,
284    /// Where crash traces are written, if anywhere.
285    pub trace_dir: Option<PathBuf>,
286    pub child: ChildFactory,
287    /// The hold every process runs under, when the host holds them to one.
288    pub frame: Option<Frame>,
289    /// Messages of each kind a held process's mailbox journal keeps; `0`
290    /// keeps no journal.
291    pub journal: usize,
292    /// The ledger the host's counting shims tally into.
293    pub host_io: HostIo,
294    /// Each definition's `def_hash`, for traces and receipts.
295    pub identities: Identities,
296    /// Each definition's file, for the place a trace names.
297    pub places: Places,
298    /// Where each held process's receipt goes when an incarnation ends.
299    pub receipts: Option<ReceiptSink>,
300    /// How a server's receipt names its state; without one, a server's
301    /// receipt carries no roots.
302    pub state_root: Option<StateRoot>,
303    /// The allocator meter that holds each process to its live-byte bound;
304    /// `None` holds no process to one.
305    pub meter: Option<Meter>,
306    pub carrier: Option<Carrier>,
307    pub dispatch_steps: Option<usize>,
308    pub count_steps: bool,
309    pub idle_lane: IdleLane,
310}
311
312impl NodeConfig {
313    /// Unbounded fuel, the default depth, [`DEFAULT_QUANTUM`], a virtual
314    /// clock, the trace directory [`set_trace_dir`] named, the journal bound
315    /// [`set_journal_bound`] named, no hold, no receipts and no meter.
316    #[must_use]
317    pub fn new(child: ChildFactory) -> Self {
318        Self {
319            budget: Budget {
320                fuel: None,
321                max_depth: Some(DEFAULT_MAX_DEPTH),
322                quantum: Some(DEFAULT_QUANTUM),
323            },
324            clock: ClockKind::Virtual,
325            max_tape: DEFAULT_MAX_TAPE,
326            trace_dir: TRACE_DIR.get().cloned().flatten(),
327            child,
328            frame: None,
329            journal: JOURNAL.get().copied().unwrap_or(DEFAULT_JOURNAL),
330            host_io: HostIo::default(),
331            identities: Identities::default(),
332            places: Places::default(),
333            receipts: None,
334            state_root: None,
335            meter: None,
336            carrier: None,
337            dispatch_steps: None,
338            count_steps: false,
339            idle_lane: IdleLane::default(),
340        }
341    }
342
343    fn journals(&self) -> bool {
344        self.frame.is_some() && self.journal > 0
345    }
346
347    fn carrying(&self) -> Carrying {
348        let cap = Capability::CommScavenger;
349        Carrying {
350            carrier: self.carrier.is_some(),
351            scoped: self.frame.as_ref().is_some_and(|f| {
352                f.provided
353                    .iter()
354                    .any(|p| p.scope.is_some() && p.provision.capability() == cap)
355            }),
356            withheld: self
357                .frame
358                .as_ref()
359                .is_some_and(|f| f.withheld.contains(&cap)),
360        }
361    }
362}
363
364#[derive(Clone, Copy)]
365struct Carrying {
366    carrier: bool,
367    scoped: bool,
368    withheld: bool,
369}
370
371#[derive(Clone, Debug)]
372struct Sup {
373    strategy: Strategy,
374    max_restarts: usize,
375    children: Vec<Pid>,
376}
377
378#[derive(Clone, Debug)]
379enum Kind {
380    Root,
381    Plain,
382    Server,
383    Supervisor(Sup),
384}
385
386/// What the node knows about one live process.
387struct Info {
388    kind: Kind,
389    /// The program a restart recompiles and a trace writes down.
390    code: Code,
391    chunk: Option<Arc<Chunk>>,
392    /// The definition the process runs, for `defmigrate`.
393    def_name: Option<String>,
394    links: BTreeSet<Pid>,
395    /// `(ref, watcher)` for every monitor on this process.
396    monitors: Vec<(i64, Pid)>,
397    trap_exit: bool,
398    sup: Option<Pid>,
399    restarts: usize,
400    /// Running or parked. A supervised child keeps its record after it
401    /// finishes, so its supervisor can restart it, and is not live then.
402    live: bool,
403    /// A server's state at its last boundary: where its tape starts.
404    snapshot: Option<Value>,
405    /// The state a server restarts with.
406    initial: Option<Value>,
407    /// The state this incarnation of a server began with.
408    entered: Option<Value>,
409    /// Live bytes this incarnation holds ([`Meter`]).
410    held: i64,
411    /// The live bytes it may hold: its spawner's bound, kept across restarts.
412    bound: Option<u64>,
413    scheduling: Scheduling,
414    dispatch: Option<usize>,
415    dispatch_from: Option<usize>,
416    armed: bool,
417    kept: Option<Value>,
418    asked: Option<usize>,
419}
420
421impl Info {
422    fn new(kind: Kind) -> Self {
423        Self {
424            kind,
425            code: Code::default(),
426            chunk: None,
427            def_name: None,
428            links: BTreeSet::new(),
429            monitors: Vec::new(),
430            trap_exit: false,
431            sup: None,
432            restarts: 0,
433            live: true,
434            snapshot: None,
435            initial: None,
436            entered: None,
437            held: 0,
438            bound: None,
439            scheduling: Scheduling::Normal,
440            dispatch: None,
441            dispatch_from: None,
442            armed: false,
443            kept: None,
444            asked: None,
445        }
446    }
447
448    fn ask_dispatch(&mut self, own: Option<usize>) {
449        self.asked = own;
450        self.dispatch = match (self.dispatch, own) {
451            (Some(app), Some(own)) => Some(app.min(own)),
452            (app, own) => app.or(own),
453        };
454    }
455}
456
457/// What a parked child is waiting for.
458#[derive(Clone, Debug)]
459enum Wait {
460    Receive {
461        pattern: Value,
462        deadline: Option<u64>,
463    },
464    /// A server between messages: its migration boundary.
465    ServerNext,
466}
467
468#[derive(Clone)]
469struct Migration {
470    request: u64,
471    code: Code,
472    new_name: String,
473    transform: Option<Code>,
474}
475
476#[derive(Default, Clone)]
477struct Tape {
478    entries: Vec<TapeEntry>,
479    truncated: bool,
480    carried: Option<Carried>,
481}
482
483impl Tape {
484    fn arrived(carried: Option<Vec<TapeEntry>>, image: &str) -> Self {
485        match carried {
486            Some(entries) => Self {
487                carried: Some(Carried::of(&entries, image)),
488                entries,
489                truncated: false,
490            },
491            None => Self {
492                entries: Vec::new(),
493                truncated: true,
494                carried: None,
495            },
496        }
497    }
498}
499
500#[derive(Clone, Debug)]
501struct Brought {
502    mark: CarriedAccount,
503    input_root: Option<String>,
504}
505
506struct NodeState {
507    current: Pid,
508    next_pid: u64,
509    next_ref: i64,
510    next_request: u64,
511    boxes: BTreeMap<Pid, VecDeque<Value>>,
512    info: BTreeMap<Pid, Info>,
513    /// Spawned, not yet in the run queue: a spawn inside a slice cannot touch
514    /// the queue the scheduler is walking.
515    pending: Vec<Proc<()>>,
516    waits: BTreeMap<Pid, Wait>,
517    tapes: BTreeMap<Pid, Tape>,
518    /// The last messages each held process took, oldest first.
519    journals: BTreeMap<Pid, VecDeque<Value>>,
520    accounts: BTreeMap<Pid, Brought>,
521    /// Exit signals sent inside a slice, delivered between slices.
522    signals: VecDeque<(Pid, Pid, Value)>,
523    migrations: BTreeMap<Pid, Migration>,
524    outcomes: BTreeMap<u64, Vec<(Pid, Result<(), String>)>>,
525    schedule: VecDeque<SliceRecord>,
526    traces: VecDeque<Trace>,
527    clock: Clock,
528    /// An abnormal exit signal the root did not trap: raised by its next wait.
529    root_signal: Option<(Pid, Value)>,
530    /// Bytes of the messages children delivered to the root while it waited,
531    /// charged to it when its wait returns.
532    root_due: i64,
533    evicting: BTreeSet<Pid>,
534    arrivals: Arrivals,
535    aloft: BTreeSet<Pid>,
536    doomed: BTreeMap<Pid, Value>,
537    recalled: BTreeSet<Pid>,
538}
539
540impl NodeState {
541    fn new(clock: ClockKind) -> Self {
542        let mut info = BTreeMap::new();
543        info.insert(ROOT, Info::new(Kind::Root));
544        let mut boxes = BTreeMap::new();
545        boxes.insert(ROOT, VecDeque::new());
546        Self {
547            current: ROOT,
548            next_pid: 1,
549            next_ref: 1,
550            next_request: 1,
551            boxes,
552            info,
553            pending: Vec::new(),
554            waits: BTreeMap::new(),
555            tapes: BTreeMap::new(),
556            journals: BTreeMap::new(),
557            accounts: BTreeMap::new(),
558            signals: VecDeque::new(),
559            migrations: BTreeMap::new(),
560            outcomes: BTreeMap::new(),
561            schedule: VecDeque::new(),
562            traces: VecDeque::new(),
563            clock: Clock::new(clock),
564            root_signal: None,
565            root_due: 0,
566            evicting: BTreeSet::new(),
567            arrivals: Arrivals::default(),
568            aloft: BTreeSet::new(),
569            doomed: BTreeMap::new(),
570            recalled: BTreeSet::new(),
571        }
572    }
573
574    fn running(&self) -> Pid {
575        scheduling::lane().unwrap_or(self.current)
576    }
577
578    fn supervisor(&mut self, strategy: Strategy, max_restarts: usize) -> Pid {
579        let sup = Pid(self.next_pid);
580        self.next_pid += 1;
581        self.info.insert(
582            sup,
583            Info::new(Kind::Supervisor(Sup {
584                strategy,
585                max_restarts,
586                children: Vec::new(),
587            })),
588        );
589        self.boxes.insert(sup, VecDeque::new());
590        sup
591    }
592
593    fn enrol(&mut self, sup: Pid, child: Pid) {
594        if let Some(Info {
595            kind: Kind::Supervisor(s),
596            ..
597        }) = self.info.get_mut(&sup)
598        {
599            s.children.push(child);
600        }
601    }
602
603    fn drop_child(&mut self, sup: Pid, child: Pid) {
604        if let Some(Info {
605            kind: Kind::Supervisor(s),
606            ..
607        }) = self.info.get_mut(&sup)
608        {
609            s.children.retain(|c| *c != child);
610        }
611    }
612
613    fn held_at_boundary(&self, pid: Pid) -> bool {
614        self.migrations.contains_key(&pid) || self.evicting.contains(&pid)
615    }
616
617    fn held_at_receive(&self, pid: Pid) -> bool {
618        self.evicting.contains(&pid)
619            && self
620                .info
621                .get(&pid)
622                .is_some_and(|i| matches!(i.kind, Kind::Plain))
623    }
624
625    fn opens_dispatch(&mut self, pid: Pid) -> bool {
626        match self.info.get_mut(&pid) {
627            Some(info) if info.dispatch.is_some() && !info.armed => {
628                info.armed = true;
629                true
630            }
631            _ => false,
632        }
633    }
634
635    fn deliver(&mut self, to: Pid, msg: Value) {
636        if let Some(q) = self.boxes.get_mut(&to) {
637            q.push_back(msg);
638        }
639    }
640
641    /// Would `pid`'s parked call succeed now? Asked by the scheduler, so it
642    /// reads; the call itself is what takes.
643    fn is_ready(&mut self, pid: Pid) -> bool {
644        let now = self.clock.now();
645        let mailbox = self.boxes.get(&pid);
646        match self.waits.get(&pid) {
647            Some(Wait::Receive { pattern, deadline }) => {
648                !self.held_at_receive(pid)
649                    && (mailbox.is_some_and(|b| pattern::first_match(b, pattern).is_some())
650                        || deadline.is_some_and(|d| now >= d))
651            }
652            Some(Wait::ServerNext) => {
653                mailbox.is_some_and(|b| !b.is_empty()) && !self.held_at_boundary(pid)
654            }
655            None => mailbox.is_some_and(|b| !b.is_empty()),
656        }
657    }
658
659    fn earliest_deadline(&self) -> Option<u64> {
660        self.waits
661            .values()
662            .filter_map(|w| match w {
663                Wait::Receive { deadline, .. } => *deadline,
664                Wait::ServerNext => None,
665            })
666            .min()
667    }
668
669    fn forget(&mut self, pid: Pid) {
670        self.info.remove(&pid);
671        self.boxes.remove(&pid);
672        self.waits.remove(&pid);
673        self.tapes.remove(&pid);
674        self.journals.remove(&pid);
675        self.accounts.remove(&pid);
676        self.evicting.remove(&pid);
677    }
678
679    /// `pid` took `msg`: keep it among its last `bound`.
680    fn note_taken(&mut self, pid: Pid, msg: &Value, bound: usize) {
681        let ring = self.journals.entry(pid).or_default();
682        if ring.len() == bound {
683            ring.pop_front();
684            if let Some(a) = self.accounts.get_mut(&pid) {
685                a.mark.dropped_oldest(ring);
686            }
687        }
688        ring.push_back(msg.clone());
689    }
690}
691
692/// The process runtime one program runs against.
693pub struct Node {
694    config: NodeConfig,
695    me: Weak<Node>,
696    state: Mutex<NodeState>,
697    /// The run queue, in spawn order. Locked only by the root while it waits,
698    /// never from inside a slice, so a child's words touch only `state`.
699    sched: Mutex<Vec<Proc<()>>>,
700    /// The interpreter every child forks: the host's, plus this node's child
701    /// words and host recorders. Built at the first spawn.
702    proto: OnceLock<Arc<Interpreter<()>>>,
703    lanes: Mutex<BTreeMap<Pid, mpsc::Sender<Proc<()>>>>,
704    flights: Mutex<mpsc::Receiver<Flight>>,
705    radio: mpsc::Sender<Flight>,
706    gate: Mutex<Gate>,
707    gated: Condvar,
708    rooted: Mutex<Option<Weak<()>>>,
709}
710
711#[derive(Default)]
712struct Gate {
713    driving: usize,
714    inside: usize,
715}
716
717struct Pass<'a>(&'a Node);
718
719impl Drop for Pass<'_> {
720    fn drop(&mut self) {
721        let mut g = lock(&self.0.gate);
722        g.inside -= 1;
723        self.0.gated.notify_all();
724    }
725}
726
727struct Driving<'a>(&'a Node);
728
729impl Drop for Driving<'_> {
730    fn drop(&mut self) {
731        let node = self.0;
732        let mut g = lock(&node.gate);
733        g.driving -= 1;
734        while g.driving == 0 && g.inside > 0 {
735            g = node
736                .gated
737                .wait(g)
738                .unwrap_or_else(std::sync::PoisonError::into_inner);
739        }
740    }
741}
742
743enum Flight {
744    Landed(Proc<()>, Slice, usize),
745    Poked,
746    Fell(Box<dyn std::any::Any + Send>),
747}
748
749fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
750    m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
751}
752
753fn kw(s: &str) -> Value {
754    Value::Keyword(Arc::from(s))
755}
756
757fn int(n: impl TryInto<i64>) -> Value {
758    Value::Int(n.try_into().unwrap_or(i64::MAX))
759}
760
761/// The blue value of an exit reason.
762#[must_use]
763pub fn reason_value(r: &ExitReason) -> Value {
764    match r {
765        ExitReason::Normal => kw("normal"),
766        ExitReason::Crashed(m) => Value::list([kw("crashed"), Value::Str(Arc::from(m.as_str()))]),
767        ExitReason::Runaway(m) => Value::list([kw("runaway"), Value::Str(Arc::from(m.as_str()))]),
768        ExitReason::Killed(m) => Value::list([kw("killed"), Value::Str(Arc::from(m.as_str()))]),
769        ExitReason::Fatal(m) => Value::list([kw("fatal"), Value::Str(Arc::from(m.as_str()))]),
770    }
771}
772
773fn is_keyword(v: &Value, name: &str) -> bool {
774    matches!(v, Value::Keyword(k) if k.trim_start_matches(':') == name)
775}
776
777fn render(v: &Value) -> String {
778    v.to_string()
779}
780
781/// Whether a crash's trace is written out, not only kept in memory: a
782/// supervised crash is the one a running system recovers from, so nobody
783/// sees it happen, and a fatal one is never restarted, so its trace is all
784/// that is left of it.
785fn written(supervised: bool, reason: &ExitReason) -> bool {
786    supervised || reason.is_fatal()
787}
788
789fn refuse(word: &str, msg: impl Into<String>, span: Span) -> EvalError {
790    EvalError::native_fn(Arc::<str>::from(word), msg.into(), span)
791}
792
793fn parking(e: &ParkError) -> String {
794    match e {
795        ParkError::NoFrames => "it holds no frames: it has not started, or it has returned".into(),
796        ParkError::NoDataForm { what, at } => format!(
797            "it holds {what} in {at}, which has no data form and does not cross a wire; open it \
798             inside the process once it has moved"
799        ),
800        ParkError::OtherChunk { what, at } => format!(
801            "it runs code it was not spawned with ({what}, {at}), and its image carries only the \
802             code it was spawned with"
803        ),
804        ParkError::ChunkMismatch { parked, chunk } => format!(
805            "its frames were parked in chunk {parked}, and its code compiles here to chunk \
806             {chunk}: this blue compiles it differently from the one it left"
807        ),
808        ParkError::UnknownNative { name } => {
809            format!("it holds the native function `{name}`, which this node does not register")
810        }
811        ParkError::Malformed(why) => format!("its frames are malformed: {why}"),
812    }
813}
814
815fn pid_arg(v: &Value, word: &str, span: Span) -> Result<Pid, EvalError> {
816    match v {
817        Value::Int(n) if *n >= 0 => Ok(Pid(n.unsigned_abs())),
818        other => Err(refuse(
819            word,
820            format!("expected a pid, got {}", other.type_name()),
821            span,
822        )),
823    }
824}
825
826fn compile(forms: &[Spanned]) -> Result<Chunk, String> {
827    compile_program(forms).map_err(|e| e.to_string())
828}
829
830/// Run `forms` to completion against `interp`, bounded.
831fn run_in(
832    interp: &mut Interpreter<()>,
833    forms: &[Spanned],
834    depth: Option<usize>,
835) -> Result<Value, String> {
836    let chunk = compile(forms)?;
837    let mut vm = Vm::with_budget(Budget {
838        fuel: Some(MIGRATION_FUEL),
839        max_depth: depth,
840        quantum: None,
841    });
842    vm.run(&chunk, interp, &mut ()).map_err(|e| e.to_string())
843}
844
845fn bind_all(interp: &mut Interpreter<()>, bindings: &[(Arc<str>, Value)]) {
846    for (n, v) in bindings {
847        interp.define_global(n.clone(), v.clone());
848    }
849}
850
851/// `receive`'s arguments: a pattern (default `:_`), a timeout in ms (default
852/// forever, `nil` too), and the value to answer on timeout (default nil).
853fn receive_args(args: &[Value], span: Span) -> Result<(Value, Option<u64>, Value), EvalError> {
854    let pattern = args
855        .first()
856        .cloned()
857        .unwrap_or_else(|| kw(pattern::WILDCARD));
858    let timeout = match args.get(1) {
859        None | Some(Value::Nil) => None,
860        Some(Value::Int(n)) if *n >= 0 => Some(n.unsigned_abs()),
861        Some(other) => {
862            return Err(refuse(
863                "receive",
864                format!(
865                    "the timeout is milliseconds (an Int >= 0) or nil, got {}",
866                    other.type_name()
867                ),
868                span,
869            ))
870        }
871    };
872    let default = args.get(2).cloned().unwrap_or(Value::Nil);
873    Ok((pattern, timeout, default))
874}
875
876/// Which side of the node a word was installed for.
877#[derive(Clone)]
878enum Side {
879    /// The program. Holds the node, so the node lives exactly as long as the
880    /// interpreter that runs the program.
881    Root(Arc<Node>),
882    /// A child. Weak, because the node holds the child's interpreter and a
883    /// strong reference back would be a cycle that frees nothing.
884    Child(Weak<Node>),
885}
886
887impl Side {
888    fn node(&self, word: &str, span: Span) -> Result<Arc<Node>, EvalError> {
889        match self {
890            Side::Root(n) => Ok(n.clone()),
891            Side::Child(w) => w
892                .upgrade()
893                .ok_or_else(|| refuse(word, "the process's node has shut down", span)),
894        }
895    }
896}
897
898/// Who called a word.
899#[derive(Clone, Copy, PartialEq, Eq)]
900enum Who {
901    Root,
902    Child(Pid),
903}
904
905impl Who {
906    fn pid(self) -> Pid {
907        match self {
908            Who::Root => ROOT,
909            Who::Child(p) => p,
910        }
911    }
912}
913
914type Word = fn(&Node, Who, &[Value], Span) -> Result<Value, EvalError>;
915
916/// The words, with the arity each is registered under.
917const WORDS: &[(&str, Arity, Word)] = &[
918    ("spawn", Arity::Range(1, 3), w_spawn),
919    ("spawn_link", Arity::Range(1, 3), w_spawn_link),
920    ("spawn_server", Arity::Range(2, 3), w_spawn_server),
921    ("send", Arity::Exact(2), w_send),
922    ("self", Arity::Exact(0), w_self),
923    ("alive?", Arity::Exact(1), w_alive),
924    ("link", Arity::Exact(1), w_link),
925    ("unlink", Arity::Exact(1), w_unlink),
926    ("monitor", Arity::Exact(1), w_monitor),
927    ("demonitor", Arity::Exact(1), w_demonitor),
928    ("exit", Arity::Exact(2), w_exit),
929    ("trap_exit", Arity::Exact(1), w_trap_exit),
930    ("supervisor", Arity::Range(2, 3), w_supervisor),
931    ("children", Arity::Exact(1), w_children),
932    ("defmigrate", Arity::Range(2, 3), w_defmigrate),
933    ("def_hash", Arity::Exact(1), w_def_hash),
934];
935
936const CARRIED: &[(&str, Arity, Word, Word)] = &[
937    ("evict", Arity::Exact(3), w_evict, w_evict_provided),
938    ("accept", Arity::Exact(2), w_accept, w_accept_provided),
939];
940
941/// Bind the process words into the interpreter that runs a program, against
942/// `node`. The interpreter becomes the node's root.
943pub fn install<H: 'static>(interp: &mut Interpreter<H>, node: Arc<Node>) {
944    let carrying = node.config.carrying();
945    let rooted = Arc::new(());
946    *lock(&node.rooted) = Some(Arc::downgrade(&rooted));
947    let side = Side::Root(node);
948    for (name, arity, word) in WORDS {
949        register(interp, &side, name, *arity, *word);
950    }
951    carry(interp, &side, carrying);
952    // The root never parks: its `receive` drives the other processes until
953    // it has an answer, so it is an ordinary function on either executor.
954    let s = side.clone();
955    interp.register_fn(
956        "receive",
957        Arity::Range(0, 3),
958        move |args: &[Value], _h: &mut H, span| {
959            let _rooted = &rooted;
960            let node = s.node("receive", span)?;
961            node.receive_root(args, span)
962        },
963    );
964}
965
966fn carry<H: 'static>(interp: &mut Interpreter<H>, side: &Side, carrying: Carrying) {
967    for (name, arity, word, provided) in CARRIED {
968        if !carrying.carrier {
969            if interp.lookup_global(name).is_some() {
970                register_as(interp, side, name, (*name).to_string(), *arity, w_uncarried);
971            }
972            continue;
973        }
974        if carrying.withheld {
975            refuse_as(interp, side, name);
976        } else {
977            register_as(interp, side, name, (*name).to_string(), *arity, *word);
978        }
979        if carrying.scoped {
980            register_as(interp, side, name, provided_name(name), *arity, *provided);
981        }
982    }
983}
984
985fn refuse_as<H: 'static>(interp: &mut Interpreter<H>, side: &Side, name: &'static str) {
986    let who = match side {
987        Side::Root(_) => "this run's frame",
988        Side::Child(_) => "this process's frame",
989    };
990    interp.register_fn(name, Arity::Any, move |_: &[Value], _h: &mut H, span| {
991        Err(breach(name, Capability::CommScavenger, who, span))
992    });
993}
994
995fn register_as<H: 'static>(
996    interp: &mut Interpreter<H>,
997    side: &Side,
998    name: &'static str,
999    bound: String,
1000    arity: Arity,
1001    word: Word,
1002) {
1003    bind_word(interp, side, name, bound, arity, word, false);
1004}
1005
1006fn register<H: 'static>(
1007    interp: &mut Interpreter<H>,
1008    side: &Side,
1009    name: &'static str,
1010    arity: Arity,
1011    word: Word,
1012) {
1013    bind_word(interp, side, name, name.to_string(), arity, word, true);
1014}
1015
1016fn bind_word<H: 'static>(
1017    interp: &mut Interpreter<H>,
1018    side: &Side,
1019    name: &'static str,
1020    bound: String,
1021    arity: Arity,
1022    word: Word,
1023    taped: bool,
1024) {
1025    let side = side.clone();
1026    interp.register_fn(bound, arity, move |args: &[Value], _h: &mut H, span| {
1027        let node = side.node(name, span)?;
1028        let _pass = node.pass(name, span)?;
1029        let who = match side {
1030            Side::Root(_) => Who::Root,
1031            Side::Child(_) => Who::Child(node.current()),
1032        };
1033        let out = word(&node, who, args, span);
1034        if let (true, Who::Child(pid)) = (taped, who) {
1035            node.record(pid, name, &out);
1036        }
1037        node.poke();
1038        out
1039    });
1040}
1041
1042/// The child side of the words, on the prototype every child forks.
1043fn install_child(proto: &mut Interpreter<()>, me: &Weak<Node>, carrying: Carrying) {
1044    let side = Side::Child(me.clone());
1045    for (name, arity, word) in WORDS {
1046        register(proto, &side, name, *arity, *word);
1047    }
1048    carry(proto, &side, carrying);
1049
1050    // `receive` in a child parks. The two phases are the guarantee
1051    // `mailbox.rs` describes: readiness MATCHES (`pattern::first_match` takes
1052    // the mailbox by shared reference) and only `call` takes. Readiness also
1053    // notes the wait — what it waits for and until when — so the scheduler can
1054    // wake it; that bookkeeping touches `waits`, never a mailbox.
1055    let (r, c) = (me.clone(), me.clone());
1056    proto.register_awaitable_fn(
1057        "receive",
1058        Arity::Range(0, 3),
1059        move |args: &[Value], _h: &()| {
1060            let Some(node) = r.upgrade() else { return true };
1061            let Ok(_pass) = node.pass("receive", Span::synthetic()) else {
1062                return true;
1063            };
1064            let Ok((pattern, timeout, _)) = receive_args(args, Span::synthetic()) else {
1065                // Let the call raise the argument error.
1066                return true;
1067            };
1068            let mut st = node.st();
1069            let pid = st.running();
1070            let held = st.held_at_receive(pid);
1071            if !held
1072                && st
1073                    .boxes
1074                    .get(&pid)
1075                    .is_some_and(|b| pattern::first_match(b, &pattern).is_some())
1076            {
1077                return true;
1078            }
1079            let now = st.clock.now();
1080            let deadline = match st.waits.get(&pid) {
1081                Some(Wait::Receive { deadline, .. }) => *deadline,
1082                _ => timeout.map(|t| now.saturating_add(t)),
1083            };
1084            st.waits.insert(pid, Wait::Receive { pattern, deadline });
1085            !held && deadline.is_some_and(|d| now >= d)
1086        },
1087        move |args: &[Value], _h: &mut (), span| {
1088            let node = Side::Child(c.clone()).node("receive", span)?;
1089            let _pass = node.pass("receive", span)?;
1090            let (pattern, _, default) = receive_args(args, span)?;
1091            let out = {
1092                let mut st = node.st();
1093                let pid = st.running();
1094                st.waits.remove(&pid);
1095                let taken = st.boxes.get_mut(&pid).and_then(|b| {
1096                    let i = pattern::first_match(b, &pattern)?;
1097                    b.remove(i)
1098                });
1099                node.journal(&mut st, pid, taken.as_ref());
1100                Ok(taken.unwrap_or(default))
1101            };
1102            node.record(node.current(), "receive", &out);
1103            out
1104        },
1105    );
1106
1107    // A server's boundary. Not ready while a migration is pending for this
1108    // process, so a busy server still stops here for the swap; the message
1109    // that was waiting stays in the mailbox for the new code.
1110    let (r, c) = (me.clone(), me.clone());
1111    proto.register_awaitable_fn(
1112        SERVER_NEXT,
1113        Arity::Exact(1),
1114        move |_args: &[Value], _h: &()| {
1115            let Some(node) = r.upgrade() else { return true };
1116            let Ok(_pass) = node.pass(SERVER_NEXT, Span::synthetic()) else {
1117                return true;
1118            };
1119            let mut st = node.st();
1120            let pid = st.running();
1121            let opens = st.opens_dispatch(pid);
1122            let ready = !opens
1123                && st.boxes.get(&pid).is_some_and(|b| !b.is_empty())
1124                && !st.held_at_boundary(pid);
1125            if !ready {
1126                st.waits.insert(pid, Wait::ServerNext);
1127            }
1128            ready
1129        },
1130        move |args: &[Value], _h: &mut (), span| {
1131            let node = Side::Child(c.clone()).node(SERVER_NEXT, span)?;
1132            let _pass = node.pass(SERVER_NEXT, span)?;
1133            let mut st = node.st();
1134            let pid = st.running();
1135            st.waits.remove(&pid);
1136            let taken = st.boxes.get_mut(&pid).and_then(VecDeque::pop_front);
1137            node.journal(&mut st, pid, taken.as_ref());
1138            let msg = taken.unwrap_or(Value::Nil);
1139            // A boundary: the tape restarts here, from this state, so a
1140            // long-lived server's trace is its last message, not its life.
1141            if let Some(info) = st.info.get_mut(&pid) {
1142                info.snapshot = Some(args.first().cloned().unwrap_or(Value::Nil));
1143                info.armed = false;
1144            }
1145            st.tapes.insert(pid, Tape::default());
1146            drop(st);
1147            let out = Ok(msg);
1148            node.record(pid, SERVER_NEXT, &out);
1149            out
1150        },
1151    );
1152}
1153
1154/// Rebind every host name the prototype binds to a recorder, so a child's
1155/// host reads land on its tape in order.
1156fn record_host_names(proto: &mut Interpreter<()>, me: &Weak<Node>) {
1157    for name in host_names() {
1158        let Some(original @ Value::NativeFn(_)) = proto.lookup_global(name) else {
1159            continue;
1160        };
1161        let Value::NativeFn(nf) = &original else {
1162            continue;
1163        };
1164        let arity = nf.arity;
1165        let wrapper = format!("%host:{name}");
1166        let me = me.clone();
1167        proto.register_higher_order_fn(
1168            wrapper.clone(),
1169            arity,
1170            move |args: &[Value], h: &mut (), caller: &tatara_lisp_eval::ffi::Caller<()>, span| {
1171                let out = caller.apply_value(&original, args.to_vec(), h, span);
1172                if let Some(node) = me.upgrade() {
1173                    let pid = node.current();
1174                    if pid != ROOT {
1175                        node.record(pid, name, &out);
1176                    }
1177                }
1178                out
1179            },
1180        );
1181        if let Some(w) = proto.lookup_global(&wrapper) {
1182            proto.define_global(name, w);
1183        }
1184    }
1185}
1186
1187/// Every host name blue has, from the closed capability universe.
1188#[must_use]
1189pub fn host_names() -> Vec<&'static str> {
1190    blue_lang_waku::capability::Capability::host_effects()
1191        .into_iter()
1192        .flat_map(blue_lang_waku::capability::Capability::names)
1193        .collect()
1194}
1195
1196fn unclassed(p: &mut Proc<()>, why: &str) -> Slice {
1197    let reason = ExitReason::Crashed(format!("the OS refused the idle class: {why}"));
1198    p.state = ProcState::Exited(reason.clone());
1199    Slice::Exited(reason)
1200}
1201
1202fn note_slice(st: &mut NodeState, pid: Pid, out: &Slice, burned: usize) {
1203    let outcome = match out {
1204        Slice::Yielded => "yielded",
1205        Slice::Blocked => "blocked",
1206        Slice::Done(_) => "done",
1207        Slice::Exited(_) => "exited",
1208    };
1209    if st.schedule.len() == SCHEDULE_RING {
1210        st.schedule.pop_front();
1211    }
1212    st.schedule.push_back(SliceRecord {
1213        pid: pid.0,
1214        outcome: outcome.to_string(),
1215        burned,
1216    });
1217}
1218
1219fn fly(me: &Weak<Node>, pid: Pid, queue: &mpsc::Receiver<Proc<()>>, radio: &mpsc::Sender<Flight>) {
1220    let lowered = scheduling::lower();
1221    scheduling::board(pid);
1222    while let Ok(mut p) = queue.recv() {
1223        let Some(node) = me.upgrade() else { break };
1224        let before = p.vm.burned();
1225        let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| match &lowered {
1226            Ok(()) => {
1227                let out = node.slice_held(&mut p);
1228                node.dispatch_held(&mut p, out)
1229            }
1230            Err(why) => unclassed(&mut p, why),
1231        }));
1232        let flight = match ran {
1233            Ok(out) => {
1234                let burned = p.vm.burned().saturating_sub(before);
1235                Flight::Landed(p, out, burned)
1236            }
1237            Err(panic) => Flight::Fell(panic),
1238        };
1239        drop(node);
1240        if radio.send(flight).is_err() {
1241            break;
1242        }
1243    }
1244}
1245
1246/// How a root wait ended.
1247enum Driven<T> {
1248    Done(T),
1249    TimedOut,
1250    Deadlock(Vec<Pid>),
1251}
1252
1253impl Node {
1254    /// A node with no processes.
1255    #[must_use]
1256    pub fn new(config: NodeConfig) -> Arc<Self> {
1257        let clock = config.clock;
1258        let (radio, flights) = mpsc::channel();
1259        Arc::new_cyclic(|me| Self {
1260            config,
1261            me: me.clone(),
1262            state: Mutex::new(NodeState::new(clock)),
1263            sched: Mutex::new(Vec::new()),
1264            proto: OnceLock::new(),
1265            lanes: Mutex::new(BTreeMap::new()),
1266            flights: Mutex::new(flights),
1267            radio,
1268            gate: Mutex::new(Gate::default()),
1269            gated: Condvar::new(),
1270            rooted: Mutex::new(None),
1271        })
1272    }
1273
1274    fn st(&self) -> MutexGuard<'_, NodeState> {
1275        lock(&self.state)
1276    }
1277
1278    fn current(&self) -> Pid {
1279        self.st().running()
1280    }
1281
1282    fn pass(&self, word: &str, span: Span) -> Result<Option<Pass<'_>>, EvalError> {
1283        if scheduling::lane().is_none() {
1284            return Ok(None);
1285        }
1286        let mut g = lock(&self.gate);
1287        while g.driving == 0 {
1288            let rooted = lock(&self.rooted)
1289                .as_ref()
1290                .is_some_and(|w| w.strong_count() > 0);
1291            if !rooted {
1292                return Err(refuse(word, "the process's node has shut down", span));
1293            }
1294            g = self
1295                .gated
1296                .wait_timeout(g, Duration::from_millis(100))
1297                .map_or_else(|e| e.into_inner().0, |(g, _)| g);
1298        }
1299        g.inside += 1;
1300        Ok(Some(Pass(self)))
1301    }
1302
1303    fn driving(&self) -> Driving<'_> {
1304        lock(&self.gate).driving += 1;
1305        self.gated.notify_all();
1306        Driving(self)
1307    }
1308
1309    fn poke(&self) {
1310        if scheduling::lane().is_some() {
1311            let _ = self.radio.send(Flight::Poked);
1312        }
1313    }
1314
1315    fn proto(&self) -> Arc<Interpreter<()>> {
1316        self.proto
1317            .get_or_init(|| {
1318                let mut p = (self.config.child)();
1319                install_child(&mut p, &self.me, self.config.carrying());
1320                record_host_names(&mut p, &self.me);
1321                Arc::new(p)
1322            })
1323            .clone()
1324    }
1325
1326    fn builder(
1327        &self,
1328        bindings: &[(Arc<str>, Value)],
1329    ) -> Arc<dyn Fn() -> Interpreter<()> + Send + Sync> {
1330        let proto = self.proto();
1331        let bindings = bindings.to_vec();
1332        Arc::new(move || {
1333            let mut i = proto.fork();
1334            bind_all(&mut i, &bindings);
1335            i
1336        })
1337    }
1338
1339    fn record(&self, pid: Pid, call: &str, out: &Result<Value, EvalError>) {
1340        let max = self.config.max_tape;
1341        let mut st = self.st();
1342        let Some(tape) = st.tapes.get_mut(&pid) else {
1343            return;
1344        };
1345        if tape.truncated {
1346            return;
1347        }
1348        if tape.entries.len() >= max {
1349            // The whole tape or nothing: a replay of a suffix would run the
1350            // process from a state it never had.
1351            tape.truncated = true;
1352            tape.entries = Vec::new();
1353            tape.carried = None;
1354            return;
1355        }
1356        tape.entries.push(TapeEntry::of(call, out));
1357    }
1358
1359    /// The process running now took `msg`, if `taken`.
1360    fn journal(&self, st: &mut NodeState, pid: Pid, taken: Option<&Value>) {
1361        if let (true, Some(msg)) = (self.config.journals(), taken) {
1362            st.note_taken(pid, msg, self.config.journal);
1363        }
1364    }
1365
1366    /// `pid` runs now: the scheduler's current process and the host ledger's.
1367    fn switch(&self, st: &mut NodeState, pid: Pid) {
1368        st.current = pid;
1369        self.config.host_io.enter(pid);
1370    }
1371
1372    /// `f` charged to the account `held` carries, held to `bound`, and
1373    /// whether it passed the bound; run uncharged with no meter.
1374    fn charged<T>(&self, held: &mut i64, bound: Option<u64>, f: impl FnOnce() -> T) -> (T, bool) {
1375        let Some(meter) = self.config.meter else {
1376            return (f(), false);
1377        };
1378        let mut f = Some(f);
1379        let mut out = None;
1380        let over = (meter.charged)(held, bound, &mut || out = f.take().map(|f| f()));
1381        match (out, f) {
1382            (Some(v), _) => (v, over),
1383            (None, Some(f)) => (f(), false),
1384            (None, None) => unreachable!("a closure that ran left its value"),
1385        }
1386    }
1387
1388    /// `f` charged to no process, and the bytes it left live.
1389    fn apart<T>(&self, f: impl FnOnce() -> T) -> (T, i64) {
1390        let mut held = 0;
1391        let (value, _) = self.charged(&mut held, None, f);
1392        (value, held)
1393    }
1394
1395    /// `bytes` now belong to `to`: a message copied into its mailbox.
1396    fn credit(&self, to: Pid, bytes: i64) {
1397        let Some(meter) = self.config.meter.filter(|_| bytes != 0) else {
1398            return;
1399        };
1400        let mut st = self.st();
1401        if to == st.running() {
1402            drop(st);
1403            (meter.charge)(bytes);
1404        } else if to == ROOT {
1405            st.root_due += bytes;
1406        } else if let Some(info) = st.info.get_mut(&to) {
1407            info.held += bytes;
1408        }
1409    }
1410
1411    /// Charge the root what its children delivered to it while it waited.
1412    fn settle_root(&self) {
1413        let due = std::mem::take(&mut self.st().root_due);
1414        if let (Some(meter), true) = (self.config.meter, due != 0) {
1415            (meter.charge)(due);
1416        }
1417    }
1418
1419    /// Give `p` one slice charged to its own account; past its bound it exits
1420    /// [`ExitReason::memory_exhausted`], whatever the slice did.
1421    fn slice_held(&self, p: &mut Proc<()>) -> Slice {
1422        let pid = p.pid;
1423        let (mut held, bound) = self
1424            .st()
1425            .info
1426            .get(&pid)
1427            .map_or((0, None), |i| (i.held, i.bound));
1428        let Some(limit) = bound else {
1429            return p.slice(&mut ());
1430        };
1431        let start = held;
1432        let (out, over) = self.charged(&mut held, bound, || p.slice(&mut ()));
1433        if let Some(info) = self.st().info.get_mut(&pid) {
1434            info.held += held - start;
1435        }
1436        if !over {
1437            return out;
1438        }
1439        let reason = ExitReason::memory_exhausted(limit);
1440        p.state = ProcState::Exited(reason.clone());
1441        Slice::Exited(reason)
1442    }
1443
1444    fn dispatch_held(&self, p: &mut Proc<()>, out: Slice) -> Slice {
1445        if !matches!(out, Slice::Yielded | Slice::Blocked) {
1446            return out;
1447        }
1448        let burned = p.vm.burned();
1449        let mut st = self.st();
1450        let boundary = matches!(st.waits.get(&p.pid), Some(Wait::ServerNext));
1451        let Some(info) = st.info.get_mut(&p.pid) else {
1452            return out;
1453        };
1454        let Some(limit) = info.dispatch else {
1455            return out;
1456        };
1457        if info
1458            .dispatch_from
1459            .is_some_and(|from| burned.saturating_sub(from) > limit)
1460        {
1461            info.dispatch_from = None;
1462            info.armed = false;
1463            info.kept.clone_from(&info.snapshot);
1464            let kept = info.kept.clone();
1465            drop(st);
1466            if let Some(state) = kept {
1467                p.interp.define_global("%state", state);
1468            }
1469            let reason = ExitReason::dispatch_exhausted(limit);
1470            p.state = ProcState::Exited(reason.clone());
1471            return Slice::Exited(reason);
1472        }
1473        if boundary {
1474            info.dispatch_from = info.armed.then_some(burned);
1475        }
1476        out
1477    }
1478
1479    fn slice_idle(&self, p: &mut Proc<()>) -> Slice {
1480        let ran = std::thread::scope(|s| {
1481            std::thread::Builder::new()
1482                .name("blue-idle".to_string())
1483                .stack_size(scheduling::LANE_STACK)
1484                .spawn_scoped(s, || scheduling::lower().map(|()| self.slice_held(p)))
1485                .map_err(|e| e.to_string())
1486                .and_then(|lane| {
1487                    lane.join()
1488                        .unwrap_or_else(|panic| std::panic::resume_unwind(panic))
1489                })
1490        });
1491        ran.unwrap_or_else(|why| unclassed(p, &why))
1492    }
1493
1494    fn launch(&self, p: Proc<()>) {
1495        let pid = p.pid;
1496        self.st().aloft.insert(pid);
1497        let mut p = p;
1498        let mut why = String::new();
1499        for _ in 0..2 {
1500            let lane = {
1501                let mut lanes = lock(&self.lanes);
1502                match lanes.get(&pid) {
1503                    Some(lane) => Ok(lane.clone()),
1504                    None => self.lane(pid).inspect(|lane| {
1505                        lanes.insert(pid, lane.clone());
1506                    }),
1507                }
1508            };
1509            match lane {
1510                Ok(lane) => match lane.send(p) {
1511                    Ok(()) => return,
1512                    Err(mpsc::SendError(back)) => {
1513                        lock(&self.lanes).remove(&pid);
1514                        why = "its lane closed".to_string();
1515                        p = back;
1516                    }
1517                },
1518                Err(e) => {
1519                    why = e;
1520                    break;
1521                }
1522            }
1523        }
1524        let out = unclassed(&mut p, &why);
1525        let _ = self.radio.send(Flight::Landed(p, out, 0));
1526    }
1527
1528    fn lane(&self, pid: Pid) -> Result<mpsc::Sender<Proc<()>>, String> {
1529        let (work, queue) = mpsc::channel::<Proc<()>>();
1530        let me = self.me.clone();
1531        let radio = self.radio.clone();
1532        std::thread::Builder::new()
1533            .name(format!("blue-idle-{}", pid.0))
1534            .stack_size(scheduling::LANE_STACK)
1535            .spawn(move || fly(&me, pid, &queue, &radio))
1536            .map(|_| work)
1537            .map_err(|e| e.to_string())
1538    }
1539
1540    fn land(&self, sched: &mut Vec<Proc<()>>) -> bool {
1541        let mut any = false;
1542        loop {
1543            let next = lock(&self.flights).try_recv();
1544            let Ok(flight) = next else { return any };
1545            any |= self.touch_down(sched, flight);
1546        }
1547    }
1548
1549    fn await_flight(&self, sched: &mut Vec<Proc<()>>, limit: Option<Duration>) {
1550        let next = {
1551            let flights = lock(&self.flights);
1552            match limit {
1553                Some(d) => flights.recv_timeout(d).ok(),
1554                None => flights.recv().ok(),
1555            }
1556        };
1557        if let Some(flight) = next {
1558            self.touch_down(sched, flight);
1559        }
1560        self.land(sched);
1561    }
1562
1563    fn touch_down(&self, sched: &mut Vec<Proc<()>>, flight: Flight) -> bool {
1564        let (p, out, burned) = match flight {
1565            Flight::Landed(p, out, burned) => (p, out, burned),
1566            Flight::Poked => return false,
1567            Flight::Fell(panic) => std::panic::resume_unwind(panic),
1568        };
1569        let pid = p.pid;
1570        let (known, doomed, recalled) = {
1571            let mut st = self.st();
1572            st.aloft.remove(&pid);
1573            note_slice(&mut st, pid, &out, burned);
1574            (
1575                st.info.contains_key(&pid),
1576                st.doomed.remove(&pid),
1577                st.recalled.remove(&pid),
1578            )
1579        };
1580        if !known {
1581            lock(&self.lanes).remove(&pid);
1582            return true;
1583        }
1584        let at = sched.partition_point(|q| q.pid < pid);
1585        sched.insert(at, p);
1586        if recalled {
1587            self.restart(sched, pid);
1588        } else {
1589            match out {
1590                Slice::Yielded | Slice::Blocked => {
1591                    if let Some(reason) = doomed {
1592                        self.kill(sched, pid, reason);
1593                    }
1594                }
1595                out => self.conclude(sched, pid, out),
1596            }
1597        }
1598        self.adopt(sched);
1599        self.process_signals(sched);
1600        true
1601    }
1602
1603    fn conclude(&self, sched: &mut Vec<Proc<()>>, pid: Pid, out: Slice) {
1604        match out {
1605            Slice::Done(v) => self.on_exit(sched, pid, &ExitReason::Normal, kw("normal"), Some(v)),
1606            Slice::Exited(r) => {
1607                let rv = reason_value(&r);
1608                self.on_exit(sched, pid, &r, rv, None);
1609            }
1610            Slice::Yielded | Slice::Blocked => {}
1611        }
1612    }
1613
1614    fn detached(&self, pid: Pid) -> bool {
1615        self.config.idle_lane == IdleLane::Apart
1616            && self
1617                .st()
1618                .info
1619                .get(&pid)
1620                .is_some_and(|i| i.scheduling == Scheduling::Idle)
1621    }
1622
1623    /// Compile `code` and queue a process running it, held to the bound of
1624    /// the account that spawns it. Building it is charged to no one.
1625    fn start(
1626        &self,
1627        code: Code,
1628        kind: Kind,
1629        def_name: Option<String>,
1630        sup: Option<Pid>,
1631        scheduling: Scheduling,
1632    ) -> Result<Pid, String> {
1633        let bound = self.config.meter.and_then(|m| (m.bound)());
1634        self.apart(|| self.start_unmetered(code, kind, def_name, sup, bound, scheduling))
1635            .0
1636    }
1637
1638    fn start_unmetered(
1639        &self,
1640        code: Code,
1641        kind: Kind,
1642        def_name: Option<String>,
1643        sup: Option<Pid>,
1644        bound: Option<u64>,
1645        scheduling: Scheduling,
1646    ) -> Result<Pid, String> {
1647        let chunk = Arc::new(compile(&code.program())?);
1648        let build = self.builder(&code.bindings);
1649        let dispatch = matches!(kind, Kind::Server)
1650            .then_some(self.config.dispatch_steps)
1651            .flatten();
1652        let initial = match kind {
1653            Kind::Server => code
1654                .bindings
1655                .iter()
1656                .find(|(n, _)| &**n == "%state")
1657                .map(|(_, v)| v.clone()),
1658            _ => None,
1659        };
1660        let mut st = self.st();
1661        let pid = Pid(st.next_pid);
1662        st.next_pid += 1;
1663        let proc = Proc::new(pid, chunk.clone(), self.config.budget, build);
1664        let mut info = Info::new(kind);
1665        info.code = code;
1666        info.chunk = Some(chunk);
1667        info.def_name = def_name;
1668        info.sup = sup;
1669        info.snapshot.clone_from(&initial);
1670        info.entered.clone_from(&initial);
1671        info.initial = initial;
1672        info.bound = bound;
1673        info.scheduling = scheduling;
1674        info.dispatch = dispatch;
1675        st.info.insert(pid, info);
1676        st.boxes.insert(pid, VecDeque::new());
1677        st.tapes.insert(pid, Tape::default());
1678        st.pending.push(proc);
1679        Ok(pid)
1680    }
1681
1682    // ── the scheduler ──────────────────────────────────────────────────────
1683
1684    fn adopt(&self, sched: &mut Vec<Proc<()>>) {
1685        let pending = std::mem::take(&mut self.st().pending);
1686        sched.extend(pending);
1687    }
1688
1689    fn has_work(&self, sched: &[Proc<()>]) -> bool {
1690        let mut st = self.st();
1691        sched
1692            .iter()
1693            .any(|p| p.is_runnable() || (p.state.is_blocked() && st.is_ready(p.pid)))
1694    }
1695
1696    /// One pass over the run queue: wake what can proceed, give every runnable
1697    /// process one quantum, deliver the exits and signals that produced.
1698    fn round(&self, sched: &mut Vec<Proc<()>>) {
1699        {
1700            let mut st = self.st();
1701            for p in sched.iter_mut() {
1702                if p.state.is_blocked() && st.is_ready(p.pid) {
1703                    p.state = ProcState::Runnable;
1704                }
1705            }
1706        }
1707        let mut end = sched.len();
1708        let mut idx = 0;
1709        while idx < end {
1710            if !sched[idx].is_runnable() {
1711                idx += 1;
1712                continue;
1713            }
1714            let pid = sched[idx].pid;
1715            if self.detached(pid) {
1716                self.launch(sched.remove(idx));
1717                end -= 1;
1718                continue;
1719            }
1720            self.switch(&mut self.st(), pid);
1721            let idle = self
1722                .st()
1723                .info
1724                .get(&pid)
1725                .is_some_and(|i| i.scheduling == Scheduling::Idle);
1726            let before = sched[idx].vm.burned();
1727            let out = if idle {
1728                self.slice_idle(&mut sched[idx])
1729            } else {
1730                self.slice_held(&mut sched[idx])
1731            };
1732            let out = self.dispatch_held(&mut sched[idx], out);
1733            let burned = sched[idx].vm.burned().saturating_sub(before);
1734            {
1735                let mut st = self.st();
1736                self.switch(&mut st, ROOT);
1737                note_slice(&mut st, pid, &out, burned);
1738            }
1739            self.conclude(sched, pid, out);
1740            self.adopt(sched);
1741            self.process_signals(sched);
1742            idx += 1;
1743        }
1744        self.reap(sched);
1745    }
1746
1747    /// Drop finished processes nothing can restart, so a long-running node's
1748    /// memory follows its live processes, not every process it ever ran.
1749    fn reap(&self, sched: &mut Vec<Proc<()>>) {
1750        let st = self.st();
1751        sched.retain(|p| {
1752            p.is_runnable()
1753                || p.state.is_blocked()
1754                || st.info.get(&p.pid).is_some_and(|i| i.sup.is_some())
1755        });
1756        let mut lanes = lock(&self.lanes);
1757        if !lanes.is_empty() {
1758            lanes.retain(|pid, _| st.aloft.contains(pid) || sched.iter().any(|p| p.pid == *pid));
1759        }
1760    }
1761
1762    fn process_signals(&self, sched: &mut Vec<Proc<()>>) {
1763        loop {
1764            let next = self.st().signals.pop_front();
1765            let Some((from, to, reason)) = next else {
1766                break;
1767            };
1768            self.signal(sched, from, to, reason);
1769        }
1770    }
1771
1772    /// Deliver an exit signal from `from` to `to`.
1773    fn signal(&self, sched: &mut Vec<Proc<()>>, from: Pid, to: Pid, reason: Value) {
1774        let normal = is_keyword(&reason, "normal");
1775        let kill = is_keyword(&reason, "kill");
1776        let mut st = self.st();
1777        let Some(info) = st.info.get(&to) else { return };
1778        let trapping = info.trap_exit && !kill;
1779        let is_sup = matches!(info.kind, Kind::Supervisor(_));
1780        if trapping {
1781            st.deliver(to, Value::list([kw("exit"), int(from.0), reason]));
1782            return;
1783        }
1784        if normal {
1785            return;
1786        }
1787        if to == ROOT {
1788            st.root_signal = Some((from, reason));
1789            return;
1790        }
1791        drop(st);
1792        let reason = if kill { kw("killed") } else { reason };
1793        if is_sup {
1794            self.shutdown_sup(sched, to, reason);
1795        } else {
1796            self.kill(sched, to, reason);
1797        }
1798    }
1799
1800    fn kill(&self, sched: &mut Vec<Proc<()>>, pid: Pid, reason: Value) {
1801        let text = render(&reason);
1802        let killed = ExitReason::Killed(text);
1803        let alive = match sched.iter_mut().find(|p| p.pid == pid) {
1804            Some(p) if p.is_runnable() || p.state.is_blocked() => {
1805                p.state = ProcState::Exited(killed.clone());
1806                true
1807            }
1808            _ => false,
1809        };
1810        if alive {
1811            self.on_exit(sched, pid, &killed, reason, None);
1812        } else {
1813            let mut st = self.st();
1814            if st.aloft.contains(&pid) {
1815                st.doomed.entry(pid).or_insert(reason);
1816            }
1817        }
1818    }
1819
1820    /// A process ended: write its trace, tell its monitors and links, and let
1821    /// its supervisor decide.
1822    fn on_exit(
1823        &self,
1824        sched: &mut Vec<Proc<()>>,
1825        pid: Pid,
1826        reason: &ExitReason,
1827        value: Value,
1828        _result: Option<Value>,
1829    ) {
1830        let mut st = self.st();
1831        if !st.info.contains_key(&pid) {
1832            return;
1833        }
1834        let trace = reason
1835            .is_abnormal()
1836            .then(|| self.trace_of(&mut st, pid, reason, &value));
1837        let receipt = self.receipt_of(&st, sched, pid, reason);
1838        let (links, monitors, sup) = {
1839            let info = st.info.get_mut(&pid).expect("checked above");
1840            (
1841                std::mem::take(&mut info.links),
1842                std::mem::take(&mut info.monitors),
1843                info.sup,
1844            )
1845        };
1846        for l in &links {
1847            if let Some(li) = st.info.get_mut(l) {
1848                li.links.remove(&pid);
1849            }
1850        }
1851        for (r, watcher) in monitors {
1852            st.deliver(
1853                watcher,
1854                Value::list([kw("down"), int(r), int(pid.0), value.clone()]),
1855            );
1856        }
1857        st.waits.remove(&pid);
1858        if let Some(m) = st.migrations.remove(&pid) {
1859            st.outcomes.entry(m.request).or_default().push((
1860                pid,
1861                Err("the process exited before it reached a boundary".into()),
1862            ));
1863        }
1864        let supervised =
1865            sup.filter(|s| matches!(st.info.get(s).map(|i| &i.kind), Some(Kind::Supervisor(_))));
1866        if reason.is_fatal() {
1867            if let Some(s) = supervised {
1868                st.drop_child(s, pid);
1869            }
1870            st.forget(pid);
1871        } else if supervised.is_none() {
1872            st.forget(pid);
1873        } else if let Some(info) = st.info.get_mut(&pid) {
1874            info.live = false;
1875        }
1876        drop(st);
1877
1878        if let (Some(r), Some(sink)) = (&receipt, &self.config.receipts) {
1879            sink(r);
1880        }
1881        if let Some(t) = trace {
1882            // Every crash is kept in memory; a supervised one is also written
1883            // out, because it is the crash a running system recovers from and
1884            // so the one nobody sees happen.
1885            self.keep_trace(t, written(supervised.is_some(), reason));
1886        }
1887        for l in links {
1888            self.signal(sched, pid, l, value.clone());
1889        }
1890        if let (Some(s), true) = (supervised, reason.is_restartable()) {
1891            self.supervise(sched, s, pid);
1892        }
1893    }
1894
1895    fn supervise(&self, sched: &mut Vec<Proc<()>>, sup: Pid, child: Pid) {
1896        let (spec, restarts) = {
1897            let st = self.st();
1898            let Some(Kind::Supervisor(spec)) = st.info.get(&sup).map(|i| i.kind.clone()) else {
1899                return;
1900            };
1901            let restarts = st.info.get(&child).map_or(0, |i| i.restarts);
1902            (spec, restarts)
1903        };
1904        // Intensity is checked on the child that exited, as `Supervisor` does:
1905        // a sibling restarted as collateral has not misbehaved.
1906        if restarts >= spec.max_restarts {
1907            self.shutdown_sup(sched, sup, Value::list([kw("gave_up"), int(child.0)]));
1908            return;
1909        }
1910        let Some(pos) = spec.children.iter().position(|c| *c == child) else {
1911            return;
1912        };
1913        for t in spec.strategy.targets(pos, spec.children.len()) {
1914            self.restart(sched, spec.children[t]);
1915        }
1916    }
1917
1918    /// A fresh incarnation under the same pid: new VM, new interpreter from
1919    /// the current code, an empty mailbox and an empty tape.
1920    fn restart(&self, sched: &mut [Proc<()>], pid: Pid) {
1921        let Some(p) = sched.iter_mut().find(|p| p.pid == pid) else {
1922            let mut st = self.st();
1923            if st.aloft.contains(&pid) {
1924                st.recalled.insert(pid);
1925            }
1926            return;
1927        };
1928        let (chunk, kept) = {
1929            let mut st = self.st();
1930            // The messages in flight were addressed to the incarnation that
1931            // died; replaying them is how a poison message kills a process
1932            // forever.
1933            if let Some(b) = st.boxes.get_mut(&pid) {
1934                b.clear();
1935            }
1936            st.waits.remove(&pid);
1937            st.tapes.insert(pid, Tape::default());
1938            st.journals.remove(&pid);
1939            st.accounts.remove(&pid);
1940            self.config.host_io.take(pid);
1941            st.info.get_mut(&pid).map_or((None, None), |info| {
1942                info.restarts += 1;
1943                info.live = true;
1944                info.held = 0;
1945                info.dispatch_from = None;
1946                info.armed = false;
1947                let kept = info.kept.take();
1948                let state = kept.clone().or_else(|| info.initial.clone());
1949                info.snapshot.clone_from(&state);
1950                info.entered = state;
1951                (info.chunk.clone(), kept)
1952            })
1953        };
1954        if let Some(c) = chunk {
1955            p.chunk = c;
1956        }
1957        p.restart();
1958        if let Some(state) = kept {
1959            p.interp.define_global("%state", state);
1960        }
1961    }
1962
1963    /// Stop a supervisor: its children first (they are not restarted), then
1964    /// the supervisor itself, which its own links and monitors hear about.
1965    fn shutdown_sup(&self, sched: &mut Vec<Proc<()>>, sup: Pid, reason: Value) {
1966        let (children, links, monitors) = {
1967            let mut st = self.st();
1968            let Some(info) = st.info.remove(&sup) else {
1969                return;
1970            };
1971            st.boxes.remove(&sup);
1972            let children = match info.kind {
1973                Kind::Supervisor(s) => s.children,
1974                _ => Vec::new(),
1975            };
1976            for c in &children {
1977                if let Some(ci) = st.info.get_mut(c) {
1978                    ci.sup = None;
1979                }
1980            }
1981            for l in &info.links {
1982                if let Some(li) = st.info.get_mut(l) {
1983                    li.links.remove(&sup);
1984                }
1985            }
1986            (children, info.links, info.monitors)
1987        };
1988        for c in children {
1989            self.kill(sched, c, kw("shutdown"));
1990            // A child that had already finished is not in the run queue to
1991            // kill; it is simply forgotten.
1992            let mut st = self.st();
1993            if st.info.get(&c).is_some_and(|i| i.sup.is_none())
1994                && !st.aloft.contains(&c)
1995                && !sched
1996                    .iter()
1997                    .any(|p| p.pid == c && (p.is_runnable() || p.state.is_blocked()))
1998            {
1999                st.forget(c);
2000            }
2001        }
2002        {
2003            let mut st = self.st();
2004            for (r, watcher) in monitors {
2005                st.deliver(
2006                    watcher,
2007                    Value::list([kw("down"), int(r), int(sup.0), reason.clone()]),
2008                );
2009            }
2010        }
2011        for l in links {
2012            self.signal(sched, sup, l, reason.clone());
2013        }
2014    }
2015
2016    // ── migrations ─────────────────────────────────────────────────────────
2017
2018    /// Apply every pending migration whose process is at a boundary.
2019    ///
2020    /// Answers whether any was applied, so a waiting root checks again.
2021    fn apply_migrations(&self, sched: &mut [Proc<()>]) -> bool {
2022        let pending: Vec<Pid> = self.st().migrations.keys().copied().collect();
2023        let mut applied = false;
2024        for pid in pending {
2025            let Some(idx) = sched.iter().position(|p| p.pid == pid) else {
2026                continue;
2027            };
2028            let (ready, m, kind) = {
2029                let st = self.st();
2030                let Some(info) = st.info.get(&pid) else {
2031                    continue;
2032                };
2033                let p = &sched[idx];
2034                let ready = match info.kind {
2035                    Kind::Server => {
2036                        p.state.is_blocked() && matches!(st.waits.get(&pid), Some(Wait::ServerNext))
2037                    }
2038                    Kind::Plain => p.is_runnable() || p.state.is_blocked(),
2039                    _ => false,
2040                };
2041                (ready, st.migrations.get(&pid).cloned(), info.kind.clone())
2042            };
2043            let (true, Some(m)) = (ready, m) else {
2044                continue;
2045            };
2046            self.switch(&mut self.st(), pid);
2047            let result = self.migrate(&mut sched[idx], &kind, &m);
2048            let mut st = self.st();
2049            self.switch(&mut st, ROOT);
2050            st.migrations.remove(&pid);
2051            st.outcomes
2052                .entry(m.request)
2053                .or_default()
2054                .push((pid, result));
2055            applied = true;
2056        }
2057        applied
2058    }
2059
2060    /// Swap `p`'s code, and for a server its state. Everything is tried on a
2061    /// fork of the process's interpreter first; only a migration that
2062    /// succeeded there is committed, so one that raises leaves the old code
2063    /// running and its state untouched.
2064    fn migrate(&self, p: &mut Proc<()>, kind: &Kind, m: &Migration) -> Result<(), String> {
2065        let depth = self.config.budget.max_depth;
2066        let (old_name, initial) = {
2067            let st = self.st();
2068            let info = st.info.get(&p.pid).ok_or("the process is gone")?;
2069            (info.def_name.clone(), info.initial.clone())
2070        };
2071        let server = matches!(kind, Kind::Server);
2072        if !server && m.transform.is_some() {
2073            return Err(
2074                "a plain process keeps its state in its own frames, so there is no state to \
2075                        transform; run it with spawn_server to migrate state"
2076                    .into(),
2077            );
2078        }
2079
2080        let mut scratch = p.interp.fork();
2081        bind_all(&mut scratch, &m.code.bindings);
2082        run_in(&mut scratch, &m.code.defs, depth)?;
2083        let (state, initial) = if server {
2084            let old = p.interp.lookup_global("%state").unwrap_or(Value::Nil);
2085            match &m.transform {
2086                Some(t) => {
2087                    bind_all(&mut scratch, &t.bindings);
2088                    run_in(&mut scratch, &t.defs, depth)?;
2089                    let apply = |scratch: &mut Interpreter<()>, v: Value| {
2090                        scratch.define_global("%old", v);
2091                        run_in(
2092                            scratch,
2093                            &[code::call(code::symbol("%transform"), &["%old"])],
2094                            depth,
2095                        )
2096                    };
2097                    let state = apply(&mut scratch, old)?;
2098                    let initial = match initial {
2099                        Some(i) => Some(apply(&mut scratch, i)?),
2100                        None => None,
2101                    };
2102                    (Some(state), initial)
2103                }
2104                None => (Some(old), initial),
2105            }
2106        } else {
2107            (None, None)
2108        };
2109        drop(scratch);
2110
2111        // Commit. The definitions already ran once, so they run again here.
2112        bind_all(&mut p.interp, &m.code.bindings);
2113        run_in(&mut p.interp, &m.code.defs, depth)?;
2114        let new_fn = p
2115            .interp
2116            .lookup_global(&m.new_name)
2117            .ok_or_else(|| format!("the migration did not define `{}`", m.new_name))?;
2118        let alias = if server {
2119            "%handler".to_string()
2120        } else {
2121            old_name.clone().unwrap_or_default()
2122        };
2123        if !alias.is_empty() && alias != m.new_name {
2124            p.interp.define_global(alias.as_str(), new_fn);
2125        }
2126        if let Some(s) = &state {
2127            p.interp.define_global("%state", s.clone());
2128        }
2129
2130        // A later restart, trace or migration starts from the new program.
2131        let mut st = self.st();
2132        let info = st.info.get_mut(&p.pid).ok_or("the process is gone")?;
2133        let mut defs = info.code.defs.clone();
2134        defs.extend(m.code.defs.iter().cloned());
2135        if !alias.is_empty() && alias != m.new_name {
2136            defs.push(code::define_value(&alias, code::symbol(&m.new_name)));
2137        }
2138        if server {
2139            // The loop stays last so it is defined after everything it calls.
2140            defs.retain(|d| !SERVER_LOOP.contains(d));
2141            defs.extend(SERVER_LOOP.iter().cloned());
2142        }
2143        let mut new_code = Code {
2144            defs,
2145            bindings: info.code.bindings.clone(),
2146            start: info.code.start.clone(),
2147        };
2148        for (n, v) in &m.code.bindings {
2149            new_code.bind(n, v.clone());
2150        }
2151        if let Some(i) = &initial {
2152            new_code.bind("%state", i.clone());
2153        }
2154        let chunk = Arc::new(compile(&new_code.program())?);
2155        info.code = new_code;
2156        info.chunk = Some(chunk);
2157        info.def_name = Some(m.new_name.clone());
2158        info.initial = initial;
2159        if state.is_some() {
2160            info.snapshot = state;
2161        }
2162        let bindings = info.code.bindings.clone();
2163        st.tapes.insert(p.pid, Tape::default());
2164        drop(st);
2165        p.build = self.builder(&bindings);
2166        Ok(())
2167    }
2168
2169    // ── eviction ───────────────────────────────────────────────────────────
2170
2171    fn tally(&self) {
2172        if self
2173            .config
2174            .frame
2175            .as_ref()
2176            .is_some_and(|f| f.granted.contains(&Capability::CommScavenger))
2177        {
2178            self.config.host_io.tally();
2179        }
2180    }
2181
2182    fn carrier(&self, word: &str, span: Span) -> Result<Carrier, EvalError> {
2183        self.config
2184            .carrier
2185            .ok_or_else(|| refuse(word, "this node has no wire to carry a process", span))
2186    }
2187
2188    fn evictable(&self, pid: Pid) -> Result<(), String> {
2189        let st = self.st();
2190        let info = st
2191            .info
2192            .get(&pid)
2193            .filter(|i| i.live)
2194            .ok_or_else(|| format!("no process {}", pid.0))?;
2195        match info.kind {
2196            Kind::Server | Kind::Plain => {}
2197            Kind::Root => {
2198                return Err(format!(
2199                    "process {} is the program: the door runs it on its own evaluator, the \
2200                     tree-walker under `blue run`, which this node does not hold, so it has no \
2201                     frames to carry; spawn the work as a process and evict that",
2202                    pid.0
2203                ))
2204            }
2205            Kind::Supervisor(_) => {
2206                return Err(format!(
2207                    "process {} is a supervisor: it runs no code of its own to carry; evict its \
2208                     children one at a time",
2209                    pid.0
2210                ))
2211            }
2212        }
2213        if st.held_at_boundary(pid) {
2214            return Err(format!("process {} is already being moved", pid.0));
2215        }
2216        drop(st);
2217        self.reachable_links(pid)
2218    }
2219
2220    fn reachable_links(&self, pid: Pid) -> Result<(), String> {
2221        let st = self.st();
2222        let deaf: Vec<String> = st
2223            .info
2224            .get(&pid)
2225            .map(|i| i.links.iter())
2226            .into_iter()
2227            .flatten()
2228            .filter(|l| !st.info.get(l).is_some_and(|i| i.trap_exit))
2229            .map(|l| l.0.to_string())
2230            .collect();
2231        if deaf.is_empty() {
2232            return Ok(());
2233        }
2234        Err(format!(
2235            "process {} is linked to {}, which does not trap exits; a link cannot follow a \
2236             process to another node, so evict breaks it and tells each partner [:exit, {}, \
2237             :{EVICTED}], which a partner hears only when it traps exits: unlink it first, or \
2238             trap exits in the partner",
2239            pid.0,
2240            deaf.join(", "),
2241            pid.0
2242        ))
2243    }
2244
2245    fn evict(
2246        &self,
2247        pid: Pid,
2248        wire: &Value,
2249        peer: &Value,
2250        loopback: bool,
2251        span: Span,
2252    ) -> Result<Value, EvalError> {
2253        let word = "evict";
2254        let carrier = self.carrier(word, span)?;
2255        self.evictable(pid).map_err(|m| refuse(word, m, span))?;
2256        self.st().evicting.insert(pid);
2257        let parked = self.drive(
2258            |st| {
2259                let info = st.info.get(&pid).filter(|i| i.live);
2260                let boundary = match (info.map(|i| &i.kind), st.waits.get(&pid)) {
2261                    (None, _) => return Some(false),
2262                    (Some(Kind::Server), Some(Wait::ServerNext))
2263                    | (Some(Kind::Plain), Some(Wait::Receive { .. })) => true,
2264                    _ => false,
2265                };
2266                (boundary && !st.aloft.contains(&pid)).then_some(true)
2267            },
2268            None,
2269            word,
2270            span,
2271        );
2272        let moved = match parked {
2273            Ok(Driven::Done(true)) => self.depart_over(pid, wire, peer, loopback, carrier),
2274            Ok(Driven::Done(false)) => Err(format!(
2275                "process {} exited before it reached a message boundary",
2276                pid.0
2277            )),
2278            Ok(Driven::Deadlock(_) | Driven::TimedOut) => Err(format!(
2279                "process {} is waiting {} and nothing can wake it, so it never reached a message \
2280                 boundary",
2281                pid.0,
2282                match self.st().info.get(&pid).map(|i| &i.kind) {
2283                    Some(Kind::Plain) => "outside receive",
2284                    _ => "inside its handler",
2285                }
2286            )),
2287            Err(e) => {
2288                self.st().evicting.remove(&pid);
2289                return Err(e);
2290            }
2291        };
2292        self.st().evicting.remove(&pid);
2293        moved.map_err(|m| refuse(word, m, span))
2294    }
2295
2296    fn image_of(&self, pid: Pid) -> Result<Image, String> {
2297        let state = lock(&self.sched)
2298            .iter()
2299            .find(|p| p.pid == pid)
2300            .and_then(|p| p.interp.lookup_global("%state"))
2301            .unwrap_or(Value::Nil);
2302        let state = copy_data(&state).map_err(|kind| {
2303            format!(
2304                "process {}'s state holds a {kind}, which is not data and does not cross a wire",
2305                pid.0
2306            )
2307        })?;
2308        let st = self.st();
2309        let info = st.info.get(&pid).ok_or("the process is gone")?;
2310        let plain = matches!(info.kind, Kind::Plain);
2311        let mut code = info.code.clone();
2312        if !plain {
2313            code.bind("%state", state);
2314        }
2315        let supervision = match info.sup.and_then(|s| st.info.get(&s)).map(|s| &s.kind) {
2316            Some(Kind::Supervisor(spec)) => Some(Supervision {
2317                max_restarts: spec.max_restarts,
2318                restarts: info.restarts,
2319                initial: info.initial.clone().unwrap_or(Value::Nil),
2320            }),
2321            _ => None,
2322        };
2323        let tape = st
2324            .tapes
2325            .get(&pid)
2326            .filter(|t| plain && !t.truncated)
2327            .map(|t| t.entries.clone());
2328        let input_root = (!plain && self.config.frame.is_some())
2329            .then(|| {
2330                st.accounts
2331                    .get(&pid)
2332                    .and_then(|b| b.input_root.clone())
2333                    .or_else(|| {
2334                        self.config
2335                            .state_root
2336                            .zip(info.entered.as_ref())
2337                            .map(|(root, entered)| root(entered))
2338                    })
2339            })
2340            .flatten();
2341        let carries = if plain {
2342            tape.is_some()
2343        } else {
2344            input_root.is_some()
2345        };
2346        let account = self
2347            .config
2348            .frame
2349            .as_ref()
2350            .filter(|_| carries)
2351            .map(|_| Account {
2352                host_io: self.config.host_io.of(pid),
2353                journal: self.config.journals().then(|| {
2354                    st.journals
2355                        .get(&pid)
2356                        .map(|r| r.iter().cloned().collect())
2357                        .unwrap_or_default()
2358                }),
2359                input_root,
2360            });
2361        let mut image = Image {
2362            definition: info.def_name.clone(),
2363            scheduling: info.scheduling,
2364            trap_exit: info.trap_exit,
2365            code,
2366            mailbox: st
2367                .boxes
2368                .get(&pid)
2369                .map(|b| b.iter().cloned().collect())
2370                .unwrap_or_default(),
2371            supervision,
2372            dispatch_steps: info.asked,
2373            parked: None,
2374            tape,
2375            account,
2376        };
2377        drop(st);
2378        if plain {
2379            image.parked = Some(self.continuation(pid)?);
2380        }
2381        Ok(image)
2382    }
2383
2384    fn continuation(&self, pid: Pid) -> Result<Continuation, String> {
2385        let sched = lock(&self.sched);
2386        let p = sched
2387            .iter()
2388            .find(|p| p.pid == pid)
2389            .ok_or("the process is gone")?;
2390        let mut st = self.st();
2391        let now = st.clock.now();
2392        let wait_ms = match st.waits.get(&pid) {
2393            Some(Wait::Receive {
2394                deadline: Some(d), ..
2395            }) => Some(d.saturating_sub(now)),
2396            _ => None,
2397        };
2398        let info = st.info.get(&pid).ok_or("the process is gone")?;
2399        let cannot = |why: String| format!("process {} cannot move: {why}", pid.0);
2400        if !info
2401            .chunk
2402            .as_ref()
2403            .is_some_and(|c| Arc::ptr_eq(c, &p.chunk))
2404        {
2405            return Err(cannot(
2406                "a migration replaced its code, and its frames still run the code it had \
2407                 before, which its image does not carry; let it restart or finish first"
2408                    .to_string(),
2409            ));
2410        }
2411        let vm = p.vm.parked(&p.chunk).map_err(|e| cannot(parking(&e)))?;
2412        let functions = info
2413            .code
2414            .defs
2415            .iter()
2416            .filter_map(defined_key)
2417            .map(|name| {
2418                let index = match p.interp.lookup_global(name) {
2419                    Some(Value::Foreign(any)) => any
2420                        .downcast_ref::<CompiledClosure>()
2421                        .filter(|c| {
2422                            c.captures.is_empty()
2423                                && (Arc::ptr_eq(&c.chunk, &p.chunk)
2424                                    || c.chunk.fingerprint() == p.chunk.fingerprint())
2425                        })
2426                        .map(|c| c.fn_index),
2427                    _ => None,
2428                };
2429                index.map(|i| (name.to_string(), i)).ok_or_else(|| {
2430                    cannot(format!(
2431                        "its global `{name}` is not the function its code defines"
2432                    ))
2433                })
2434            })
2435            .collect::<Result<_, _>>()?;
2436        Ok(Continuation {
2437            vm,
2438            functions,
2439            wait_ms,
2440        })
2441    }
2442
2443    fn resume(&self, pid: Pid, c: &Continuation, tape: Tape) -> Result<(), String> {
2444        let mut st = self.st();
2445        let p = st
2446            .pending
2447            .iter_mut()
2448            .find(|p| p.pid == pid)
2449            .ok_or("the process is gone")?;
2450        for (name, index) in &c.functions {
2451            let body = FnRef::Table(*index)
2452                .resolve(&p.chunk)
2453                .filter(|f| f.captures.is_empty())
2454                .ok_or_else(|| {
2455                    format!(
2456                        "its function `{name}` names entry {index}, which its code does not define"
2457                    )
2458                })?;
2459            let function = CompiledClosure {
2460                body: Arc::new(body.clone()),
2461                captures: Vec::new(),
2462                chunk: Arc::clone(&p.chunk),
2463                globals: p.interp.globals_snapshot().clone(),
2464                fn_index: *index,
2465            };
2466            p.interp
2467                .define_global(name.as_str(), Value::Foreign(Arc::new(function)));
2468        }
2469        p.vm = Vm::from_parked(&c.vm, &p.chunk, &p.interp, self.config.budget)
2470            .map_err(|e| parking(&e))?;
2471        if let Some(ms) = c.wait_ms {
2472            let deadline = st.clock.now().saturating_add(ms);
2473            st.waits.insert(
2474                pid,
2475                Wait::Receive {
2476                    pattern: kw(pattern::WILDCARD),
2477                    deadline: Some(deadline),
2478                },
2479            );
2480        }
2481        st.tapes.insert(pid, tape);
2482        Ok(())
2483    }
2484
2485    fn depart_over(
2486        &self,
2487        pid: Pid,
2488        wire: &Value,
2489        peer: &Value,
2490        loopback: bool,
2491        carrier: Carrier,
2492    ) -> Result<Value, String> {
2493        let image = self.image_of(pid)?;
2494        let queue = |image: &Image| -> Result<(String, usize, usize), String> {
2495            let text = image.text();
2496            let id = image::id_of(&text);
2497            let frames = image::frames(&id, &text, carrier.room)?;
2498            self.reachable_links(pid)?;
2499            let bytes = (carrier.queue)(wire, peer, &frames, loopback)?;
2500            Ok((id, frames.len(), bytes))
2501        };
2502        let (id, frames, bytes) = match queue(&image) {
2503            Err(_) if image.tape.is_some() || image.account.is_some() => queue(&Image {
2504                tape: None,
2505                account: None,
2506                ..image
2507            }),
2508            queued => queued,
2509        }?;
2510        self.depart(pid);
2511        Ok(record(&[
2512            ("image", Value::Str(Arc::from(format!("b3:{id}")))),
2513            ("frames", int(frames)),
2514            ("bytes", int(bytes)),
2515        ]))
2516    }
2517
2518    fn depart(&self, pid: Pid) {
2519        lock(&self.sched).retain(|p| p.pid != pid);
2520        lock(&self.lanes).remove(&pid);
2521        let mut st = self.st();
2522        let (monitors, links, sup) = st
2523            .info
2524            .get_mut(&pid)
2525            .map(|i| {
2526                (
2527                    std::mem::take(&mut i.monitors),
2528                    std::mem::take(&mut i.links),
2529                    i.sup,
2530                )
2531            })
2532            .unwrap_or_default();
2533        for (r, watcher) in monitors {
2534            st.deliver(
2535                watcher,
2536                Value::list([kw("down"), int(r), int(pid.0), kw(EVICTED)]),
2537            );
2538        }
2539        for l in links {
2540            let trapping = st.info.get_mut(&l).is_some_and(|i| {
2541                i.links.remove(&pid);
2542                i.trap_exit
2543            });
2544            if trapping {
2545                st.deliver(l, Value::list([kw("exit"), int(pid.0), kw(EVICTED)]));
2546            }
2547        }
2548        if let Some(s) = sup {
2549            st.drop_child(s, pid);
2550        }
2551        st.forget(pid);
2552        drop(st);
2553        self.config.host_io.take(pid);
2554    }
2555
2556    fn accept(
2557        &self,
2558        wire: &Value,
2559        timeout: Duration,
2560        loopback: bool,
2561        span: Span,
2562    ) -> Result<Value, EvalError> {
2563        let word = "accept";
2564        let carrier = self.carrier(word, span)?;
2565        let deadline = Instant::now() + timeout;
2566        loop {
2567            let left = deadline.saturating_duration_since(Instant::now());
2568            let frames = (carrier.take)(wire, image::FRAME, left, loopback)
2569                .map_err(|m| refuse(word, m, span))?;
2570            let (done, flaw) = {
2571                let mut st = self.st();
2572                let mut flaw = None;
2573                for f in &frames {
2574                    if let Err(why) = st.arrivals.file(f) {
2575                        flaw.get_or_insert(why);
2576                    }
2577                }
2578                (st.arrivals.complete(), flaw)
2579            };
2580            match (done, flaw) {
2581                (Some(Ok(text)), _) => return self.admit(&text, span),
2582                (Some(Err(why)), _) | (None, Some(why)) => {
2583                    return Err(raised("decode_error", why, Value::Nil, span))
2584                }
2585                (None, None) if frames.is_empty() => return Ok(Value::Nil),
2586                (None, None) => {}
2587            }
2588        }
2589    }
2590
2591    fn admit(&self, text: &str, span: Span) -> Result<Value, EvalError> {
2592        let (parsed, _) = self.apart(|| Image::parse(text));
2593        let mut image = parsed.map_err(|why| raised("decode_error", why, Value::Nil, span))?;
2594        let parked = image.parked.take();
2595        let id = image::id_of(text);
2596        let tape = Tape::arrived(image.tape.take(), &id);
2597        let account = image.account.take();
2598        let state = bound_state(&image.code);
2599        let handed = image.supervision.take();
2600        if let (Some(s), None) = (&handed, &parked) {
2601            image.code.bind("%state", s.initial.clone());
2602        }
2603        let sup = handed
2604            .as_ref()
2605            .map(|s| self.st().supervisor(Strategy::OneForOne, s.max_restarts));
2606        let started = self.start(
2607            image.code,
2608            if parked.is_some() {
2609                Kind::Plain
2610            } else {
2611                Kind::Server
2612            },
2613            image.definition,
2614            sup,
2615            image.scheduling,
2616        );
2617        let resumed = started.and_then(|pid| match &parked {
2618            Some(c) => self
2619                .apart(|| self.resume(pid, c, tape))
2620                .0
2621                .map(|()| pid)
2622                .map_err(|why| {
2623                    let mut st = self.st();
2624                    st.pending.retain(|p| p.pid != pid);
2625                    st.forget(pid);
2626                    format!("the image cannot resume here: {why}")
2627                }),
2628            None => Ok(pid),
2629        });
2630        let pid = match resumed {
2631            Ok(pid) => pid,
2632            Err(m) => {
2633                if let Some(sup) = sup {
2634                    self.st().forget(sup);
2635                }
2636                return Err(refuse("accept", m, span));
2637            }
2638        };
2639        if let (Some(sup), Some(s)) = (sup, &handed) {
2640            let mut st = self.st();
2641            st.enrol(sup, pid);
2642            if parked.is_none() {
2643                if let Some(p) = st.pending.iter_mut().find(|p| p.pid == pid) {
2644                    p.interp.define_global("%state", state.clone());
2645                }
2646            }
2647            if let Some(i) = st.info.get_mut(&pid) {
2648                i.restarts = s.restarts;
2649                if parked.is_none() {
2650                    i.snapshot = Some(state.clone());
2651                    i.entered = Some(state);
2652                }
2653            }
2654        }
2655        let (mailbox, bytes) = self.apart(|| {
2656            image
2657                .mailbox
2658                .iter()
2659                .filter_map(|m| copy_data(m).ok())
2660                .collect::<VecDeque<_>>()
2661        });
2662        {
2663            let mut st = self.st();
2664            if let Some(i) = st.info.get_mut(&pid) {
2665                i.trap_exit = image.trap_exit;
2666                i.ask_dispatch(image.dispatch_steps);
2667            }
2668            st.boxes.insert(pid, mailbox);
2669        }
2670        self.credit(pid, bytes);
2671        if let Some(a) = account {
2672            self.brought(pid, a, &id);
2673        }
2674        Ok(int(pid.0))
2675    }
2676
2677    fn brought(&self, pid: Pid, account: Account, image: &str) {
2678        if self.config.frame.is_none() {
2679            return;
2680        }
2681        let journal: Option<Vec<Value>> =
2682            account.journal.filter(|_| self.config.journals()).map(|j| {
2683                let skip = j.len().saturating_sub(self.config.journal);
2684                j.into_iter().skip(skip).collect()
2685            });
2686        let mark = CarriedAccount::of(journal.as_deref(), account.host_io, image);
2687        self.config.host_io.carry(pid, account.host_io);
2688        let mut st = self.st();
2689        if let Some(j) = journal {
2690            st.journals.insert(pid, j.into());
2691        }
2692        st.accounts.insert(
2693            pid,
2694            Brought {
2695                mark,
2696                input_root: account.input_root,
2697            },
2698        );
2699    }
2700
2701    // ── traces ─────────────────────────────────────────────────────────────
2702
2703    fn trace_of(&self, st: &mut NodeState, pid: Pid, reason: &ExitReason, value: &Value) -> Trace {
2704        let info = st.info.get(&pid).expect("a live process");
2705        let mut code = info.code.clone();
2706        if let (Kind::Server, Some(s)) = (&info.kind, &info.snapshot) {
2707            code.bind("%state", s.clone());
2708        }
2709        let tape = st.tapes.get(&pid).cloned().unwrap_or_default();
2710        let bound = self.config.journal;
2711        let mailbox = self.config.journals().then(|| {
2712            let waiting = st.boxes.get(&pid);
2713            Journal {
2714                bound,
2715                recent: st
2716                    .journals
2717                    .get(&pid)
2718                    .map(|r| r.iter().cloned().collect())
2719                    .unwrap_or_default(),
2720                unconsumed: waiting
2721                    .map(|b| b.iter().take(bound).cloned().collect())
2722                    .unwrap_or_default(),
2723                unconsumed_total: waiting.map_or(0, VecDeque::len),
2724            }
2725        });
2726        let reason_kind = crate::trace::reason_kind(reason).to_string();
2727        let places = &self.config.places;
2728        let violation = Violation::of(&reason_kind, &tape.entries, &code.defs, places);
2729        let defined_at = info.def_name.as_deref().and_then(|d| {
2730            let form = code.defs.iter().find(|f| defined_key(f) == Some(d))?;
2731            places.locate(d, form.span)
2732        });
2733        Trace {
2734            pid: pid.0,
2735            incarnation: info.restarts,
2736            definition: info.def_name.clone(),
2737            def_hash: info
2738                .def_name
2739                .as_deref()
2740                .and_then(|d| self.config.identities.of(d)),
2741            hold: self.config.frame.clone(),
2742            host_io: self
2743                .config
2744                .frame
2745                .as_ref()
2746                .map(|_| self.config.host_io.of(pid)),
2747            mailbox,
2748            violation,
2749            defined_at,
2750            reason: value.clone(),
2751            reason_text: crate::trace::reason_text(reason),
2752            reason_kind,
2753            budget: self.config.budget,
2754            code,
2755            carried: tape.carried,
2756            carried_account: st.accounts.get(&pid).map(|b| b.mark.clone()),
2757            tape: tape.entries,
2758            truncated: tape.truncated,
2759            schedule: st.schedule.iter().cloned().collect(),
2760            clock: self.config.clock,
2761        }
2762    }
2763
2764    /// The receipt of `pid`'s incarnation, which is ending: only a held
2765    /// node's processes have one, since only a hold's shims count host calls.
2766    fn receipt_of(
2767        &self,
2768        st: &NodeState,
2769        sched: &[Proc<()>],
2770        pid: Pid,
2771        reason: &ExitReason,
2772    ) -> Option<Receipt> {
2773        let host_io = self.config.host_io.take(pid);
2774        let frame = self.config.frame.clone()?;
2775        self.config.receipts.as_ref()?;
2776        let info = st.info.get(&pid)?;
2777        let proc = sched.iter().find(|p| p.pid == pid);
2778        let steps = proc
2779            .filter(|_| self.config.count_steps)
2780            .map(|p| u64::try_from(p.vm.burned()).unwrap_or(u64::MAX));
2781        let brought = st.accounts.get(&pid);
2782        let roots = match (&info.kind, self.config.state_root, &info.entered) {
2783            (Kind::Server, Some(root), Some(entered)) => {
2784                let left = proc
2785                    .and_then(|p| p.interp.lookup_global("%state"))
2786                    .or_else(|| info.snapshot.clone())
2787                    .unwrap_or(Value::Nil);
2788                Some(Roots {
2789                    input: brought
2790                        .and_then(|b| b.input_root.clone())
2791                        .unwrap_or_else(|| root(entered)),
2792                    output: root(&left),
2793                })
2794            }
2795            _ => None,
2796        };
2797        let carried = brought.map(|b| CarriedIo {
2798            image: b.mark.image.clone(),
2799            host_io: b.mark.host_io,
2800            input_root: b.input_root.clone().filter(|_| roots.is_some()),
2801        });
2802        Some(Receipt {
2803            subject: Subject::Actor {
2804                pid,
2805                incarnation: info.restarts,
2806            },
2807            definition: info.def_name.clone(),
2808            def_hash: info
2809                .def_name
2810                .as_deref()
2811                .and_then(|d| self.config.identities.of(d)),
2812            frame,
2813            host_io,
2814            exit_kind: crate::trace::reason_kind(reason).to_string(),
2815            exit_text: crate::trace::reason_text(reason),
2816            roots,
2817            steps,
2818            carried,
2819        })
2820    }
2821
2822    fn keep_trace(&self, trace: Trace, write: bool) {
2823        if let Some(dir) = self.config.trace_dir.as_ref().filter(|_| write) {
2824            match trace.write_to(dir) {
2825                Ok(path) => eprintln!(
2826                    "blue: process {} exited ({}); trace: {}",
2827                    trace.pid,
2828                    trace.reason_text,
2829                    path.display()
2830                ),
2831                Err(e) => eprintln!(
2832                    "blue: process {} exited; its trace could not be written: {e}",
2833                    trace.pid
2834                ),
2835            }
2836        }
2837        let mut st = self.st();
2838        if st.traces.len() == KEPT_TRACES {
2839            st.traces.pop_front();
2840        }
2841        st.traces.push_back(trace);
2842    }
2843
2844    /// The crash traces this node kept, oldest first.
2845    #[must_use]
2846    pub fn traces(&self) -> Vec<Trace> {
2847        self.st().traces.iter().cloned().collect()
2848    }
2849
2850    // ── driving, for the root ──────────────────────────────────────────────
2851
2852    /// Run the other processes until `goal` answers, `deadline` passes, or
2853    /// nothing can change.
2854    ///
2855    /// The scheduling is charged to no one: each slice to its process, the
2856    /// node's own work (restarts, traces, reaping) to nobody, and what the
2857    /// children sent the root to the root once it has its answer.
2858    fn drive<T>(
2859        &self,
2860        goal: impl FnMut(&mut NodeState) -> Option<T>,
2861        deadline: Option<u64>,
2862        word: &str,
2863        span: Span,
2864    ) -> Result<Driven<T>, EvalError> {
2865        let driving = self.driving();
2866        let (driven, _) = self.apart(|| self.drive_unmetered(goal, deadline, word, span));
2867        drop(driving);
2868        self.settle_root();
2869        driven
2870    }
2871
2872    fn drive_unmetered<T>(
2873        &self,
2874        mut goal: impl FnMut(&mut NodeState) -> Option<T>,
2875        deadline: Option<u64>,
2876        word: &str,
2877        span: Span,
2878    ) -> Result<Driven<T>, EvalError> {
2879        let mut sched = lock(&self.sched);
2880        loop {
2881            self.adopt(&mut sched);
2882            self.process_signals(&mut sched);
2883            {
2884                let mut st = self.st();
2885                if let Some((from, reason)) = st.root_signal.take() {
2886                    return Err(refuse(
2887                        word,
2888                        format!("linked process {} exited: {}", from.0, render(&reason)),
2889                        span,
2890                    ));
2891                }
2892                if let Some(t) = goal(&mut st) {
2893                    return Ok(Driven::Done(t));
2894                }
2895                if deadline.is_some_and(|d| st.clock.now() >= d) {
2896                    return Ok(Driven::TimedOut);
2897                }
2898            }
2899            if self.apply_migrations(&mut sched) {
2900                continue;
2901            }
2902            if self.land(&mut sched) {
2903                continue;
2904            }
2905            if self.has_work(&sched) {
2906                self.round(&mut sched);
2907                continue;
2908            }
2909            // Nothing can run now: move time to the next deadline, or say that
2910            // nothing will ever change.
2911            let next = {
2912                let st = self.st();
2913                match (st.earliest_deadline(), deadline) {
2914                    (Some(a), Some(b)) => Some(a.min(b)),
2915                    (a, b) => a.or(b),
2916                }
2917            };
2918            let aloft = {
2919                let mut st = self.st();
2920                let now = st.clock.now();
2921                (!st.aloft.is_empty()).then(|| match st.clock.kind {
2922                    ClockKind::Real => next.map(|t| Duration::from_millis(t.saturating_sub(now))),
2923                    ClockKind::Virtual => None,
2924                })
2925            };
2926            if let Some(limit) = aloft {
2927                self.await_flight(&mut sched, limit);
2928                continue;
2929            }
2930            match next {
2931                Some(t) => {
2932                    let pause = self.st().clock.advance_to(t);
2933                    if let Some(d) = pause {
2934                        std::thread::sleep(d);
2935                    }
2936                }
2937                None => {
2938                    if self.fail_stuck_migrations() {
2939                        continue;
2940                    }
2941                    let blocked = sched
2942                        .iter()
2943                        .filter(|p| p.state.is_blocked())
2944                        .map(|p| p.pid)
2945                        .collect();
2946                    return Ok(Driven::Deadlock(blocked));
2947                }
2948            }
2949        }
2950    }
2951
2952    /// Migrations whose process is parked somewhere other than a boundary,
2953    /// with nothing left to wake it, can never apply: answer them.
2954    fn fail_stuck_migrations(&self) -> bool {
2955        let mut st = self.st();
2956        let stuck: Vec<(Pid, Migration)> = std::mem::take(&mut st.migrations).into_iter().collect();
2957        let any = !stuck.is_empty();
2958        for (pid, m) in stuck {
2959            st.outcomes.entry(m.request).or_default().push((
2960                pid,
2961                Err("the process is waiting inside its handler and nothing can wake it, so it never \
2962                     reached a boundary"
2963                    .into()),
2964            ));
2965        }
2966        any
2967    }
2968
2969    fn receive_root(&self, args: &[Value], span: Span) -> Result<Value, EvalError> {
2970        let (pattern, timeout, default) = receive_args(args, span)?;
2971        let deadline = timeout.map(|t| self.st().clock.now().saturating_add(t));
2972        let driven = self.drive(
2973            |st| {
2974                let b = st.boxes.get_mut(&ROOT)?;
2975                let i = pattern::first_match(b, &pattern)?;
2976                b.remove(i)
2977            },
2978            deadline,
2979            "receive",
2980            span,
2981        )?;
2982        match driven {
2983            Driven::Done(msg) => Ok(msg),
2984            Driven::TimedOut => Ok(default),
2985            Driven::Deadlock(blocked) => Err(refuse(
2986                "receive",
2987                format!(
2988                    "would wait forever: no process can send (deadlock; waiting: {})",
2989                    blocked
2990                        .iter()
2991                        .map(|p| p.0.to_string())
2992                        .collect::<Vec<_>>()
2993                        .join(", ")
2994                ),
2995                span,
2996            )),
2997        }
2998    }
2999
3000    /// Run every process until none can make progress. For hosts and tests;
3001    /// a blue program waits with `receive`.
3002    pub fn run_until_idle(&self) {
3003        let _ = self.drive(|_| None::<()>, None, "run", Span::synthetic());
3004    }
3005
3006    /// Processes the node still holds: queued, running, parked, or finished
3007    /// under a supervisor that may restart them. A finished unsupervised
3008    /// process is not among them; its interpreter has been dropped.
3009    #[must_use]
3010    pub fn process_count(&self) -> usize {
3011        let sched = lock(&self.sched).len();
3012        let st = self.st();
3013        sched + st.pending.len() + st.aloft.len()
3014    }
3015
3016    /// The mailbox of `pid`, oldest first. For hosts and tests.
3017    #[must_use]
3018    pub fn mailbox(&self, pid: Pid) -> Vec<Value> {
3019        self.st()
3020            .boxes
3021            .get(&pid)
3022            .map(|b| b.iter().cloned().collect())
3023            .unwrap_or_default()
3024    }
3025}
3026
3027// ── the words ──────────────────────────────────────────────────────────────
3028
3029fn bound_state(code: &Code) -> Value {
3030    code.bindings
3031        .iter()
3032        .find(|(n, _)| &**n == "%state")
3033        .map_or(Value::Nil, |(_, v)| v.clone())
3034}
3035
3036fn record(fields: &[(&str, Value)]) -> Value {
3037    Value::map(
3038        fields
3039            .iter()
3040            .map(|(k, v)| (MapKey::Keyword(Arc::from(*k)), v.clone())),
3041    )
3042}
3043
3044const DISPATCH_OPTION: &str = "dispatch_steps";
3045
3046const SPAWN_OPTIONS: &[&str] = &[scheduling::OPTION];
3047
3048const SERVER_OPTIONS: &[&str] = &[scheduling::OPTION, DISPATCH_OPTION];
3049
3050#[derive(Clone, Copy, Debug, Default)]
3051struct Opts {
3052    scheduling: Scheduling,
3053    dispatch_steps: Option<usize>,
3054}
3055
3056fn dispatch_steps(v: &Value, span: Span) -> Result<usize, EvalError> {
3057    match v {
3058        Value::Int(n) if *n > 0 => Ok(usize::try_from(*n).unwrap_or(usize::MAX)),
3059        other => Err(raised(
3060            "bad_option",
3061            format!(
3062                "{DISPATCH_OPTION} is a positive integer, got {}",
3063                match other {
3064                    Value::Int(n) => n.to_string(),
3065                    other => other.type_name().to_string(),
3066                }
3067            ),
3068            Value::Nil,
3069            span,
3070        )),
3071    }
3072}
3073
3074fn spawn_opts(
3075    v: Option<&Value>,
3076    word: &str,
3077    takes: &[&str],
3078    span: Span,
3079) -> Result<Opts, EvalError> {
3080    let m = match v {
3081        None | Some(Value::Nil) => return Ok(Opts::default()),
3082        Some(Value::Map(m)) => m,
3083        Some(other) => {
3084            return Err(raised(
3085                "bad_option",
3086                format!(
3087                    "{word}'s options are a map {{{}: …}}, got {}",
3088                    scheduling::OPTION,
3089                    other.type_name()
3090                ),
3091                Value::Nil,
3092                span,
3093            ))
3094        }
3095    };
3096    let mut opts = Opts::default();
3097    for (k, val) in m.iter() {
3098        let key = match k {
3099            MapKey::Keyword(s) | MapKey::Symbol(s) | MapKey::Str(s) => {
3100                s.trim_start_matches(':').to_string()
3101            }
3102            other => other.to_value().to_string(),
3103        };
3104        if !takes.contains(&key.as_str()) {
3105            let named: Vec<String> = takes.iter().map(|t| format!("{t}:")).collect();
3106            return Err(raised(
3107                "bad_option",
3108                format!("{word} takes {}, got {key}:", named.join(" or ")),
3109                Value::Nil,
3110                span,
3111            ));
3112        }
3113        if key == DISPATCH_OPTION {
3114            opts.dispatch_steps = Some(dispatch_steps(val, span)?);
3115            continue;
3116        }
3117        opts.scheduling = match val {
3118            Value::Keyword(w) => Scheduling::named(w),
3119            other => Err(scheduling::Refusal::Unknown(format!(
3120                "a {}",
3121                other.type_name()
3122            ))),
3123        }
3124        .map_err(|r| raised(r.kind(), r.to_string(), Value::Nil, span))?;
3125    }
3126    Ok(opts)
3127}
3128
3129fn spawn_args(args: &[Value], word: &str, span: Span) -> Result<Vec<Value>, EvalError> {
3130    match args.get(1) {
3131        None | Some(Value::Nil) => Ok(Vec::new()),
3132        Some(Value::List(l)) => Ok(l.iter().cloned().collect()),
3133        Some(other) => Err(refuse(
3134            word,
3135            format!("the arguments are a list, got {}", other.type_name()),
3136            span,
3137        )),
3138    }
3139}
3140
3141fn spawn_one(
3142    node: &Node,
3143    f: &Value,
3144    args: &[Value],
3145    sup: Option<Pid>,
3146    word: &str,
3147    span: Span,
3148    scheduling: Scheduling,
3149) -> Result<Pid, EvalError> {
3150    let (code, entry) = code::spawn_code(f, args).map_err(|m| refuse(word, m, span))?;
3151    let def_name = match entry {
3152        Entry::Named(n) | Entry::Builtin(n) => Some(n),
3153        Entry::Lambda(_) => None,
3154    };
3155    node.start(code, Kind::Plain, def_name, sup, scheduling)
3156        .map_err(|m| refuse(word, m, span))
3157}
3158
3159fn w_spawn(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3160    let a = spawn_args(args, "spawn", span)?;
3161    let class = spawn_opts(args.get(2), "spawn", SPAWN_OPTIONS, span)?.scheduling;
3162    Ok(int(spawn_one(
3163        node, &args[0], &a, None, "spawn", span, class,
3164    )?
3165    .0))
3166}
3167
3168fn w_spawn_link(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3169    let a = spawn_args(args, "spawn_link", span)?;
3170    let class = spawn_opts(args.get(2), "spawn_link", SPAWN_OPTIONS, span)?.scheduling;
3171    let pid = spawn_one(node, &args[0], &a, None, "spawn_link", span, class)?;
3172    let mut st = node.st();
3173    let me = who.pid();
3174    if let Some(i) = st.info.get_mut(&me) {
3175        i.links.insert(pid);
3176    }
3177    if let Some(i) = st.info.get_mut(&pid) {
3178        i.links.insert(me);
3179    }
3180    Ok(int(pid.0))
3181}
3182
3183/// The `def_hash` the node's identities give the definition `f` is bound
3184/// under, as a trace and a receipt name it; nil when they give none.
3185fn w_def_hash(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3186    let f = &args[0];
3187    if !code::is_function(f) {
3188        return Err(refuse(
3189            "def_hash",
3190            format!("expected a function, got {}", f.type_name()),
3191            span,
3192        ));
3193    }
3194    Ok(code::name_of(f)
3195        .and_then(|key| node.config.identities.of(&key))
3196        .map_or(Value::Nil, |hash| Value::Str(Arc::from(hash))))
3197}
3198
3199fn w_spawn_server(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3200    let opts = spawn_opts(args.get(2), "spawn_server", SERVER_OPTIONS, span)?;
3201    Ok(int(server(
3202        node,
3203        &args[0],
3204        &args[1],
3205        None,
3206        "spawn_server",
3207        span,
3208        opts,
3209    )?
3210    .0))
3211}
3212
3213fn server(
3214    node: &Node,
3215    handler: &Value,
3216    state: &Value,
3217    sup: Option<Pid>,
3218    word: &str,
3219    span: Span,
3220    opts: Opts,
3221) -> Result<Pid, EvalError> {
3222    let mut code = code::function_code(handler, "%handler").map_err(|m| refuse(word, m, span))?;
3223    code.defs.extend(SERVER_LOOP.iter().cloned());
3224    code.start = Some(code::call(code::symbol("%serve"), &[]));
3225    let state = copy_data(state).map_err(|kind| {
3226        refuse(
3227            word,
3228            format!("a server's state is copied into the process, and a {kind} is not data"),
3229            span,
3230        )
3231    })?;
3232    code.bind("%state", state);
3233    let pid = node
3234        .start(
3235            code,
3236            Kind::Server,
3237            code::name_of(handler),
3238            sup,
3239            opts.scheduling,
3240        )
3241        .map_err(|m| refuse(word, m, span))?;
3242    if let Some(info) = node.st().info.get_mut(&pid) {
3243        info.ask_dispatch(opts.dispatch_steps);
3244    }
3245    Ok(pid)
3246}
3247
3248fn w_evict(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3249    node.tally();
3250    evicting(node, who, args, false, span)
3251}
3252
3253fn w_evict_provided(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3254    evicting(node, who, args, true, span)
3255}
3256
3257fn evicting(
3258    node: &Node,
3259    who: Who,
3260    args: &[Value],
3261    loopback: bool,
3262    span: Span,
3263) -> Result<Value, EvalError> {
3264    if who != Who::Root {
3265        return Err(refuse(
3266            "evict",
3267            "evict waits for the process to reach a message boundary, and only the program \
3268             waits; a process asks the program to evict",
3269            span,
3270        ));
3271    }
3272    let pid = pid_arg(&args[0], "evict", span)?;
3273    node.evict(pid, &args[1], &args[2], loopback, span)
3274}
3275
3276fn w_accept(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3277    node.tally();
3278    accepting(node, args, false, span)
3279}
3280
3281fn w_accept_provided(
3282    node: &Node,
3283    _who: Who,
3284    args: &[Value],
3285    span: Span,
3286) -> Result<Value, EvalError> {
3287    accepting(node, args, true, span)
3288}
3289
3290fn accepting(node: &Node, args: &[Value], loopback: bool, span: Span) -> Result<Value, EvalError> {
3291    let timeout = match &args[1] {
3292        Value::Int(n) if *n >= 0 => Duration::from_millis(n.unsigned_abs()),
3293        other => {
3294            return Err(refuse(
3295                "accept",
3296                format!(
3297                    "the timeout is milliseconds (an Int >= 0), got {}",
3298                    other.type_name()
3299                ),
3300                span,
3301            ))
3302        }
3303    };
3304    node.accept(&args[0], timeout, loopback, span)
3305}
3306
3307fn w_uncarried(_node: &Node, _who: Who, _args: &[Value], span: Span) -> Result<Value, EvalError> {
3308    Err(refuse(
3309        "evict",
3310        "this node has no wire to carry a process",
3311        span,
3312    ))
3313}
3314
3315fn w_send(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3316    let to = pid_arg(&args[0], "send", span)?;
3317    let (copied, bytes) = node.apart(|| {
3318        copy_data(&args[1]).map(|msg| {
3319            // A send to a process that has exited is dropped, as in Erlang: the
3320            // sender cannot know, and a send that raised would race the exit.
3321            node.st().deliver(to, msg);
3322        })
3323    });
3324    copied.map_err(|kind| {
3325        refuse(
3326            "send",
3327            format!("a message is copied to the receiver, and a {kind} is not data"),
3328            span,
3329        )
3330    })?;
3331    node.credit(to, bytes);
3332    Ok(args[1].clone())
3333}
3334
3335fn w_self(_node: &Node, who: Who, _args: &[Value], _span: Span) -> Result<Value, EvalError> {
3336    Ok(int(who.pid().0))
3337}
3338
3339fn w_alive(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3340    let pid = pid_arg(&args[0], "alive?", span)?;
3341    Ok(Value::Bool(
3342        node.st().info.get(&pid).is_some_and(|i| i.live),
3343    ))
3344}
3345
3346fn w_link(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3347    let other = pid_arg(&args[0], "link", span)?;
3348    let me = who.pid();
3349    let mut st = node.st();
3350    if other == me || !st.info.contains_key(&other) {
3351        return Ok(Value::Bool(false));
3352    }
3353    if let Some(i) = st.info.get_mut(&me) {
3354        i.links.insert(other);
3355    }
3356    if let Some(i) = st.info.get_mut(&other) {
3357        i.links.insert(me);
3358    }
3359    Ok(Value::Bool(true))
3360}
3361
3362fn w_unlink(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3363    let other = pid_arg(&args[0], "unlink", span)?;
3364    let me = who.pid();
3365    let mut st = node.st();
3366    if let Some(i) = st.info.get_mut(&me) {
3367        i.links.remove(&other);
3368    }
3369    if let Some(i) = st.info.get_mut(&other) {
3370        i.links.remove(&me);
3371    }
3372    Ok(Value::Bool(true))
3373}
3374
3375fn w_monitor(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3376    let target = pid_arg(&args[0], "monitor", span)?;
3377    let me = who.pid();
3378    let mut st = node.st();
3379    let r = st.next_ref;
3380    st.next_ref += 1;
3381    match st.info.get_mut(&target) {
3382        Some(i) => i.monitors.push((r, me)),
3383        None => st.deliver(
3384            me,
3385            Value::list([kw("down"), int(r), int(target.0), kw("noproc")]),
3386        ),
3387    }
3388    Ok(int(r))
3389}
3390
3391fn w_demonitor(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3392    let Value::Int(r) = args[0] else {
3393        return Err(refuse(
3394            "demonitor",
3395            "expected a monitor reference (an Int)",
3396            span,
3397        ));
3398    };
3399    let mut st = node.st();
3400    for i in st.info.values_mut() {
3401        i.monitors.retain(|(x, _)| *x != r);
3402    }
3403    Ok(Value::Bool(true))
3404}
3405
3406fn w_exit(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3407    let to = pid_arg(&args[0], "exit", span)?;
3408    let reason = copy_data(&args[1])
3409        .map_err(|kind| refuse("exit", format!("a reason is data, not a {kind}"), span))?;
3410    let me = who.pid();
3411    if to == me {
3412        return Err(refuse(
3413            "exit",
3414            format!("process {} exits: {}", me.0, render(&reason)),
3415            span,
3416        ));
3417    }
3418    node.st().signals.push_back((me, to, reason));
3419    if who == Who::Root {
3420        // The root is not inside a slice, so the signal is delivered now.
3421        let mut sched = lock(&node.sched);
3422        node.adopt(&mut sched);
3423        node.process_signals(&mut sched);
3424        node.reap(&mut sched);
3425    }
3426    Ok(Value::Bool(true))
3427}
3428
3429fn truthy(v: &Value) -> bool {
3430    !matches!(v, Value::Nil | Value::Bool(false))
3431}
3432
3433fn w_trap_exit(node: &Node, who: Who, args: &[Value], _span: Span) -> Result<Value, EvalError> {
3434    let mut st = node.st();
3435    let Some(i) = st.info.get_mut(&who.pid()) else {
3436        return Ok(Value::Bool(false));
3437    };
3438    let previous = i.trap_exit;
3439    i.trap_exit = truthy(&args[0]);
3440    Ok(Value::Bool(previous))
3441}
3442
3443fn w_supervisor(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3444    let strategy = match &args[0] {
3445        Value::Keyword(k) => Strategy::from_keyword(k.trim_start_matches(':')),
3446        _ => None,
3447    }
3448    .ok_or_else(|| {
3449        refuse(
3450            "supervisor",
3451            "the strategy is :one_for_one, :one_for_all or :rest_for_one",
3452            span,
3453        )
3454    })?;
3455    let Value::List(specs) = &args[1] else {
3456        return Err(refuse(
3457            "supervisor",
3458            "the children are a list of functions or [f, args] pairs",
3459            span,
3460        ));
3461    };
3462    let max_restarts = match args.get(2) {
3463        None | Some(Value::Nil) => DEFAULT_MAX_RESTARTS,
3464        Some(Value::Int(n)) if *n >= 0 => usize::try_from(*n).unwrap_or(usize::MAX),
3465        Some(other) => {
3466            return Err(refuse(
3467                "supervisor",
3468                format!("max_restarts is an Int >= 0, got {}", other.type_name()),
3469                span,
3470            ))
3471        }
3472    };
3473    let sup = node.st().supervisor(strategy, max_restarts);
3474    for spec in specs.iter() {
3475        let pid = match spec {
3476            // `[:server, handler, state]`: a supervised spawn_server.
3477            Value::List(t) if (3..=4).contains(&t.len()) && is_keyword(&t[0], "server") => server(
3478                node,
3479                &t[1],
3480                &t[2],
3481                Some(sup),
3482                "supervisor",
3483                span,
3484                spawn_opts(t.get(3), "supervisor", SERVER_OPTIONS, span)?,
3485            )?,
3486            Value::List(pair) if pair.len() == 2 && code::is_function(&pair[0]) => {
3487                let Value::List(a) = &pair[1] else {
3488                    return Err(refuse(
3489                        "supervisor",
3490                        "a child [f, args] takes its args as a list",
3491                        span,
3492                    ));
3493                };
3494                let a: Vec<Value> = a.iter().cloned().collect();
3495                spawn_one(
3496                    node,
3497                    &pair[0],
3498                    &a,
3499                    Some(sup),
3500                    "supervisor",
3501                    span,
3502                    Scheduling::Normal,
3503                )?
3504            }
3505            f => spawn_one(
3506                node,
3507                f,
3508                &[],
3509                Some(sup),
3510                "supervisor",
3511                span,
3512                Scheduling::Normal,
3513            )?,
3514        };
3515        node.st().enrol(sup, pid);
3516    }
3517    Ok(int(sup.0))
3518}
3519
3520fn w_children(node: &Node, _who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3521    let sup = pid_arg(&args[0], "children", span)?;
3522    let st = node.st();
3523    match st.info.get(&sup).map(|i| &i.kind) {
3524        Some(Kind::Supervisor(s)) => Ok(Value::list(s.children.iter().map(|p| int(p.0)))),
3525        _ => Ok(Value::list([])),
3526    }
3527}
3528
3529fn w_defmigrate(node: &Node, who: Who, args: &[Value], span: Span) -> Result<Value, EvalError> {
3530    let targets: Vec<Pid> = match &args[0] {
3531        Value::Int(_) => vec![pid_arg(&args[0], "defmigrate", span)?],
3532        f if code::is_function(f) => {
3533            let Some(name) = code::name_of(f) else {
3534                return Err(refuse(
3535                    "defmigrate",
3536                    "the target function has no name to match processes by",
3537                    span,
3538                ));
3539            };
3540            let st = node.st();
3541            st.info
3542                .iter()
3543                .filter(|(_, i)| {
3544                    matches!(i.kind, Kind::Plain | Kind::Server)
3545                        && i.def_name.as_deref() == Some(&name)
3546                })
3547                .map(|(p, _)| *p)
3548                .collect()
3549        }
3550        other => {
3551            return Err(refuse(
3552                "defmigrate",
3553                format!(
3554                    "the target is a pid or the function processes run, got {}",
3555                    other.type_name()
3556                ),
3557                span,
3558            ))
3559        }
3560    };
3561    let request = {
3562        let mut st = node.st();
3563        let r = st.next_request;
3564        st.next_request += 1;
3565        r
3566    };
3567    let new_name = code::name_of(&args[1]).unwrap_or_else(|| format!("%migration{request}"));
3568    let new_code =
3569        code::function_code(&args[1], &new_name).map_err(|m| refuse("defmigrate", m, span))?;
3570    let transform = match args.get(2) {
3571        None | Some(Value::Nil) => None,
3572        Some(t) => {
3573            Some(code::function_code(t, "%transform").map_err(|m| refuse("defmigrate", m, span))?)
3574        }
3575    };
3576    let mut answers: Vec<(Pid, Result<(), String>)> = Vec::new();
3577    let mut queued = Vec::new();
3578    {
3579        let mut st = node.st();
3580        for pid in targets {
3581            match st.info.get(&pid).map(|i| &i.kind) {
3582                Some(Kind::Plain | Kind::Server) => {
3583                    st.migrations.insert(
3584                        pid,
3585                        Migration {
3586                            request,
3587                            code: new_code.clone(),
3588                            new_name: new_name.clone(),
3589                            transform: transform.clone(),
3590                        },
3591                    );
3592                    queued.push(pid);
3593                }
3594                _ => answers.push((pid, Err("no such process".into()))),
3595            }
3596        }
3597    }
3598    if who != Who::Root {
3599        // A child cannot wait for other processes inside its slice; the
3600        // migration is applied at the target's next boundary.
3601        return Ok(Value::list(
3602            queued.iter().map(|p| Value::list([int(p.0), kw("queued")])),
3603        ));
3604    }
3605    if !queued.is_empty() {
3606        let want = queued.len();
3607        let driven = node.drive(
3608            |st| {
3609                (st.outcomes.get(&request).map_or(0, Vec::len) >= want)
3610                    .then(|| st.outcomes.remove(&request).unwrap_or_default())
3611            },
3612            None,
3613            "defmigrate",
3614            span,
3615        )?;
3616        if let Driven::Done(done) = driven {
3617            answers.extend(done);
3618        }
3619    }
3620    answers.sort_by_key(|(p, _)| *p);
3621    Ok(Value::list(answers.into_iter().map(|(p, r)| {
3622        Value::list([
3623            int(p.0),
3624            match r {
3625                Ok(()) => kw("ok"),
3626                Err(m) => Value::list([kw("error"), Value::Str(Arc::from(m.as_str()))]),
3627            },
3628        ])
3629    })))
3630}