narju/

world.rs

use std::collections::HashMap;

use tokio::sync::mpsc::{channel, Receiver, Sender, WeakSender};

use crate::floor::machine::Fault;
use crate::floor::{list, list_elements, pair, rc_val, Val};
use crate::link::{self, Frame};
use crate::sched::{self, addr, Addr, Heap, Id};
use crate::wire;

fn sym(s: &str) -> Val {
    Val::Sym(s.into())
}

/// Local: the same peer is a different number seen from the other end.
pub type ConnId = u64;

/// One queue carries everything arriving from outside the heap, so a quiescent
/// node waits on a single receiver rather than selecting over a set that
/// changes as peers come and go.
pub enum Event {
    Post(Val, Val),
    /// A connection established while the loop runs, which is every one a node
    /// accepts rather than makes; the counterpart before it starts is
    /// [`World::link`].
    Link(ConnId, Sender<Frame>),
    /// The only liveness edge a node has about a far side.
    Down(ConnId),
    /// A frame addressed to the node rather than a task in it
    /// ([`crate::link::NODE`]):
    ///
    /// ```text
    /// ('watch id)         watch a task of yours for me
    /// ('down  id result)  it ended, and this is what with
    /// ```
    ///
    /// Not a service reached through the greeter, because a table held by a
    /// task is lost when the task restarts with no event to say so.
    Node(ConnId, Val),
    /// A host peer's adapter has dropped the receiving end. Raised by a waiter
    /// the loop spawns, so a peer ending is an edge rather than a question.
    Gone(String),
}

/// `('host name)`, deliberately the same shape as a task's: a call site cannot
/// tell a peer written in Rust from one in the object language.
pub fn host(name: &str) -> Val {
    list(&[sym("host"), sym(name)])
}

/// `('req from id body)` is what the prelude's `call` sends. An adapter answers
/// it exactly as a task would, which is why `call` works against Rust.
pub fn request(msg: &Val) -> Option<(Val, Val, Val)> {
    match list_elements(msg)?.as_slice() {
        [tag, from, id, body] if *tag == sym("req") => {
            Some((from.clone(), id.clone(), body.clone()))
        }
        _ => None,
    }
}

/// `('reply id v)`, the answer to a request.
pub fn reply(id: Val, v: Val) -> Val {
    list(&[sym("reply"), id, v])
}

/// `noconnection` rather than an exit value, because no exit was seen: the link
/// going is this node losing the ability to hear about the far task, not the
/// far task ending. Distinct from `noproc`, a task known to have ended.
fn lost(conn: ConnId, id: Id) -> Val {
    let why = pair(sym("throw"), sym("noconnection"));
    sched::down(Addr::Remote(conn, id).val(), why)
}

/// `noproc` and no value, the same answer a monitor of an already-ended task
/// gets. Nothing re-registers a peer, so an adapter that has returned is gone
/// for good and the distinction `noconnection` draws does not arise.
fn gone(target: Val) -> Val {
    sched::down(target, pair(sym("throw"), sym("noproc")))
}

/// The id needs no translating: it was written in the far node's numbering by
/// whoever handed the address out.
fn watch_of(id: Id) -> Val {
    list(&[sym("watch"), Val::Num(id as i64)])
}

fn down_of(id: Id, result: Val) -> Val {
    list(&[sym("down"), Val::Num(id as i64), result])
}

/// One heap and the peers it can reach.
pub struct World {
    heap: Heap,
    hosts: HashMap<String, Sender<Val>>,
    links: HashMap<ConnId, Sender<Frame>>,
    /// `conn -> [(watcher, far id)]`. Kept here because the death this node can
    /// observe is the link's, and the links are here.
    watches: HashMap<ConnId, Vec<(Id, Id)>>,
    /// One waiter is spawned per name watched, not per watcher.
    peer_watches: HashMap<String, Vec<Id>>,
    /// Tasks here that links asked to be told about, the other way round from
    /// `watches`.
    remote_watches: HashMap<Id, Vec<ConnId>>,
    /// Template for [`World::port`], dropped when the loop starts so the inbox
    /// is open exactly while some adapter or link still holds a clone.
    port: Option<Sender<Event>>,
    /// The same way in, held weakly so the loop does not keep itself alive.
    /// Failing to upgrade means nothing is left that could answer this world.
    back: WeakSender<Event>,
    inbox: Receiver<Event>,
}

impl World {
    pub fn new(budget: usize) -> World {
        let (port, inbox) = channel(64);
        let back = port.downgrade();
        World {
            heap: Heap::new(budget),
            hosts: HashMap::new(),
            links: HashMap::new(),
            watches: HashMap::new(),
            peer_watches: HashMap::new(),
            remote_watches: HashMap::new(),
            port: Some(port),
            back,
            inbox,
        }
    }

    pub fn heap(&mut self) -> &mut Heap {
        &mut self.heap
    }

    /// Register a peer under `name`, reached from the object language as
    /// `('host name)`. This end knows only where to put a message.
    pub fn serve(&mut self, name: &str, tx: Sender<Val>) {
        self.hosts.insert(name.to_string(), tx);
    }

