broker: remove op deletes an actor

{"type":"remove","actor"} shuts the actor down normally, drops its
mailbox and supervisor, and deletes /actor/<id>/. Native actors cannot
be removed. Test in cli/tests/host.rs, documented in PROTOCOL.md.

.gitea/workflows/ci.yml

@@ -1,4 +1,4 @@
-# Runs on a self-hosted Nix runner (label `nixos`, host executor).
+# Runs on the Nix runner on nexus (label `nixos`, host executor).
 name: ci
 on: [push, pull_request]
 

PROTOCOL.md

@@ -68,7 +68,7 @@ Each instance sees three folders:
 | `/actor` | read-only | the `actor/` folder of the world |
 
 `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
-(installed version, running, subscriptions, tapped), `source/`, `public/`, `data/`,
+(installed version, running, subscriptions), `source/`, `public/`, `data/`,
 `stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors.
 
 Files are plain POSIX files: a write is visible at once. To publish a file
@@ -85,13 +85,6 @@ Every actor can read and write them. The host turns each file operation into a
 message from `fs` to `<id>` and waits up to 10 seconds for the reply. `path`
 is relative to `fs/` and starts with `/` (the folder itself is `/`).
 
-To make this virtual directory discoverable, an actor that serves it declares
-`Actor-served files` and `fs_read` in its leading `//!` documentation in
-`source/src/main.rs`. The host uses that declaration when listing
-`/actor/<id>/`; it does not probe every actor (which could stall on actors that
-do not handle filesystem messages). This is discovery metadata only: opening
-and reading the path still requires a valid `fs_read` reply.
-
 | Request | Reply |
 | --- | --- |
 | `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` |
@@ -117,7 +110,7 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
 Requests to `broker`. The answer is a message whose `reply_to` is the ID of
 the request: the request's `type` in past tense (`spawned`, `upgraded`,
 `rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
-`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
+`scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
 
 | `type` | Fields | Effect |
 | --- | --- | --- |
@@ -127,30 +120,11 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
 | `stop`, `start` | `actor` | stop or start an actor |
 | `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
 | `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
-| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
 | `scan` | | boot new or changed `actor.wasm` files |
 | `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
 
 Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
 
-**Taps.** After `tap`, the sender gets a copy of every message that the
-broker routes (direct messages, replies, and publishes, between any actors,
-including its own), for debugging, logging, or a live view of the world:
-
-```json
-{"type": "tapped", "message": {"id": "m-17", "from": "console", "to": "agent", "reply_to": null, "body": {...}}, "ts": 1792132579067}
-```
-
-- `to` is `null` for a publish: one copy per publish, not per subscriber.
-- `ts` is the time of routing, in milliseconds since the Unix epoch.
-- Copies come from `broker`. They are not routed themselves, so they are
-  never tapped, and each message is tapped once per tapper.
-- A copy that does not fit into the tapper's mailbox is dropped; routing
-  never waits for a tapper, and nobody gets `undeliverable` for a copy.
-- Never answer a copy: the answer would be tapped again, forever.
-- Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and
-  survives restarts and upgrades. `untap` ends it.
-
 **Versions.** A new version (from `spawn`, `upgrade`, or a changed
 `actor.wasm` on disk) is tested first, while the old version keeps running.
 Only when its tests pass does the broker copy it to `versions/<n>/`

README.md

@@ -10,11 +10,9 @@ change its own code.
 
 ## Start
 
