Run without a clone: bare sausage creates ./world if needed and runs it

- cli: the subcommand is optional; without one, sausage runs ./world and
  runs init first when ./world is not a world yet (test: cli/tests/cli.rs).
- flake: .#loop without a folder runs bare sausage.
- README: nix run git+https://gitea.surma.technology/surma/sausage(#loop),
  clone only for working on Sausage.

.gitea/workflows/ci.yml

@@ -1,4 +1,4 @@
-# Runs on the Nix runner on nexus (label `nixos`, host executor).
+# Runs on a self-hosted Nix runner (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), `source/`, `public/`, `data/`,
+(installed version, running, subscriptions, tapped), `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
@@ -110,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`,
-`scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
+`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
 
 | `type` | Fields | Effect |
 | --- | --- | --- |
@@ -120,11 +120,30 @@ 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,28 +10,43 @@ change its own code.
 
 ## Start
 
-You need [Nix](https://nixos.org/download) with flakes.
+You need [Nix](https://nixos.org/download) with flakes, and an API token for
+an LLM (OpenAI Responses or Anthropic Messages API). No clone needed:
 
 ```sh
-nix run -- init     # creates ./world and builds the agent (a few minutes)
-nix run -- run      # starts the world and a console
+nix run git+https://gitea.surma.technology/surma/sausage
+```
+
+This creates a world in `./world` the first time (it builds the agent, which
+takes a few minutes) and then starts it with a console. Later runs start the
+existing world. The two steps also exist on their own:
+
+```sh
+nix run git+https://gitea.surma.technology/surma/sausage -- init   # create ./world
+nix run git+https://gitea.surma.technology/surma/sausage -- run    # start it
 ```
 
 `init` asks for the API shape (`openai-responses` or `anthropic-messages`),
 the base URL, the model, and the token. It reads `SAUSAGE_API`,
 `SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The
-values go to `world/actor/agent/data/config.json`. Both commands take another
-folder as an argument, for example `nix run -- run my-world`.
+values go to `world/actor/agent/data/config.json`. `init` and `run` take
+another folder as an argument, for example `-- run my-world`.
 
 To keep a world on the newest version, run it in a loop:
 
 ```sh
-nix run .#loop -- my-world            # or .#loop-internal (Gitea on the LAN)
+nix run git+https://gitea.surma.technology/surma/sausage#loop             # ./world
+nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world # an existing world
 ```
 
-The loop runs `nix run --refresh git+ssh://…/surma/sausage -- run my-world`.
+The loop runs the newest `sausage` from that URL (`nix run --refresh`).
 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.
+fetches the newest commit and starts it again. `/quit` ends the loop. Set
+`SAUSAGE_FLAKE` to run another flake URL instead, for example your fork.
+
+To work on Sausage itself, clone it
+(`git clone https://gitea.surma.technology/surma/sausage`) and use `nix run`,
+`nix run .#loop` and `nix develop` in the checkout.
 
 In the console:
 

cli/src/broker.rs

@@ -1,6 +1,7 @@
-//! The broker: message routing, mailboxes, subscriptions, the version store,
-//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
-//! remove, subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
+//! The broker: message routing, mailboxes, subscriptions, taps, the version
+//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
+//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
+//! 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
@@ -133,17 +134,23 @@ impl Broker {
             Some(own) => format!("{from}/{own}"),
             None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
         };
-        let recipients: Vec<String> = match to {
-            Some(to) => vec![to.to_string()],
-            None => self
-                .metas
-                .lock()
-                .unwrap()
-                .iter()
-                .filter(|(_, m)| m.subscriptions.contains(from))
-                .map(|(id, _)| id.clone())
-                .collect(),
+        let (recipients, tappers) = {
+            let metas = self.metas.lock().unwrap();
+            let recipients: Vec<String> = match to {
+                Some(to) => vec![to.to_string()],
+                None => metas
+                    .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(),
@@ -156,6 +163,25 @@ 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();
@@ -497,6 +523,12 @@ 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));
@@ -504,7 +536,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, scan, exit"),
+            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
         }
     }
 

cli/src/main.rs

@@ -12,11 +12,13 @@ use sausage::{Host, cargo_actor, console};
 /// The source of the agent actor. `init` builds it like any other actor.
 static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
 
+// Without a command, `sausage` runs the world in `./world` and creates it
+// first (like `init`) if it does not exist yet.
 #[derive(Parser)]
 #[command(version, about)]
 struct Cli {
     #[command(subcommand)]
-    command: Cmd,
+    command: Option<Cmd>,
 }
 
 #[derive(Subcommand)]
@@ -112,7 +114,15 @@ async fn run(dir: PathBuf) -> Result<()> {
 #[tokio::main]
 async fn main() -> Result<()> {
     match Cli::parse().command {
-        Cmd::Init { dir } => init(dir).await,
-        Cmd::Run { dir } => run(dir).await,
+        Some(Cmd::Init { dir }) => init(dir).await,
+        Some(Cmd::Run { dir }) => run(dir).await,
+        None => {
+            let dir = PathBuf::from("world");
+            if !dir.join("actor").is_dir() {
+                println!("No world in ./world yet, creating it (same as `sausage init`).");
+                init(dir.clone()).await?;
+            }
+            run(dir).await
+        }
     }
 }

cli/src/world.rs

@@ -5,7 +5,7 @@
 //! <world>/
 //!   actor/<id>/
 //!     actor.wasm        the current version
-//!     actor.json        version, running, subscriptions (broker)
+//!     actor.json        version, running, subscriptions, tapped (broker)
 //!     source/           source of the current version
 //!     public/           written by the actor, read-only for others
 //!     data/             state, writable only by the actor
@@ -43,6 +43,9 @@ 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/cli.rs

@@ -0,0 +1,41 @@
+//! Tests of the `sausage` command line.
+
+use std::path::PathBuf;
+use std::process::{Command, Stdio};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+fn empty_dir(label: &str) -> PathBuf {
+    let nonce = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
+    let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap()
+        .join("target/test-cli")
+        .join(format!("{label}-{}-{nonce}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+/// Bare `sausage` (as in `nix run <flake>`) without a world starts `init`
+/// for ./world instead of printing a usage error.
+#[test]
+fn bare_command_without_a_world_starts_init() {
+    let dir = empty_dir("bare");
+    let output = Command::new(env!("CARGO_BIN_EXE_sausage"))
+        .current_dir(&dir)
+        .env_remove("SAUSAGE_API")
+        .env_remove("SAUSAGE_BASE_URL")
+        .env_remove("SAUSAGE_MODEL")
+        .env_remove("SAUSAGE_API_TOKEN")
+        .stdin(Stdio::null())
+        .output()
+        .unwrap();
+    let stdout = String::from_utf8_lossy(&output.stdout);
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    // With no answers on stdin, init stops at the first required value.
+    assert!(!output.status.success(), "stdout:\n{stdout}\nstderr:\n{stderr}");
+    assert!(stdout.contains("No world in ./world yet"), "stdout:\n{stdout}\nstderr:\n{stderr}");
+    assert!(stderr.contains("SAUSAGE_BASE_URL"), "stdout:\n{stdout}\nstderr:\n{stderr}");
+    assert!(!stderr.contains("Usage:"), "stderr:\n{stderr}");
+    // It asks before it creates anything.
+    assert!(!dir.join("world").exists());
+}

cli/tests/host.rs

@@ -545,7 +545,8 @@ 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");
-    assert!(unknown.body["reason"].as_str().unwrap().contains("remove"));
+    let reason = unknown.body["reason"].as_str().unwrap();
+    assert!(reason.contains("remove") && reason.contains("tap"), "{reason}");
     host.stop().await;
 }
 
@@ -797,3 +798,114 @@ 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,18 +36,19 @@
           meta.mainProgram = "sausage";
         };
         # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
-        # from the Gitea repo, again and again: after a crash or a broker
+        # (without a folder: bare `sausage`, i.e. ./world, created if needed)
+        # from the public 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.
-        # `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN.
-        publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage";
-        internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage";
-        loop = name: url: pkgs.writeShellScriptBin name ''
-          url="${url}"
-          if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
-          if [ $# -eq 0 ]; then set -- world; fi
+        # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
+        # e.g. a fork or a local checkout).
+        repoUrl = "git+https://gitea.surma.technology/surma/sausage";
+        loop = pkgs.writeShellScriptBin "sausage-loop" ''
+          url="''${SAUSAGE_FLAKE:-${repoUrl}}"
+          # No folder: bare `sausage` (creates ./world first if needed).
+          if [ $# -gt 0 ]; then set -- run "$@"; fi
           while true; do
-            nix run --refresh "$url#default" -- run "$@"
+            nix run --refresh "$url#default" -- "$@"
             code=$?
             if [ "$code" -eq 0 ]; then exit 0; fi
             echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2
@@ -72,8 +73,7 @@
         });
 
         apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
-        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"; };
+        apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
 
         devShells.default = pkgs.mkShell {
           packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];