    /// A link established before the loop starts; the counterpart for one
    /// established while it runs is [`Event::Link`].
    pub fn link(&mut self, conn: ConnId, tx: Sender<Frame>) {
        self.links.insert(conn, tx);
    }

    /// Every live clone holds the world open, so an adapter that has finished
    /// must drop its own.
    pub fn port(&self) -> Sender<Event> {
        self.port.clone().expect("the world has not started")
    }

    /// Run until nothing can wake the heap again, answering with the
    /// tasks that ended and what they ended with.
    pub async fn run(mut self) -> Result<Vec<(Id, Val)>, Fault> {
        self.port = None;
        let mut ended = Vec::new();
        loop {
            self.heap.run()?;
            let exits = self.heap.take_exits();
            // Before the await, or a task waiting on a reply to the message it
            // just sent would wait for it to be sent. Watches first, since a
            // watch is itself a frame and must not be overtaken by the message
            // it was set for - watching early cannot lose a death, late can.
            let mut woke = false;
            for (watcher, target) in self.heap.take_watches() {
                woke |= self.watch(watcher, target).await;
            }
            for (to, msg) in self.heap.take_away() {
                woke |= self.forward(to, msg).await;
            }
            // After those and never before, the order the heap uses for its own
            // watchers: a death that overtook an answer would report
            // `callee-down` for a call that was in fact answered.
            for (id, result) in &exits {
                self.ended(*id, result).await;
            }
            ended.extend(exits);
            if self.heap.idle() {
                return Ok(ended);
            }
            // The world answers some addresses itself, so a task can be
            // runnable again with nothing having arrived from outside.
            if woke {
                continue;
            }
            match self.inbox.recv().await {
                Some(Event::Post(to, msg)) => self.heap.post(&to, msg),
                Some(Event::Link(conn, tx)) => {
                    self.links.insert(conn, tx);
                }
                Some(Event::Down(conn)) => {
                    self.links.remove(&conn);
                    for (watcher, id) in self.watches.remove(&conn).unwrap_or_default() {
                        self.heap.post(&addr(watcher), lost(conn, id));
                    }
                    self.remote_watches.retain(|_, conns| {
                        conns.retain(|c| *c != conn);
                        !conns.is_empty()
                    });
                }
                Some(Event::Node(conn, body)) => self.node(conn, body).await,
                Some(Event::Gone(name)) => {
                    self.hosts.remove(&name);
                    for watcher in self.peer_watches.remove(&name).unwrap_or_default() {
                        self.heap.post(&addr(watcher), gone(host(&name)));
                    }
                }
                None => return Ok(ended),
            }
        }
    }

    /// Record a watch on something outside the heap, answering `true` if it was
    /// answered at once - which is what makes a watch race-free to set. Same
    /// dispatch as [`World::forward`]: the two must agree on what an address
    /// names.
    async fn watch(&mut self, watcher: Id, target: Val) -> bool {
        if let Some(Addr::Remote(conn, id)) = Addr::of(&target) {
            let (conn, id) = (*conn, *id);
            if !self.links.contains_key(&conn) {
                self.heap.post(&addr(watcher), lost(conn, id));
                return true;
            }
            // Both: the far node reports the task ending, the link dying is
            // what is left when there is no far node to ask. An answered watch
            // leaves the table, so the two cannot both fire.
            let waiting = self.watches.entry(conn).or_default();
            let first = !waiting.iter().any(|(_, far)| *far == id);
            waiting.push((watcher, id));
            if first {
                self.forward_to_link(conn, link::NODE, watch_of(id)).await;
            }
            return false;
        }
        match list_elements(&target).as_deref() {
            Some([tag, Val::Sym(name)]) if *tag == sym("host") => {
                let name = name.to_string();
                self.watch_host(watcher, name)
            }
            // An address this node cannot route names nothing that could die,
            // so there is nothing further to learn.
            _ => {
                self.heap.post(&addr(watcher), gone(target));
                true
            }
        }
    }

    /// The edge is the adapter dropping the receiving end, which `Sender::closed`
    /// waits on, so one waiter per name serves any number of watchers and
    /// nothing is ever polled.
    ///
    /// The waiter holds a port deliberately: it lives exactly as long as the
    /// adapter it watches, and lets go when the loop's inbox closes, dropping
    /// its clone of the peer's sender. An adapter parked on `recv` would
    /// otherwise never see its own inbox close and would outlive the world.
    fn watch_host(&mut self, watcher: Id, name: String) -> bool {
        let peer = self.hosts.get(&name).cloned();
        let (Some(peer), Some(back)) = (peer, self.back.upgrade()) else {
            self.heap.post(&addr(watcher), gone(host(&name)));
            return true;
        };
        let watchers = self.peer_watches.entry(name.clone()).or_default();
        let first = watchers.is_empty();
        watchers.push(watcher);
        if first {
            tokio::spawn(async move {
                tokio::select! {
                    () = peer.closed() => {
                        let _ = back.send(Event::Gone(name)).await;
                    }
                    () = back.closed() => {}
                }
            });
        }
        false
    }