-You need [Nix](https://nixos.org/download) with flakes, and an API token for
+You need [Nix](https://nixos.org/download) with flakes.
-an LLM (OpenAI Responses or Anthropic Messages API).
 
 ```sh
-git clone https://gitea.surma.technology/surma/sausage && cd sausage
 nix run -- init     # creates ./world and builds the agent (a few minutes)
 nix run -- run      # starts the world and a console
 ```
@@ -28,13 +26,12 @@ folder as an argument, for example `nix run -- run my-world`.
 To keep a world on the newest version, run it in a loop:
 
 ```sh
-nix run .#loop -- my-world
+nix run .#loop -- my-world            # or .#loop-internal (Gitea on the LAN)
 ```
 
-The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
+The loop runs `nix run --refresh git+ssh://…/surma/sausage -- run my-world`.
 When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop
-fetches the newest commit and starts it again. `/quit` ends the loop. Set
+fetches the newest commit and starts it again. `/quit` ends the loop.
-`SAUSAGE_FLAKE` to run another flake URL instead, for example your fork.
 
 In the console:
 

cli/src/broker.rs

@@ -1,7 +1,6 @@
-//! The broker: message routing, mailboxes, subscriptions, taps, the version
+//! The broker: message routing, mailboxes, subscriptions, the version store,
-//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
+//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
-//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
+//! remove, subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
-//! runs each WASM actor.
 //!
 //! The broker accepts only `.wasm` files. It does not know any source language.
 //! The world folder is the source of truth: `scan` boots every
@@ -134,23 +133,17 @@ impl Broker {
             Some(own) => format!("{from}/{own}"),
             None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
         };
-        let (recipients, tappers) = {
-            let metas = self.metas.lock().unwrap();
         let recipients: Vec<String> = match to {
             Some(to) => vec![to.to_string()],
-                None => metas
+            None => self
+                .metas
+                .lock()
+                .unwrap()
                 .iter()
                 .filter(|(_, m)| m.subscriptions.contains(from))
                 .map(|(id, _)| id.clone())
                 .collect(),
         };
-            let tappers: Vec<String> = metas.iter().filter(|(_, m)| m.tapped).map(|(id, _)| id.clone()).collect();
-            (recipients, tappers)
-        };
-        if !tappers.is_empty() {
-            let message = json!({"id": id, "from": from, "to": to, "reply_to": reply_to, "body": body});
-            self.tap(&tappers, message);
-        }
         for to in recipients {
             self.deliver(Envelope {
                 id: id.clone(),
@@ -163,25 +156,6 @@ impl Broker {
         id
     }
 
-    /// Give every tapper a copy of one routed message. Copies skip `send`, so
-    /// they are never tapped themselves; a full mailbox drops the copy.
-    fn tap(&self, tappers: &[String], message: Value) {
-        let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64);
-        let body = json!({"type": "tapped", "message": message, "ts": ts});
-        let mailboxes = self.mailboxes.lock().unwrap();
-        for tapper in tappers {
-            if let Some(tx) = mailboxes.get(tapper) {
-                let _ = tx.try_send(Envelope {
-                    id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
-                    from: "broker".into(),
-                    to: tapper.clone(),
-                    reply_to: None,
-                    body: body.clone(),
-                });
-            }
-        }
-    }
-
     /// Route one stdout line of a WASM actor.
     pub fn route_line(&self, from: &str, line: &str) {
         let line = line.trim();
@@ -523,12 +497,6 @@ impl Broker {
                 })?;
                 Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
             }
-            "tap" | "untap" => {
-                let on = ty == "tap";
-                self.update_meta(&env.from, |m| m.tapped = on)?;
-                let answer = if on { "tapping" } else { "untapped" };
-                Ok(json!({"type": answer}))
-            }
             "scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
             "exit" => {
                 self.log("broker", &format!("exit requested by {}", env.from));
@@ -536,7 +504,7 @@ impl Broker {
                 self.exit.notify_one();
                 Ok(json!({"type": "exiting"}))
             }
-            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
+            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, scan, exit"),
         }
     }
 

cli/src/vfs.rs

@@ -4,8 +4,7 @@
 //! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
 //! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the
 //! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
-//! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
+//! the reply. Every other path goes to `wasmtime-wasi` unchanged.
-//! actor directory listing advertises `fs/` when its source docs declare it.
 //!
 //! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
 //! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
@@ -15,7 +14,7 @@
 //! A file is read once when it is opened. After a change, it is written back
 //! in one `fs_write` when it is synced or closed.
 
-use std::collections::HashMap;
+use std::collections::HashSet;
 use std::hash::{Hash, Hasher};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, Weak};
@@ -42,14 +41,13 @@ pub struct VfsState {
     broker: Weak<Broker>,
     /// The actor of this instance, for notes in its `stderr.log`.
     me: String,
-    /// Native directory descriptors and their absolute paths. Keeping every
+    /// Descriptors of the `/actor` preopen.
-    /// `/actor` child (not only the preopen) lets `fs/` work after traversal.
+    roots: HashSet<u32>,
-    dirs: HashMap<u32, String>,
 }
 
 impl VfsState {
     pub fn new(broker: Weak<Broker>, me: String) -> Self {
-        Self { broker, me, dirs: HashMap::new() }
+        Self { broker, me, roots: HashSet::new() }
     }
 }
 
@@ -201,33 +199,6 @@ fn join(base: &str, rel: &str) -> String {
     format!("/{}", parts.join("/"))
 }
 
-/// Split a global path under an actor's virtual `fs/` into its owner and
-/// actor-relative path.
-fn actor_fs_path(full: &str) -> Option<(String, String)> {
-    let rest = full.strip_prefix("/actor/")?;
-    let (actor, rest) = rest.split_once('/')?;
-    let (fs, path) = rest.split_once('/').unwrap_or((rest, ""));
-    (fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{path}")))
-}
-
-fn actor_root(path: &str) -> Option<&str> {
-    let actor = path.strip_prefix("/actor/")?;
-    (!actor.is_empty() && !actor.contains('/')).then_some(actor)
-}
-
-/// An actor advertises its virtual filesystem in its leading `//!` docs.
-/// This is discovery metadata; actual reads still go through `fs_read`.
-fn declares_actor_fs(source: &str) -> bool {
-    let docs = source
-        .lines()
-        .take_while(|line| line.trim().starts_with("//!"))
-        .map(|line| line.trim().trim_start_matches("//!").trim())
-        .collect::<Vec<_>>()
-        .join(" ")
-        .to_ascii_lowercase();
-    docs.contains("actor-served files") && docs.contains("fs_read")
-}
-
 fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
     types::DescriptorStat {
         type_,
@@ -295,14 +266,6 @@ impl VfsView<'_> {
         Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
     }
 
-    fn actor_serves_fs(&self, actor: &str) -> bool {
-        self.state
-            .broker
-            .upgrade()
-            .and_then(|broker| std::fs::read_to_string(broker.world.source_dir(actor).join("src/main.rs")).ok())
-            .is_some_and(|source| declares_actor_fs(&source))
-    }
-
     fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
         self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
     }
@@ -319,27 +282,16 @@ impl VfsView<'_> {
         match self.virt(fd) {
             Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
             Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
-            None => {
+            None if self.state.roots.contains(&fd.rep()) => {
-                let base = self.state.dirs.get(&fd.rep())?;
+                let full = join("/", path);
-                actor_fs_path(&join(base, path))
+                let mut parts = full.splitn(4, '/').skip(1);
+                let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or(""));
+                (fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}")))
             }
+            None => None,
         }
     }
 
