Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d6d3c9a2f2 | ||
|
|
0951ba18e1 | ||
|
|
d0876626df | ||
|
|
c1f182d9b6 |
@@ -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
|
name: ci
|
||||||
on: [push, pull_request]
|
on: [push, pull_request]
|
||||||
|
|
||||||
|
|||||||
+2
-1
@@ -109,7 +109,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
|
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`,
|
the request: the request's `type` in past tense (`spawned`, `upgraded`,
|
||||||
`rolled_back`, `stopped`, `started`, `subscribed`, `unsubscribed`,
|
`rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
|
||||||
`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
|
`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
|
||||||
|
|
||||||
| `type` | Fields | Effect |
|
| `type` | Fields | Effect |
|
||||||
@@ -118,6 +118,7 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
|
|||||||
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
|
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
|
||||||
| `rollback` | `actor`, `version` | go back to an older version |
|
| `rollback` | `actor`, `version` | go back to an older version |
|
||||||
| `stop`, `start` | `actor` | stop or start an actor |
|
| `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 |
|
| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
|
||||||
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
|
| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
|
||||||
| `scan` | | boot new or changed `actor.wasm` files |
|
| `scan` | | boot new or changed `actor.wasm` files |
|
||||||
|
|||||||
@@ -10,28 +10,43 @@ change its own code.
|
|||||||
|
|
||||||
## Start
|
## 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
|
```sh
|
||||||
nix run -- init # creates ./world and builds the agent (a few minutes)
|
nix run git+https://gitea.surma.technology/surma/sausage
|
||||||
nix run -- run # starts the world and a console
|
```
|
||||||
|
|
||||||
|
This creates a world in `./world` the first time (it builds the agent, which
|
||||||
|
takes a few minutes), then starts it with a console. Later runs start the
|
||||||
|
existing world. The two steps on their own:
|
||||||
|
|
||||||
|
```sh
|
||||||
|
nix run git+https://gitea.surma.technology/surma/sausage -- init # creates ./world and builds the agent
|
||||||
|
nix run git+https://gitea.surma.technology/surma/sausage -- run # starts the world and a console
|
||||||
```
|
```
|
||||||
|
|
||||||
`init` asks for the API shape (`openai-responses` or `anthropic-messages`),
|
`init` asks for the API shape (`openai-responses` or `anthropic-messages`),
|
||||||
the base URL, the model, and the token. It reads `SAUSAGE_API`,
|
the base URL, the model, and the token. It reads `SAUSAGE_API`,
|
||||||
`SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The
|
`SAUSAGE_BASE_URL`, `SAUSAGE_MODEL`, and `SAUSAGE_API_TOKEN` first. The
|
||||||
values go to `world/actor/agent/data/config.json`. Both commands take another
|
values go to `world/actor/agent/data/config.json`. Both commands take another
|
||||||
folder as an argument, for example `nix run -- run my-world`.
|
folder as an argument, for example `... -- run my-world`.
|
||||||
|
|
||||||
To keep a world on the newest version, run it in a loop:
|
To keep a world on the newest version, run it in a loop:
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
nix run .#loop -- my-world # or .#loop-internal (Gitea on the LAN)
|
nix run git+https://gitea.surma.technology/surma/sausage#loop # ./world, created if needed
|
||||||
|
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 `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
|
||||||
When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop
|
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.
|
||||||
|
|
||||||
|
From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
|
||||||
|
same commands are `nix run`, `nix run -- init`, `nix run -- run` and
|
||||||
|
`nix run .#loop`.
|
||||||
|
|
||||||
In the console:
|
In the console:
|
||||||
|
|
||||||
|
|||||||
+33
-3
@@ -1,7 +1,7 @@
|
|||||||
//! The broker: message routing, mailboxes, subscriptions, taps, the version
|
//! The broker: message routing, mailboxes, subscriptions, taps, the version
|
||||||
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
|
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
|
||||||
//! subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` runs each
|
//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
|
||||||
//! WASM actor.
|
//! runs each WASM actor.
|
||||||
//!
|
//!
|
||||||
//! The broker accepts only `.wasm` files. It does not know any source language.
|
//! The broker accepts only `.wasm` files. It does not know any source language.
|
||||||
//! The world folder is the source of truth: `scan` boots every
|
//! The world folder is the source of truth: `scan` boots every
|
||||||
@@ -377,6 +377,35 @@ impl Broker {
|
|||||||
Ok(result)
|
Ok(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Stop an actor (normal shutdown) and delete its folder `actor/<id>/`.
|
||||||
|
/// Afterwards the ID is unknown: messages to it are undeliverable, and
|
||||||
|
/// it can be spawned again from scratch.
|
||||||
|
pub async fn remove_actor(&self, id: &str) -> Result<Value> {
|
||||||
|
if NATIVE.contains(&id) {
|
||||||
|
bail!("{id} is part of the host and cannot be removed");
|
||||||
|
}
|
||||||
|
if !world::valid_id(id) {
|
||||||
|
bail!("invalid actor ID {id:?}");
|
||||||
|
}
|
||||||
|
let lock = self.op_lock(id);
|
||||||
|
let _guard = lock.lock().await;
|
||||||
|
let dir = self.world.actor_dir(id);
|
||||||
|
if !self.is_supervised(id) && !dir.is_dir() {
|
||||||
|
bail!("no actor with ID {id}");
|
||||||
|
}
|
||||||
|
if self.is_supervised(id) {
|
||||||
|
self.control(id, |reply| Control::Shutdown { reply }).await?;
|
||||||
|
}
|
||||||
|
// Without its mailbox, messages to `id` are undeliverable.
|
||||||
|
self.controls.lock().unwrap().remove(id);
|
||||||
|
self.mailboxes.lock().unwrap().remove(id);
|
||||||
|
self.metas.lock().unwrap().remove(id);
|
||||||
|
if dir.exists() {
|
||||||
|
std::fs::remove_dir_all(&dir).with_context(|| format!("delete {}", dir.display()))?;
|
||||||
|
}
|
||||||
|
Ok(json!({"type": "removed", "actor": id}))
|
||||||
|
}
|
||||||
|
|
||||||
/// Boot the actors in `actor/`. Called at start and on `scan`.
|
/// Boot the actors in `actor/`. Called at start and on `scan`.
|
||||||
///
|
///
|
||||||
/// - An `actor.wasm` equal to the installed version starts as is.
|
/// - An `actor.wasm` equal to the installed version starts as is.
|
||||||
@@ -481,6 +510,7 @@ impl Broker {
|
|||||||
}
|
}
|
||||||
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
|
"stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
|
||||||
"start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
|
"start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
|
||||||
|
"remove" => self.remove_actor(&actor()?).await,
|
||||||
"subscribe" | "unsubscribe" => {
|
"subscribe" | "unsubscribe" => {
|
||||||
let publisher = actor()?;
|
let publisher = actor()?;
|
||||||
let on = ty == "subscribe";
|
let on = ty == "subscribe";
|
||||||
@@ -506,7 +536,7 @@ impl Broker {
|
|||||||
self.exit.notify_one();
|
self.exit.notify_one();
|
||||||
Ok(json!({"type": "exiting"}))
|
Ok(json!({"type": "exiting"}))
|
||||||
}
|
}
|
||||||
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, tap, untap, scan, exit"),
|
other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+13
-3
@@ -12,11 +12,13 @@ use sausage::{Host, cargo_actor, console};
|
|||||||
/// The source of the agent actor. `init` builds it like any other actor.
|
/// The source of the agent actor. `init` builds it like any other actor.
|
||||||
static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
|
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)]
|
#[derive(Parser)]
|
||||||
#[command(version, about)]
|
#[command(version, about)]
|
||||||
struct Cli {
|
struct Cli {
|
||||||
#[command(subcommand)]
|
#[command(subcommand)]
|
||||||
command: Cmd,
|
command: Option<Cmd>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Subcommand)]
|
#[derive(Subcommand)]
|
||||||
@@ -112,7 +114,15 @@ async fn run(dir: PathBuf) -> Result<()> {
|
|||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main() -> Result<()> {
|
async fn main() -> Result<()> {
|
||||||
match Cli::parse().command {
|
match Cli::parse().command {
|
||||||
Cmd::Init { dir } => init(dir).await,
|
Some(Cmd::Init { dir }) => init(dir).await,
|
||||||
Cmd::Run { dir } => run(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
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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());
|
||||||
|
}
|
||||||
@@ -494,6 +494,62 @@ async fn explicit_rollback_keeps_current_data() {
|
|||||||
host.stop().await;
|
host.stop().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn remove_shuts_an_actor_down_and_deletes_its_folder() {
|
||||||
|
let world = create_world("remove");
|
||||||
|
let wasm = toolbox_wasm().clone();
|
||||||
|
let (host, mut mailbox) = start(&world).await;
|
||||||
|
let source = passing_source(&world, "source");
|
||||||
|
spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await;
|
||||||
|
let write = json!({"type": "write", "path": "/data/state", "text": "x"});
|
||||||
|
ask_type(&host, &mut mailbox, "doomed", write, "ok").await;
|
||||||
|
ask_type(&host, &mut mailbox, "doomed", json!({"type": "arm-shutdown"}), "armed").await;
|
||||||
|
|
||||||
|
// Normal shutdown: the actor gets "shutdown" and its last message is routed.
|
||||||
|
let id = host.broker.send("console", Some("broker"), None, None, json!({"type": "remove", "actor": "doomed"}));
|
||||||
|
let (mut removed, mut last_words) = (false, false);
|
||||||
|
tokio::time::timeout(Duration::from_secs(10), async {
|
||||||
|
while !removed || !last_words {
|
||||||
|
let message = mailbox.recv().await.expect("console mailbox closed");
|
||||||
|
if message.reply_to.as_deref() == Some(id.as_str()) {
|
||||||
|
assert_eq!(message.body, json!({"type": "removed", "actor": "doomed"}));
|
||||||
|
removed = true;
|
||||||
|
} else if message.reply_to.as_deref() == Some("broker/shutdown") {
|
||||||
|
assert_eq!(message.body["type"], "shutdown-reply");
|
||||||
|
last_words = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("remove did not answer or did not shut the actor down");
|
||||||
|
assert!(!world.actor_dir("doomed").exists());
|
||||||
|
|
||||||
|
let gone = ask(&host, &mut mailbox, "doomed", json!({"type": "echo", "value": "hi"})).await;
|
||||||
|
assert_eq!(gone.body["type"], "undeliverable");
|
||||||
|
let again = request(&host, &mut mailbox, json!({"type": "remove", "actor": "doomed"})).await;
|
||||||
|
assert_eq!(again.body["type"], "error");
|
||||||
|
let broker = request(&host, &mut mailbox, json!({"type": "remove", "actor": "broker"})).await;
|
||||||
|
assert_eq!(broker.body["type"], "error");
|
||||||
|
assert!(world.actor_dir("broker").exists());
|
||||||
|
|
||||||
|
// The ID is free again: a new spawn starts at version 1 with empty data.
|
||||||
|
assert_eq!(spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await.body["version"], 1);
|
||||||
|
let read = json!({"type": "read", "path": "/data/state"});
|
||||||
|
ask_type(&host, &mut mailbox, "doomed", read, "error").await;
|
||||||
|
host.stop().await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn unknown_broker_request_lists_the_known_ones() {
|
||||||
|
let world = create_world("unknown-request");
|
||||||
|
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!(reason.contains("remove") && reason.contains("tap"), "{reason}");
|
||||||
|
host.stop().await;
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
|
async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
|
||||||
let world = create_world("cargo-actor");
|
let world = create_world("cargo-actor");
|
||||||
|
|||||||
@@ -36,18 +36,19 @@
|
|||||||
meta.mainProgram = "sausage";
|
meta.mainProgram = "sausage";
|
||||||
};
|
};
|
||||||
# `nix run .#loop -- <world>` runs the newest `sausage run <world>`
|
# `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,
|
# `exit` request it refreshes the flake and starts the host again,
|
||||||
# so the world updates itself. `/quit` (exit code 0) ends the loop.
|
# so the world updates itself. `/quit` (exit code 0) ends the loop.
|
||||||
# `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN.
|
# SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
|
||||||
publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage";
|
# e.g. a fork or a local checkout).
|
||||||
internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage";
|
repoUrl = "git+https://gitea.surma.technology/surma/sausage";
|
||||||
loop = name: url: pkgs.writeShellScriptBin name ''
|
loop = pkgs.writeShellScriptBin "sausage-loop" ''
|
||||||
url="${url}"
|
url="''${SAUSAGE_FLAKE:-${repoUrl}}"
|
||||||
if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
|
# No folder: bare `sausage` (creates ./world first if needed).
|
||||||
if [ $# -eq 0 ]; then set -- world; fi
|
if [ $# -gt 0 ]; then set -- run "$@"; fi
|
||||||
while true; do
|
while true; do
|
||||||
nix run --refresh "$url#default" -- run "$@"
|
nix run --refresh "$url#default" -- "$@"
|
||||||
code=$?
|
code=$?
|
||||||
if [ "$code" -eq 0 ]; then exit 0; fi
|
if [ "$code" -eq 0 ]; then exit 0; fi
|
||||||
echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2
|
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.default = { type = "app"; program = "${sausage}/bin/sausage"; };
|
||||||
apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; };
|
apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
|
||||||
apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
|
|
||||||
|
|
||||||
devShells.default = pkgs.mkShell {
|
devShells.default = pkgs.mkShell {
|
||||||
packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];
|
packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];
|
||||||
|
|||||||
Reference in New Issue
Block a user