    /// Nothing here reports whether the heap woke: the loop reaches this only
    /// after deciding to wait, so a mailbox is drained on the next round.
    async fn node(&mut self, conn: ConnId, body: Val) {
        match list_elements(&body).as_deref() {
            Some([verb, Val::Num(n)]) if *verb == sym("watch") => {
                let id = *n as Id;
                if self.heap.alive(id) {
                    self.remote_watches.entry(id).or_default().push(conn);
                } else {
                    // The round trip is the window, so a watch arriving late
                    // would otherwise be one that is never answered.
                    let why = pair(sym("throw"), sym("noproc"));
                    self.forward_to_link(conn, link::NODE, down_of(id, why))
                        .await;
                }
            }
            Some([verb, Val::Num(n), result]) if *verb == sym("down") => {
                let id = *n as Id;
                let target = Addr::Remote(conn, id).val();
                let mut waiting = self.watches.remove(&conn).unwrap_or_default();
                waiting.retain(|(watcher, far)| {
                    if *far != id {
                        return true;
                    }
                    let told = sched::down(target.clone(), result.clone());
                    self.heap.post(&addr(*watcher), told);
                    false
                });
                if !waiting.is_empty() {
                    self.watches.insert(conn, waiting);
                }
            }
            // Dropped rather than refused: the frame is from a node, not a
            // caller, so there is nothing to refuse it to.
            _ => {}
        }
    }

    /// The exit value crosses as an ordinary payload, so a watcher over a link
    /// learns what a local one would. When it cannot encode, the death is
    /// still reported: a watcher not told waits for ever, which is worse than
    /// being told the result could not come with it.
    async fn ended(&mut self, id: Id, result: &Val) {
        let Some(conns) = self.remote_watches.remove(&id) else {
            return;
        };
        for conn in conns {
            let carried = match link::departs(result, conn) {
                true => result.clone(),
                false => pair(sym("throw"), sym("third-party-address")),
            };
            self.forward_to_link(conn, link::NODE, down_of(id, carried))
                .await;
        }
    }

    /// Hand a message to the peer its address names, answering `true` if the
    /// answer went back into the heap rather than out of it. An address naming
    /// nothing is answered rather than dropped when it carries a request: a
    /// caller that can never be replied to should learn so at once.
    async fn forward(&mut self, to: Val, msg: Val) -> bool {
        if let Some(Addr::Remote(conn, id)) = Addr::of(&to) {
            return self.forward_to_link(*conn, *id as u64, msg).await;
        }
        match list_elements(&to).as_deref() {
            Some([tag, Val::Sym(name)]) if *tag == sym("host") => {
                let name = name.to_string();
                self.forward_to_host(&name, msg).await
            }
            _ => self.undeliverable(&msg, sym("not-local")),
        }
    }

    async fn forward_to_host(&mut self, name: &str, msg: Val) -> bool {
        let Some(tx) = self.hosts.get(name) else {
            return self.undeliverable(&msg, pair(sym("no-host"), sym(name)));
        };
        if tx.send(msg.clone()).await.is_err() {
            self.hosts.remove(name);
            return self.undeliverable(&msg, pair(sym("no-host"), sym(name)));
        }
        false
    }

    /// `send` refuses a non-data payload at the primitive, so a failure to
    /// encode here is a bug and is reported as one. An address naming a third
    /// node is a program's error and is answered.
    async fn forward_to_link(&mut self, conn: ConnId, to: u64, msg: Val) -> bool {
        let Some(tx) = self.links.get(&conn) else {
            return self.undeliverable(&msg, pair(sym("no-link"), Val::Num(conn as i64)));
        };
        if !link::departs(&msg, conn) {
            return self.undeliverable(&msg, sym("third-party-address"));
        }
        let body = wire::encode(&msg).expect("a payload is data");
        if tx.send(Frame { to, body }).await.is_err() {
            self.links.remove(&conn);
            return self.undeliverable(&msg, pair(sym("no-link"), Val::Num(conn as i64)));
        }
        false
    }