-    async fn native_directory_entries(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Vec<DirectoryEntry>> {
-        let stream = HostDescriptor::read_directory(&mut self.inner, fd).await?;
-        let stream_rep = stream.rep();
-        let mut entries = Vec::new();
-        loop {
-            match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, Resource::new_borrow(stream_rep)).await? {
-                Some(entry) => entries.push(entry),
-                None => break,
-            }
-        }
-        HostDirectoryEntryStream::drop(&mut self.inner, stream).map_err(|_| ErrorCode::Io)?;
-        Ok(entries)
-    }
-
     fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
         let r = self.inner.table.push(v)?;
         Ok(Resource::new_own(r.rep()))
@@ -391,7 +343,7 @@ impl preopens::Host for VfsView<'_> {
         let dirs = preopens::Host::get_directories(&mut self.inner)?;
         for (fd, name) in &dirs {
             if name == "/actor" {
-                self.state.dirs.insert(fd.rep(), "/actor".to_string());
+                self.state.roots.insert(fd.rep());
             }
         }
         Ok(dirs)
@@ -501,19 +453,7 @@ impl HostDescriptor for VfsView<'_> {
                 Node::File(_) => Err(ErrorCode::NotDirectory.into()),
             },
             Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
-            None => {
+            None => HostDescriptor::read_directory(&mut self.inner, fd).await,
-                let actor = self.state.dirs.get(&fd.rep()).and_then(|path| actor_root(path)).map(str::to_string);
-                if actor.as_deref().is_some_and(|id| self.actor_serves_fs(id)) {
-                    let mut entries = self.native_directory_entries(fd).await?;
-                    if !entries.iter().any(|entry| entry.name == "fs") {
-                        entries.push(DirectoryEntry { type_: DescriptorType::Directory, name: "fs".to_string() });
-                    }
-                    let r = self.inner.table.push(VDir(entries.into_iter()))?;
-                    Ok(Resource::new_own(r.rep()))
-                } else {
-                    HostDescriptor::read_directory(&mut self.inner, fd).await
-                }
-            }
         }
     }
 