    fn undeliverable(&mut self, msg: &Val, why: Val) -> bool {
        let Some((from, id, _)) = request(msg) else {
            return false;
        };
        let thrown = Val::Pair(rc_val(sym("throw")), rc_val(why));
        self.heap.post(&from, reply(id, thrown));
        true
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::floor::host::Str;
    use crate::floor::machine::{Machine, NoMail, Store};
    use crate::floor::Env;

    fn val(src: &str) -> Val {
        let Val::Code(e) = crate::surface::compile(src).expect("compiles") else {
            panic!("not code")
        };
        Machine::new(Env::default(), e)
            .run(&mut Store::<NoMail>::default())
            .expect("no fault")
    }

    fn ok(v: Val) -> Val {
        pair(sym("ok"), v)
    }

    #[tokio::test]
    async fn a_world_runs_a_heap_that_needs_nothing() {
        let mut w = World::new(1000);
        let id = w.heap().spawn(val("(lambda t (_) (+ 1 2))"));
        assert_eq!(w.run().await.expect("no bug"), vec![(id, ok(Val::Num(3)))]);
    }

    #[tokio::test]
    async fn a_world_gives_up_when_nothing_could_wake_it() {
        let mut w = World::new(1000);
        w.heap().spawn(val("(lambda t (_) (receive 0))"));
        assert_eq!(w.run().await.expect("no bug"), vec![]);
    }

    #[tokio::test]
    async fn a_message_crosses_a_link() {
        let (near, far) = tokio::io::duplex(64);

        let mut b = World::new(1000);
        let there = b.heap().spawn(val("(lambda t (_) (receive))"));
        b.link(0, crate::link::bridge(0, far, b.port()));

        let mut a = World::new(1000);
        // The second `receive` is never answered: it keeps this side up
        // until the far side is done, so the frame is written rather than
        // dropped with the world that queued it.
        let here = a
            .heap()
            .spawn(val("(lambda t (_) (begin (send (receive) 'hi) (receive)))"));
        a.heap().post(&addr(here), Addr::Remote(0, there).val());
        a.link(0, crate::link::bridge(0, near, a.port()));

        let (ran_a, ran_b) = tokio::join!(a.run(), b.run());
        assert_eq!(ran_a.expect("no bug"), vec![]);
        assert_eq!(ran_b.expect("no bug"), vec![(there, ok(sym("hi")))]);
    }

    #[tokio::test]
    async fn a_reply_finds_its_way_back_across_a_link() {
        let (near, far) = tokio::io::duplex(64);

        let mut b = World::new(1000);
        let there = b
            .heap()
            .spawn(val("(lambda t (_) (send (car (receive)) 'pong))"));
        b.link(0, crate::link::bridge(0, far, b.port()));

        let mut a = World::new(1000);
        let here = a
            .heap()
            .spawn(val("(lambda t (me) (begin (send (receive) (cons me 'ping))
                                              (receive)))"));
        a.heap().post(&addr(here), Addr::Remote(0, there).val());
        a.link(0, crate::link::bridge(0, near, a.port()));

        let (ran_a, ran_b) = tokio::join!(a.run(), b.run());
        assert_eq!(ran_a.expect("no bug"), vec![(here, ok(sym("pong")))]);
        assert_eq!(ran_b.expect("no bug"), vec![(there, ok(Val::Nil))]);
    }

    #[tokio::test]
    async fn an_address_for_a_third_node_does_not_cross() {
        let (near, _far) = tokio::io::duplex(64);
        let mut w = World::new(1000);
        let id = w.heap().spawn(val("(lambda t (me)
               (let ((m (receive)))
                 (begin (send (car m) (list 'req me 7 (cdr m))) (receive))))"));
        w.heap().post(
            &addr(id),
            pair(Addr::Remote(0, 1).val(), Addr::Remote(1, 2).val()),
        );
        w.link(0, crate::link::bridge(0, near, w.port()));
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(
                id,
                ok(list(&[
                    sym("reply"),
                    Val::Num(7),
                    pair(sym("throw"), sym("third-party-address"))
                ]))
            )]
        );
    }

    #[tokio::test]
    async fn a_call_across_a_link_that_is_not_there_is_answered() {
        let mut w = World::new(1000);
        let id = w.heap().spawn(val("(lambda t (me)
               (begin (send (receive) (list 'req me 7 'anything))
                      (receive)))"));
        w.heap().post(&addr(id), Addr::Remote(7, 1).val());
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(
                id,
                ok(list(&[
                    sym("reply"),
                    Val::Num(7),
                    pair(sym("throw"), pair(sym("no-link"), Val::Num(7)))
                ]))
            )]
        );
    }

    #[tokio::test]
    async fn a_watch_across_a_link_fires_when_the_link_dies() {
        let (near, far) = tokio::io::duplex(64);
        let mut w = World::new(1000);
        let id = w
            .heap()
            .spawn(val("(lambda t (_) (begin (monitor (receive)) (receive)))"));
        w.heap().post(&addr(id), Addr::Remote(0, 1).val());
        w.link(0, crate::link::bridge(0, near, w.port()));
        drop(far);
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(
                id,
                ok(sched::down(
                    Addr::Remote(0, 1).val(),
                    pair(sym("throw"), sym("noconnection"))
                ))
            )]
        );
    }

    #[tokio::test]
    async fn a_watch_across_a_link_fires_when_the_far_task_ends() {
        let (near, far) = tokio::io::duplex(64);

        let mut b = World::new(1000);
        let there = b.heap().spawn(val("(lambda t (_) (receive))"));
        b.link(0, crate::link::bridge(0, far, b.port()));

        let mut a = World::new(1000);
        let here = a.heap().spawn(val("(lambda t (_) (let ((to (receive)))
                             (begin (monitor to) (send to 'bye) (receive))))"));
        a.heap().post(&addr(here), Addr::Remote(0, there).val());
        a.link(0, crate::link::bridge(0, near, a.port()));

        let (ran_a, ran_b) = tokio::join!(a.run(), b.run());
        assert_eq!(
            ran_a.expect("no bug"),
            vec![(
                here,
                ok(sched::down(Addr::Remote(0, there).val(), ok(sym("bye"))))
            )]
        );
        assert_eq!(ran_b.expect("no bug"), vec![(there, ok(sym("bye")))]);
    }

    #[tokio::test]
    async fn a_watch_on_a_far_task_that_has_ended_is_answered() {
        let (near, far) = tokio::io::duplex(64);

        let mut b = World::new(1000);
        // Ends before anything is said to it, so the watch cannot be in
        // time however the two loops interleave.
        let there = b.heap().spawn(val("(lambda t (_) 'done)"));
        // And something to keep that node up long enough to be asked. A
        // node whose last task ended is finished, and would answer by
        // dropping the link instead - which is `noconnection`, and true,
        // but not what is being tested.
        b.heap().spawn(val("(lambda t (_) (receive))"));
        b.link(0, crate::link::bridge(0, far, b.port()));

        let mut a = World::new(1000);
        let here = a
            .heap()
            .spawn(val("(lambda t (_) (begin (monitor (receive)) (receive)))"));
        a.heap().post(&addr(here), Addr::Remote(0, there).val());
        a.link(0, crate::link::bridge(0, near, a.port()));

        let (ran_a, ran_b) = tokio::join!(a.run(), b.run());
        assert_eq!(ran_b.expect("no bug"), vec![(there, ok(sym("done")))]);
        assert_eq!(
            ran_a.expect("no bug"),
            vec![(
                here,
                ok(sched::down(
                    Addr::Remote(0, there).val(),
                    pair(sym("throw"), sym("noproc"))
                ))
            )]
        );
    }

    #[tokio::test]
    async fn a_watch_across_a_link_that_is_not_there_is_answered() {
        let mut w = World::new(1000);
        let id = w
            .heap()
            .spawn(val("(lambda t (_) (begin (monitor (receive)) (receive)))"));
        w.heap().post(&addr(id), Addr::Remote(7, 1).val());
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(
                id,
                ok(sched::down(
                    Addr::Remote(7, 1).val(),
                    pair(sym("throw"), sym("noconnection"))
                ))
            )]
        );
    }

    #[tokio::test]
    async fn a_syscall_is_a_send_and_a_receive() {
        let mut w = World::new(1000);
        let id = w.heap().spawn(val("(lambda t (me)
               (begin (send '(host upper) (cons me 'ping)) (receive)))"));

        let (tx, mut rx) = channel(8);
        w.serve("upper", tx);
        let back = w.port();
        tokio::spawn(async move {
            let msg = rx.recv().await.expect("a request");
            let Val::Pair(reply_to, payload) = msg else {
                panic!("not a request")
            };
            let answer = pair(sym("pong"), Val::clone(&payload));
            back.send(Event::Post(Val::clone(&reply_to), answer))
                .await
                .expect("open");
        });

        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(id, ok(pair(sym("pong"), sym("ping"))))]
        );
    }

    #[tokio::test]
    async fn a_call_to_an_adapter_that_ends_raises_at_the_call_site() {
        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(
            "(task-loop self
                        (lambda h (st msg)
                          (cons 'stop (attempt (lambda () (call '(host quiet) 'hi)))))
                        0)",
        )
        .expect("reads");
        let entry = t
            .task(&crate::surface::desugar_body(&forms).expect("desugars"))
            .expect("no fault");

        let mut w = World::new(1_000_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("go"));

        let (tx, mut rx) = channel(8);
        w.serve("quiet", tx);
        tokio::spawn(async move {
            rx.recv().await.expect("a request");
        });

        let why = pair(
            sym("callee-down"),
            pair(host("quiet"), pair(sym("throw"), sym("noproc"))),
        );
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(id, ok(pair(sym("throw"), why)))]
        );
    }

    #[tokio::test]
    async fn an_object_task_calls_a_host_adapter() {
        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(
            "(task-loop self
                        (lambda h (st msg) (cons 'stop (call '(host echo) 'hi)))
                        0)",
        )
        .expect("reads");
        let entry = t
            .task(&crate::surface::desugar_body(&forms).expect("desugars"))
            .expect("no fault");

        let mut w = World::new(100_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("go"));

        let (tx, mut rx) = channel(8);
        w.serve("echo", tx);
        let back = w.port();
        tokio::spawn(async move {
            let msg = rx.recv().await.expect("a request");
            let (from, rid, body) = request(&msg).expect("a request");
            back.send(Event::Post(from, reply(rid, pair(sym("heard"), body))))
                .await
                .expect("open");
        });

        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(id, ok(pair(sym("heard"), sym("hi"))))]
        );
    }

    #[tokio::test]
    async fn a_protected_region_can_still_suspend() {
        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(
            "(task-loop self
                        (lambda h (st msg)
                          (cons 'stop (attempt (lambda ()
                                        (call '(host echo) 'hi)))))
                        0)",
        )
        .expect("reads");
        let entry = t
            .task(&crate::surface::desugar_body(&forms).expect("desugars"))
            .expect("no fault");

        let mut w = World::new(100_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("go"));

        let (tx, mut rx) = channel(8);
        w.serve("echo", tx);
        let back = w.port();
        tokio::spawn(async move {
            let msg = rx.recv().await.expect("a request");
            let (from, rid, body) = request(&msg).expect("a request");
            back.send(Event::Post(from, reply(rid, pair(sym("heard"), body))))
                .await
                .expect("open");
        });

        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(id, ok(pair(sym("ok"), pair(sym("heard"), sym("hi")))))]
        );
    }

    #[tokio::test]
    async fn a_compiled_caller_calls_a_compiled_callee_rather_than_unfolding_it() {
        let mut t = crate::tower::Tower::load().expect("loads");
        // Both `clambda`s run, so `down` is a compiled function by the time
        // `go` stages a call to it. Unfolding one is unfolding a recursion,
        // and the turn limit is what says so if this regresses.
        let forms = crate::floor::read::read(
            "(limit-turns self 100)
             (define down (clambda down (n) (if (< n 1) 0 (down (- n 1)))))
             (define go (clambda go (n) (down n)))
             (go 5)",
        )
        .expect("reads");
        let entry = t.script(&forms).expect("no fault");

        let mut w = World::new(10_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("run"));

        assert_eq!(w.run().await.expect("no bug"), vec![(id, ok(Val::Num(0)))]);
    }

    #[tokio::test]
    async fn a_file_runs_as_a_task_and_can_still_call() {
        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(
            "(define (twice x) (* 2 x))
             (call '(host echo) (twice 21))",
        )
        .expect("reads");
        let entry = t.script(&forms).expect("no fault");

        let mut w = World::new(100_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("run"));

        let (tx, mut rx) = channel(8);
        w.serve("echo", tx);
        let back = w.port();
        tokio::spawn(async move {
            let msg = rx.recv().await.expect("a request");
            let (from, rid, body) = request(&msg).expect("a request");
            back.send(Event::Post(from, reply(rid, body)))
                .await
                .expect("open");
        });

        assert_eq!(w.run().await.expect("no bug"), vec![(id, ok(Val::Num(42)))]);
    }

    #[tokio::test]
    async fn a_file_body_is_not_re_entered_by_a_message_that_arrives_during_it() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::Arc;

        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(
            "(define a (call '(host echo) 1))
             (define b (call '(host echo) 2))
             (+ a b)",
        )
        .expect("reads");
        let entry = t.script(&forms).expect("no fault");

        let mut w = World::new(100_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("run"));
        w.heap().post(&addr(id), sym("poke"));

        let (tx, mut rx) = channel(8);
        w.serve("echo", tx);
        let back = w.port();
        let seen = Arc::new(AtomicUsize::new(0));
        let count = seen.clone();
        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                count.fetch_add(1, Ordering::SeqCst);
                let (from, rid, body) = request(&msg).expect("a request");
                back.send(Event::Post(from, reply(rid, body)))
                    .await
                    .expect("open");
            }
        });

        assert_eq!(w.run().await.expect("no bug"), vec![(id, ok(Val::Num(3)))]);
        assert_eq!(seen.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn a_module_is_loaded_by_name_and_imported() {
        assert_eq!(
            loading(
                "(import (load \"arith.naj\"))
                 (down (up 5))",
                &[(
                    "arith.naj",
                    "(define base 10)
                     (list (cons 'up (lambda u (n) (+ n base)))
                           (cons 'down (lambda d (n) (- n base))))",
                )],
            )
            .await,
            ok(Val::Num(5))
        );
    }

    #[tokio::test]
    async fn a_module_may_load_another() {
        assert_eq!(
            loading(
                "(import (load \"outer.naj\"))
                 (twice 21)",
                &[
                    (
                        "outer.naj",
                        "(import (load \"inner.naj\"))
                         (list (cons 'twice (lambda t (n) (add n n))))",
                    ),
                    ("inner.naj", "(list (cons 'add (lambda a (x y) (+ x y))))"),
                ],
            )
            .await,
            ok(Val::Num(42))
        );
    }

    #[tokio::test]
    async fn a_module_cannot_see_the_scope_it_is_imported_into() {
        let v = loading(
            "(define secret 7)
             (import (load \"peek.naj\"))
             (get)",
            &[("peek.naj", "(list (cons 'get (lambda g () secret)))")],
        )
        .await;
        assert_eq!(v, pair(sym("throw"), pair(sym("unbound"), sym("secret"))));
    }

    #[tokio::test]
    async fn a_call_answered_with_a_throw_raises_at_the_call_site() {
        assert_eq!(
            loading("(load \"missing.naj\")", &[]).await,
            pair(sym("throw"), pair(sym("io"), Str::val("no such file")))
        );
    }

    #[tokio::test]
    async fn a_failed_call_is_catchable() {
        assert_eq!(
            loading(
                "(attempt (lambda () (load \"missing.naj\")))",
                &[("other.naj", "'()")],
            )
            .await,
            ok(pair(
                sym("throw"),
                pair(sym("io"), Str::val("no such file"))
            ))
        );
    }

    const ROOT: &str = "/store";

    #[tokio::test]
    async fn a_module_is_resolved_through_the_lock_file() {
        assert_eq!(
            loading(
                "(import (need 'arith))
                 (down (up 5))",
                &[
                    (
                        "naj.lock",
                        "((narju-lock 1)
                          (modules
                           (arith (url \"https://forge/thorn/naj-arith\")
                                  (rev \"a1b2c3\")
                                  (path \"forge/thorn/naj-arith\")
                                  (file \"arith.naj\"))))",
                    ),
                    (
                        "/store/forge/thorn/naj-arith/a1b2c3/arith.naj",
                        "(define base 10)
                         (list (cons 'up (lambda u (n) (+ n base)))
                               (cons 'down (lambda d (n) (- n base))))",
                    ),
                ],
            )
            .await,
            ok(Val::Num(5))
        );
    }

    #[tokio::test]
    async fn a_module_the_lock_does_not_carry_is_refused() {
        assert_eq!(
            loading(
                "(attempt (lambda () (need 'missing)))",
                &[("naj.lock", "((narju-lock 1) (modules))")],
            )
            .await,
            ok(pair(
                sym("throw"),
                list(&[sym("no-such-module"), sym("missing")])
            ))
        );
    }

    #[tokio::test]
    async fn a_locked_module_needs_another_through_its_own_lock() {
        assert_eq!(
            loading(
                "(import (need 'outer))
                 (twice 21)",
                &[
                    (
                        "naj.lock",
                        "((narju-lock 1)
                          (modules
                           (outer (url \"https://f/t/outer\") (rev \"r1\")
                                  (path \"f/t/outer\") (file \"o.naj\"))))",
                    ),
                    (
                        "/store/f/t/outer/r1/naj.lock",
                        "((narju-lock 1)
                          (modules
                           (inner (url \"https://f/t/inner\") (rev \"r2\")
                                  (path \"f/t/inner\") (file \"i.naj\"))))",
                    ),
                    (
                        "/store/f/t/outer/r1/o.naj",
                        "(import (need 'inner))
                         (list (cons 'twice (lambda t (n) (add n n))))",
                    ),
                    (
                        "/store/f/t/inner/r2/i.naj",
                        "(list (cons 'add (lambda a (x y) (+ x y))))",
                    ),
                ],
            )
            .await,
            ok(Val::Num(42))
        );
    }

    #[tokio::test]
    async fn two_modules_may_mean_different_things_by_one_name() {
        assert_eq!(
            loading(
                "(import (need 'left))
                 (import (need 'right))
                 (list (l) (r))",
                &[
                    (
                        "naj.lock",
                        "((narju-lock 1)
                          (modules
                           (left (url \"https://f/left\") (rev \"r1\")
                                 (path \"f/left\") (file \"m.naj\"))
                           (right (url \"https://f/right\") (rev \"r1\")
                                  (path \"f/right\") (file \"m.naj\"))))",
                    ),
                    (
                        "/store/f/left/r1/naj.lock",
                        "((narju-lock 1)
                          (modules
                           (dep (url \"https://f/one\") (rev \"a\")
                                (path \"f/one\") (file \"d.naj\"))))",
                    ),
                    (
                        "/store/f/right/r1/naj.lock",
                        "((narju-lock 1)
                          (modules
                           (dep (url \"https://f/two\") (rev \"b\")
                                (path \"f/two\") (file \"d.naj\"))))",
                    ),
                    (
                        "/store/f/left/r1/m.naj",
                        "(import (need 'dep))
                         (list (cons 'l (lambda l () (it))))",
                    ),
                    (
                        "/store/f/right/r1/m.naj",
                        "(import (need 'dep))
                         (list (cons 'r (lambda r () (it))))",
                    ),
                    ("/store/f/one/a/d.naj", "(list (cons 'it (lambda i () 1)))",),
                    ("/store/f/two/b/d.naj", "(list (cons 'it (lambda i () 2)))",),
                ],
            )
            .await,
            ok(list(&[Val::Num(1), Val::Num(2)]))
        );
    }

    #[tokio::test]
    async fn a_module_may_be_more_than_one_file() {
        assert_eq!(
            loading(
                "(import (need 'lib))
                 (top)",
                &[
                    (
                        "naj.lock",
                        "((narju-lock 1)
                          (modules
                           (lib (url \"https://f/lib\") (rev \"r1\")
                                (path \"f/lib\") (file \"src/m.naj\"))))",
                    ),
                    (
                        "/store/f/lib/r1/naj.lock",
                        "((narju-lock 1)
                          (modules
                           (dep (url \"https://f/dep\") (rev \"a\")
                                (path \"f/dep\") (file \"d.naj\"))))",
                    ),
                    (
                        "/store/f/lib/r1/src/m.naj",
                        "(import (load \"src/helper.naj\"))
                         (list (cons 'top (lambda t () (helped))))",
                    ),
                    (
                        "/store/f/lib/r1/src/helper.naj",
                        "(import (need 'dep))
                         (list (cons 'helped (lambda h () (it))))",
                    ),
                    ("/store/f/dep/a/d.naj", "(list (cons 'it (lambda i () 7)))"),
                ],
            )
            .await,
            ok(Val::Num(7))
        );
    }

    #[tokio::test]
    async fn a_module_with_no_lock_of_its_own_cannot_need() {
        assert_eq!(
            loading(
                "(attempt (lambda () (need 'solo)))",
                &[
                    (
                        "naj.lock",
                        "((narju-lock 1)
                          (modules
                           (solo (url \"https://f/solo\") (rev \"r1\")
                                 (path \"f/solo\") (file \"s.naj\"))))",
                    ),
                    (
                        "/store/f/solo/r1/s.naj",
                        "(import (need 'anything))
                         (list)",
                    ),
                ],
            )
            .await,
            ok(pair(
                sym("throw"),
                pair(sym("io"), Str::val("no such file"))
            ))
        );
    }

    #[tokio::test]
    async fn a_lock_from_a_later_format_is_refused() {
        assert_eq!(
            loading(
                "(attempt (lambda () (need 'x)))",
                &[("naj.lock", "((narju-lock 2) (modules))")],
            )
            .await,
            ok(pair(
                sym("throw"),
                list(&[sym("lock-version"), Val::Num(2)])
            ))
        );
    }

    async fn loading(src: &str, files: &[(&str, &str)]) -> Val {
        let mut t = crate::tower::Tower::load().expect("loads");
        let forms = crate::floor::read::read(src).expect("reads");
        let entry = t.script(&forms).expect("no fault");

        let mut w = World::new(10_000_000);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("run"));

        let (tx, mut rx) = channel(8);
        w.serve("files", tx);
        let back = w.port();
        let table: Vec<(String, String)> = files
            .iter()
            .map(|(p, s)| ((*p).to_string(), (*s).to_string()))
            .collect();
        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                let (from, rid, body) = request(&msg).expect("a request");
                let asked = list_elements(&body).expect("a list");
                let v = match asked.as_slice() {
                    [verb] if *verb == sym("root") => Str::val(ROOT),
                    [verb, path] if *verb == sym("read") => {
                        let want = Str::of(path).expect("a path");
                        match table.iter().find(|(p, _)| p == want) {
                            Some((_, text)) => Str::val(text.clone()),
                            None => pair(sym("throw"), pair(sym("io"), Str::val("no such file"))),
                        }
                    }
                    _ => panic!("unexpected {body}"),
                };
                if back.send(Event::Post(from, reply(rid, v))).await.is_err() {
                    return;
                }
            }
        });

        let ended = w.run().await.expect("no bug");
        assert_eq!(ended.len(), 1);
        ended[0].1.clone()
    }

    #[tokio::test]
    async fn the_prompt_carries_its_environment_from_line_to_line() {
        let said = console(
            &[
                "(define (double x) (* 2 x))",
                "(double 21)",
                "",
                "(car 3)",
                "\"hi\"",
            ],
            100_000,
        )
        .await;
        assert_eq!(
            said,
            ["'double", "42", "('error 'wrong-type . 'car)", "\"hi\""]
        );
    }

    #[tokio::test]
    async fn the_prompt_takes_an_import_a_line_at_a_time() {
        let said = console(
            &["(import (list (cons 'z 9) (cons 'w 4)))", "(+ z w)"],
            100_000,
        )
        .await;
        assert_eq!(said, ["('z 'w)", "13"]);
    }

    async fn console(lines: &[&str], budget: usize) -> Vec<String> {
        use std::sync::{Arc, Mutex};

        let mut t = crate::tower::Tower::load().expect("loads");
        let entry = t.repl().expect("no fault");

        let mut w = World::new(budget);
        let id = w.heap().spawn(entry);
        w.heap().post(&addr(id), sym("go"));

        let (tx, mut rx) = channel(8);
        w.serve("stdin", tx.clone());
        w.serve("stdout", tx);
        let back = w.port();
        let said = Arc::new(Mutex::new(Vec::new()));
        let out = Arc::clone(&said);
        let mut script = lines.iter().map(|s| s.to_string()).collect::<Vec<_>>();
        script.reverse();

        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                let (from, rid, body) = request(&msg).expect("a request");
                let v = match list_elements(&body).as_deref() {
                    Some([verb, _]) if *verb == sym("read") => match script.pop() {
                        Some(line) => Str::val(line),
                        None => sym("eof"),
                    },
                    Some([verb, arg]) if *verb == sym("line") => {
                        out.lock()
                            .expect("no panic")
                            .push(Str::of(arg).map_or_else(|| arg.to_string(), str::to_string));
                        sym("ok")
                    }
                    _ => panic!("unexpected {body}"),
                };
                if back.send(Event::Post(from, reply(rid, v))).await.is_err() {
                    return;
                }
            }
        });

        assert_eq!(w.run().await.expect("no bug"), vec![(id, ok(sym("bye")))]);
        let held = said.lock().expect("no panic");
        held.clone()
    }

    #[tokio::test]
    async fn a_call_to_a_peer_that_is_not_there_is_answered() {
        let mut w = World::new(1000);
        let id = w.heap().spawn(val("(lambda t (me)
               (begin (send '(host nowhere) (list 'req me 7 'anything))
                      (receive)))"));
        assert_eq!(
            w.run().await.expect("no bug"),
            vec![(
                id,
                ok(list(&[
                    sym("reply"),
                    Val::Num(7),
                    pair(sym("throw"), pair(sym("no-host"), sym("nowhere")))
                ]))
            )]
        );
    }
}