@@ -585,16 +525,7 @@ impl HostDescriptor for VfsView<'_> {
     ) -> FsResult<Resource<types::Descriptor>> {
         match self.target(&fd, &path) {
             Some((actor, path)) => self.open(actor, path, oflags, flags).await,
-            None => {
+            None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await,
-                let child_path = self.state.dirs.get(&fd.rep()).map(|base| join(base, &path));
-                let opened = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await?;
-                if let Some(path) = child_path {
-                    if HostDescriptor::get_type(&mut self.inner, Resource::new_borrow(opened.rep())).await? == DescriptorType::Directory {
-                        self.state.dirs.insert(opened.rep(), path);
-                    }
-                }
-                Ok(opened)
-            }
         }
     }
 
@@ -607,7 +538,7 @@ impl HostDescriptor for VfsView<'_> {
             }
             return Ok(());
         }
-        self.state.dirs.remove(&fd.rep());
+        self.state.roots.remove(&fd.rep());
         HostDescriptor::drop(&mut self.inner, fd)
     }
 
@@ -767,17 +698,6 @@ fn base64_decode(text: &str) -> Option<Vec<u8>> {
 mod tests {
     use super::*;
 
-    #[test]
-    fn actor_fs_paths_and_discovery_docs() {
-        assert_eq!(actor_fs_path("/actor/brain/fs"), Some(("brain".to_string(), "/".to_string())));
-        assert_eq!(actor_fs_path("/actor/brain/fs/docs/readme.md"), Some(("brain".to_string(), "/docs/readme.md".to_string())));
-        assert_eq!(actor_fs_path("/actor/brain/public/fs/file"), None);
-        assert_eq!(actor_root("/actor/brain"), Some("brain"));
-        assert_eq!(actor_root("/actor/brain/fs"), None);
-        assert!(declares_actor_fs("//! brain\n//! Actor-served files include fs_read.\nfn main() {}"));
-        assert!(!declares_actor_fs("//! actor without fs support\nfn main() {}"));
-    }
-
     #[test]
     fn base64_round_trip() {
         for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {

cli/src/world.rs

@@ -5,7 +5,7 @@
 //! <world>/
 //!   actor/<id>/
 //!     actor.wasm        the current version
-//!     actor.json        version, running, subscriptions, tapped (broker)
+//!     actor.json        version, running, subscriptions (broker)
 //!     source/           source of the current version
 //!     public/           written by the actor, read-only for others
 //!     data/             state, writable only by the actor
@@ -43,9 +43,6 @@ pub struct ActorMeta {
     /// The actors whose published messages this actor receives.
     #[serde(default)]
     pub subscriptions: BTreeSet<String>,
-    /// Whether this actor gets a copy of every routed message (`tap`).
-    #[serde(default)]
-    pub tapped: bool,
 }
 
 fn yes() -> bool {

cli/tests/host.rs

@@ -246,71 +246,6 @@ async fn actors_serve_files_under_fs() {
     host.stop().await;
 }
 
-#[tokio::test]
-async fn actor_fs_is_discoverable_only_when_its_source_docs_advertise_it() {
-    let world = create_world("actor-fs-discovery");
-    let wasm = toolbox_wasm().clone();
-    let (host, mut mailbox) = start(&world).await;
-
-    let advertised = passing_source(&world, "advertised-fs");
-    std::fs::create_dir_all(advertised.join("src")).unwrap();
-    std::fs::write(
-        advertised.join("src/main.rs"),
-        "//! test actor\n//! Actor-served files: implements fs_read and fs_dir.\n",
-    )
-    .unwrap();
-    spawn_type(&host, &mut mailbox, "served", &wasm, &advertised).await;
-
-    let undocumented = passing_source(&world, "undocumented-fs");
-    std::fs::create_dir_all(undocumented.join("src")).unwrap();
-    std::fs::write(undocumented.join("src/main.rs"), "//! test actor without a virtual filesystem\n").unwrap();
-    spawn_type(&host, &mut mailbox, "plain", &wasm, &undocumented).await;
-
-    let reader_source = passing_source(&world, "fs-reader");
-    spawn_type(&host, &mut mailbox, "reader", &wasm, &reader_source).await;
-
-    let served_entries = ask_type(
-        &host,
-        &mut mailbox,
-        "reader",
-        json!({"type": "list", "path": "/actor/served"}),
-        "entries",
-    )
-    .await;
-    assert!(served_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
-
-    let plain_entries = ask_type(
-        &host,
-        &mut mailbox,
-        "reader",
-        json!({"type": "list", "path": "/actor/plain"}),
-        "entries",
-    )
-    .await;
-    assert!(!plain_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
-
-    let path = "/actor/served/fs/readme.txt";
-    ask_type(
-        &host,
-        &mut mailbox,
-        "reader",
-        json!({"type": "write", "path": path, "text": "virtual document"}),
-        "ok",
-    )
-    .await;
-    let document = ask_type(
-        &host,
-        &mut mailbox,
-        "reader",
-        json!({"type": "read", "path": path}),
-        "content",
-    )
-    .await;
-    assert_eq!(document.body["text"], "virtual document");
-
-    host.stop().await;
-}
-
 #[tokio::test]
 async fn files_are_plain_posix_files() {
     let (_, host, mut mailbox) = filesystem_world("posix").await;
@@ -610,8 +545,7 @@ async fn unknown_broker_request_lists_the_known_ones() {
     let (host, mut mailbox) = start(&world).await;
     let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
     assert_eq!(unknown.body["type"], "error");
-    let reason = unknown.body["reason"].as_str().unwrap();
+    assert!(unknown.body["reason"].as_str().unwrap().contains("remove"));
-    assert!(reason.contains("remove") && reason.contains("tap"), "{reason}");
     host.stop().await;
 }
 
@@ -863,114 +797,3 @@ async fn running_actors_have_sockets_and_candidates_do_not() {
     assert_eq!(answer.body, json!({"type": "tcp", "line": "echo: hi"}));
     host.stop().await;
 }
-
-/// Every message that arrives until the mailbox is quiet for `quiet`.
-async fn drain(mailbox: &mut mpsc::Receiver<Envelope>, quiet: Duration) -> Vec<Envelope> {
-    let mut messages = Vec::new();
-    while let Ok(Some(message)) = tokio::time::timeout(quiet, mailbox.recv()).await {
-        messages.push(message);
-    }
-    messages
-}
-
-#[tokio::test]
-async fn tap_copies_every_routed_message_once() {
-    let world = create_world("tap");
-    let wasm = toolbox_wasm().clone();
-    let (host, mut mailbox) = start(&world).await;
-    let source = passing_source(&world, "source");
-    spawn_type(&host, &mut mailbox, "tapped", &wasm, &source).await;
-    let subscribe = json!({"type": "subscribe", "actor": "tapped"});
-    assert_eq!(request(&host, &mut mailbox, subscribe).await.body["type"], "subscribed");
-    assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
-
-    let echo = json!({"type": "echo", "value": "hi"});
-    let echo_id = host.broker.send("console", Some("tapped"), None, None, echo.clone());
-    let publish = json!({"type": "publish", "value": "news"});
-    host.broker.send("console", Some("tapped"), None, None, publish);
-    let messages = drain(&mut mailbox, Duration::from_secs(2)).await;
-    let copies: Vec<&Value> = messages
-        .iter()
-        .filter(|m| m.from == "broker" && m.body["type"] == "tapped")
-        .map(|m| &m.body)
-        .collect();
-    // 4 copies (2 requests, 1 reply, 1 publish) + the reply + the publish itself.
-    assert_eq!((copies.len(), messages.len()), (4, 6), "{messages:?}");
-    for copy in &copies {
-        assert!(copy["ts"].as_u64().unwrap() > 0, "{copy}");
-    }
-    let count = |want: Value| copies.iter().filter(|c| c["message"] == want).count();
-    let request_copy = json!({"id": echo_id, "from": "console", "to": "tapped", "reply_to": null, "body": echo});
-    assert_eq!(count(request_copy), 1, "{copies:?}");
-    let reply_copies = copies
-        .iter()
-        .filter(|c| c["message"]["reply_to"] == echo_id.as_str() && c["message"]["from"] == "tapped" && c["message"]["to"] == "console")
-        .count();
-    assert_eq!(reply_copies, 1, "{copies:?}");
-    let publish_copy = copies.iter().find(|c| c["message"]["body"]["type"] == "published").expect("publish not tapped");
-    assert_eq!(publish_copy["message"]["from"], "tapped");
-    assert_eq!(publish_copy["message"]["to"], Value::Null);
-
-    assert_eq!(request(&host, &mut mailbox, json!({"type": "untap"})).await.body["type"], "untapped");
-    let echo_id = host.broker.send("console", Some("tapped"), None, None, json!({"type": "echo", "value": "quiet"}));
-    let messages = drain(&mut mailbox, Duration::from_secs(1)).await;
-    assert_eq!(messages.len(), 1, "{messages:?}");
-    assert_eq!(messages[0].reply_to.as_deref(), Some(echo_id.as_str()));
-    host.stop().await;
-}
-
-#[tokio::test]
-async fn tap_survives_a_restart() {
-    let world = create_world("tap-restart");
-    let (host, mut mailbox) = start(&world).await;
-    assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
-    host.stop().await;
-    assert!(world.read_meta("console").tapped);
-
-    let (host, mut mailbox) = start(&world).await;
-    let id = host.broker.send("console", Some("nobody"), None, None, json!({"type": "hello"}));
-    let copy = tokio::time::timeout(Duration::from_secs(10), async {
-        loop {
-            let m = mailbox.recv().await.expect("console mailbox closed");
-            if m.body["type"] == "tapped" && m.body["message"]["id"] == id.as_str() {
-                return m;
-            }
-        }
-    })
-    .await
-    .expect("no tapped copy after the restart");
-    assert_eq!(copy.body["message"]["to"], "nobody");
-    host.stop().await;
-}
-
-#[tokio::test]
-async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
-    let world = create_world("tap-full");
-    let (host, mut mailbox) = start(&world).await;
-    // "ghost" taps and never reads its mailbox.
-    let mut ghost = host.mailbox("ghost");
-    let mut sink = host.mailbox("sink");
-    host.broker.send("ghost", Some("broker"), None, None, json!({"type": "tap"}));
-    tokio::time::timeout(Duration::from_secs(10), async {
-        while !world.read_meta("ghost").tapped {
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("tap was not saved");
-
-    let total = sausage::broker::MAILBOX_SIZE + 500;
-    for n in 0..total {
-        host.broker.send("console", Some("sink"), None, None, json!({"type": "n", "n": n}));
-        assert_eq!(sink.try_recv().unwrap().body["n"], n);
-    }
-    assert!(mailbox.try_recv().is_err(), "a dropped copy reported something to the sender");
-    let mut held = 0;
-    while ghost.try_recv().is_ok() {
-        held += 1;
-    }
-    assert_eq!(held, sausage::broker::MAILBOX_SIZE);
-    host.broker.send("console", Some("sink"), None, None, json!({"type": "after"}));
-    assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
-    host.stop().await;
-}

flake.nix

@@ -36,14 +36,15 @@
           meta.mainProgram = "sausage";
         };
         # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
-        # from the public repo, again and again: after a crash or a broker
+        # from the Gitea repo, again and again: after a crash or a broker
         # `exit` request it refreshes the flake and starts the host again,
         # so the world updates itself. `/quit` (exit code 0) ends the loop.
-        # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
+        # `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN.
-        # e.g. a fork or a local checkout).
+        publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage";
-        repoUrl = "git+https://gitea.surma.technology/surma/sausage";
+        internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage";
-        loop = pkgs.writeShellScriptBin "sausage-loop" ''
+        loop = name: url: pkgs.writeShellScriptBin name ''
-          url="''${SAUSAGE_FLAKE:-${repoUrl}}"
+          url="${url}"
+          if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
           if [ $# -eq 0 ]; then set -- world; fi
           while true; do
             nix run --refresh "$url#default" -- run "$@"
@@ -71,7 +72,8 @@
         });
 
         apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
-        apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
+        apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; };
+        apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
 
         devShells.default = pkgs.mkShell {
           packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];