Compare commits

..
Author SHA1 Message Date
agent-pool-8 14f086423b Keep discovery test file at served root
ci / check (push) Failing after 6m54s
2026-10-16 20:59:09 +00:00
agent-pool-8 e32bfc57e2 Expose advertised actor-served files in directory listings 2026-10-16 20:59:09 +00:00
21 changed files with 156 additions and 1315 deletions
Generated
-97
View File
@@ -678,15 +678,6 @@ dependencies = [
"windows-sys 0.59.0", "windows-sys 0.59.0",
] ]
[[package]]
name = "fsevent-sys"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "futures" name = "futures"
version = "0.3.34" version = "0.3.34"
@@ -1075,26 +1066,6 @@ dependencies = [
"serde_core", "serde_core",
] ]
[[package]]
name = "inotify"
version = "0.11.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4cc00ea907cab49550b7da656f80ebb97be1b997d931fbcd28d39734e17ce592"
dependencies = [
"bitflags",
"inotify-sys",
"libc",
]
[[package]]
name = "inotify-sys"
version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "instant" name = "instant"
version = "0.1.13" version = "0.1.13"
@@ -1194,26 +1165,6 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
] ]
[[package]]
name = "kqueue"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea"
dependencies = [
"kqueue-sys",
"libc",
]
[[package]]
name = "kqueue-sys"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087"
dependencies = [
"bitflags",
"libc",
]
[[package]] [[package]]
name = "leb128" name = "leb128"
version = "0.2.7" version = "0.2.7"
@@ -1308,7 +1259,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
dependencies = [ dependencies = [
"libc", "libc",
"log",
"wasi", "wasi",
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]
@@ -1319,33 +1269,6 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d" checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d"
[[package]]
name = "notify"
version = "8.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3"
dependencies = [
"bitflags",
"fsevent-sys",
"inotify",
"kqueue",
"libc",
"log",
"mio",
"notify-types",
"walkdir",
"windows-sys 0.60.2",
]
[[package]]
name = "notify-types"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a"
dependencies = [
"bitflags",
]
[[package]] [[package]]
name = "object" name = "object"
version = "0.40.0" version = "0.40.0"
@@ -1657,15 +1580,6 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
[[package]]
name = "same-file"
version = "1.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
dependencies = [
"winapi-util",
]
[[package]] [[package]]
name = "sausage" name = "sausage"
version = "0.1.0" version = "0.1.0"
@@ -1676,7 +1590,6 @@ dependencies = [
"http", "http",
"http-body-util", "http-body-util",
"include_dir", "include_dir",
"notify",
"serde", "serde",
"serde_json", "serde_json",
"tokio", "tokio",
@@ -2162,16 +2075,6 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7"
[[package]]
name = "walkdir"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b"
dependencies = [
"same-file",
"winapi-util",
]
[[package]] [[package]]
name = "want" name = "want"
version = "0.3.2" version = "0.3.2"
+14 -58
View File
@@ -63,14 +63,12 @@ Each instance sees three folders:
| Path | Access | Contents | | Path | Access | Contents |
| --- | --- | --- | | --- | --- | --- |
| `/data` | read-write by its actor; readable by every actor | the actor's state | | `/data` | read-write | the actor's state |
| `/public` | read-write | the files that the actor publishes | | `/public` | read-write | the files that the actor publishes |
| `/actor` | read-only | the `actor/` folder of the world | | `/actor` | read-only | the `actor/` folder of the world |
An actor can write only its own `/data`, but can read every actor's `/data` through `/actor/<id>/data/`. `/data` is not private: any unprotected secret file there is readable by every actor.
`/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json` `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
(installed version, running, subscriptions, tapped, watches), `source/`, `public/`, `data/`, (installed version, running, subscriptions, tapped), `source/`, `public/`, `data/`,
`stderr.log`, and `versions/<n>/`. `ls /actor` lists all actors. `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 Files are plain POSIX files: a write is visible at once. To publish a file
@@ -112,15 +110,14 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
- No reply within 10 seconds, or an actor that does not exist, makes the - No reply within 10 seconds, or an actor that does not exist, makes the
operation fail (`io`, or `no-entry`). An actor that opens its own `fs/` operation fail (`io`, or `no-entry`). An actor that opens its own `fs/`
waits for itself and fails after the timeout. waits for itself and fails after the timeout.
- `fs` is reserved: no actor can have that ID. It also sends `fs_changed` - `fs` is reserved: no actor can have that ID.
to watchers (see Watches).
## The broker ## The broker
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`, `removed`, `subscribed`, `unsubscribed`, `rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
`tapping`, `untapped`, `watching`, `unwatched`, `exiting`), or `{"type": "error", "reason": "..."}`. `tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
| `type` | Fields | Effect | | `type` | Fields | Effect |
| --- | --- | --- | | --- | --- | --- |
@@ -131,10 +128,8 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
| `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` | | `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 |
| `watch`, `unwatch` | `path` | receive (or not) `fs_changed` for every change of `path` or anything under it; answer `{"type": "watching" \| "unwatched", "path"}` with the path as an `/actor/...` path | | `scan` | | boot new or changed `actor.wasm` files |
| `exit` | | answer `exiting`, stop every actor, and end the `sausage` process with exit code 75 (a wrapper like `nix run .#loop` starts it again, updated) | | `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
**Headless.** `sausage run --headless [dir]` (or `sausage --headless`, or `SAUSAGE_HEADLESS=1`) turns the console input off: the host does not read stdin, so end of input (`/dev/null`) does not stop the world. `nix run .#loop -- --headless <world>` passes it through the loop. It runs until SIGTERM or SIGINT (stops every actor, exit code 0) or a broker `exit` request (exit code 75, as above). Without the flag, EOF on stdin or `/quit` ends the host.
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
@@ -156,31 +151,6 @@ including its own), for debugging, logging, or a live view of the world:
- Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and - Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and
survives restarts and upgrades. `untap` ends it. survives restarts and upgrades. `untap` ends it.
**Watches.** The host watches the whole world folder with the OS file
watcher: every change inside it counts, whoever makes it (actors, a 9P mount,
a shell or `git` on the host). Actor-served files (`/actor/<id>/fs/`) are not
on disk and do not count. After `watch {path}`, the sender gets for each
change of `path` or a path under it (`/actor/a` covers `/actor/a/x`, not
`/actor/ab`):
```json
{"type": "fs_changed", "path": "/actor/notes/public/list.md", "op": "write"}
```
- `op` is `write`, `create`, `remove`, or `rename` (then `to` is the new
path; a watcher of either path gets it).
- Changes of one path within about 100 ms become one event (a `write` after
a `create` stays a `create`).
- The host's temporary files (names starting with `.sausage-tmp-`) are
skipped; renaming one onto a file is a `write` of that file.
- Events come from `fs`. Like tap copies, they are not routed, a full
mailbox drops them, and they never block the writer. Never answer them.
- If the OS drops events, every watcher gets one
`{"type": "fs_changed", "path": "/actor", "op": "rescan"}`: anything may
have changed.
- A watch is saved in `actor.json` (`"watches": [...]`) and survives
restarts and upgrades. `unwatch` with the same path ends it.
**Versions.** A new version (from `spawn`, `upgrade`, or a changed **Versions.** A new version (from `spawn`, `upgrade`, or a changed
`actor.wasm` on disk) is tested first, while the old version keeps running. `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>/` Only when its tests pass does the broker copy it to `versions/<n>/`
@@ -219,15 +189,11 @@ the delay up to 30 s. A new version that crashes 3 times within 30 seconds
after an upgrade goes back to the previous version. The actor that asked for after an upgrade goes back to the previous version. The actor that asked for
the upgrade gets `{"type": "rolled_back", "from", "to", "reason", "last_exit"}`. the upgrade gets `{"type": "rolled_back", "from", "to", "reason", "last_exit"}`.
**The folder is the truth.** At start, the broker boots every **The folder is the truth.** At start, and on `scan`, the broker boots every
`actor/<id>/actor.wasm`. Afterwards, the file watcher reports a new or `actor/<id>/actor.wasm`. If the file differs from the installed version, it
changed `actor/<id>/actor.wasm` (or a new folder `actor/<id>/`) once it is is a new version, tested with the tests in `actor/<id>/source/`. If its
quiet for about 100 ms: a new file becomes a new actor, a changed one an tests fail, the file stays where it is, the reason goes to `stderr.log`, and
upgrade, tested first with the tests in `actor/<id>/source/`. A file equal the next scan tries again.
to the installed version changes nothing. If its tests fail, the file stays
where it is, the reason goes to `stderr.log`, and the next change of the
file tries again. When the OS drops events, the broker checks every actor
folder again, as at start.
## The `cargo` actor ## The `cargo` actor
@@ -235,19 +201,9 @@ folder again, as at start.
{"type": "build", "source": "/actor/agent/public/drafts/notes"} {"type": "build", "source": "/actor/agent/public/drafts/notes"}
``` ```
The build request may include `"skip_tests": true` to skip Rust unit tests. The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "built_by": "cargo", ...}`
By default, `cargo` first runs `cargo test` for the host target (not WASI), or `{"type": "build_failed", "errors": "..."}`. Pass `wasm`, `source`, and
so pure-logic unit tests in a `wasm32-wasip2` actor crate can run without a `built_by` on to `spawn` or `upgrade`.
WASI test runner. If tests fail or cannot compile, the build fails before the
WASI build. The reply carries `tests: {"passed": <count>, "failed": <count>,
"output": "<cargo test stdout and stderr>"}`; counts sum Cargo's test-result
summaries, and a test command/compilation failure without a summary counts as
one failure. With `skip_tests: true`, counts are zero and output says tests
were skipped.
The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "built_by": "cargo", "tests": {...}, ...}`
or `{"type": "build_failed", "errors": "...", "tests": {...}}`. Pass
`wasm`, `source`, and `built_by` on to `spawn` or `upgrade`.
A Rust actor depends on the SDK with `sausage-actor = "0.1"`. `cargo` points A Rust actor depends on the SDK with `sausage-actor = "0.1"`. `cargo` points
that dependency at the SDK of the host, which it publishes at that dependency at the SDK of the host, which it publishes at
+9 -34
View File
@@ -11,32 +11,24 @@ change its own code.
## Start ## Start
You need [Nix](https://nixos.org/download) with flakes, and an API token for 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: an LLM (OpenAI Responses or Anthropic Messages API).
```sh ```sh
nix run git+https://gitea.surma.technology/surma/sausage 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
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 `... -- run my-world`. folder as an argument, for example `nix run -- 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 git+https://gitea.surma.technology/surma/sausage#loop # ./world, created if needed nix run .#loop -- my-world
nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world # an existing world
``` ```
The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`. The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
@@ -44,22 +36,6 @@ 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. Set
`SAUSAGE_FLAKE` to run another flake URL instead, for example your fork. `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`.
### Headless
`--headless` (or `SAUSAGE_HEADLESS=1`) turns the console input off: the host
never reads stdin, so end of input does not stop the world. It runs until
SIGTERM/SIGINT (clean stop, exit code 0) or a broker `exit` request (exit
code 75, so the loop starts the newest version). Create the world first
(`init`), then pass the flag through the loop:
```sh
nix run git+https://gitea.surma.technology/surma/sausage#loop -- --headless my-world
```
In the console: In the console:
```text ```text
@@ -78,15 +54,14 @@ actor.wasm the current version
actor.json installed version, running, subscriptions actor.json installed version, running, subscriptions
source/ the source of the current version source/ the source of the current version
public/ files that the actor publishes public/ files that the actor publishes
data/ actor state; owner-writable, world-readable via /actor/<id>/data/ (not private) data/ the actor's state
stderr.log what the actor and the broker write about it stderr.log what the actor and the broker write about it
versions/<n>/ every installed version: actor.wasm, source/, meta.json versions/<n>/ every installed version: actor.wasm, source/, meta.json
``` ```
To add or replace an actor by hand, put an `actor.wasm` (and its `source/`, To add or replace an actor by hand, put an `actor.wasm` (and its `source/`,
with the tests in `source/tests/`) into its folder; the host watches the with the tests in `source/tests/`) into its folder. Then restart the world or
world folder and picks the file up once it stops changing. The broker tests send `@broker {"type":"scan"}`. The broker tests the file before it uses it.
the file before it uses it.
If the tests fail, the old version keeps running, and `stderr.log` says why. If the tests fail, the old version keeps running, and `stderr.log` says why.
## More ## More
+10 -84
View File
@@ -14,49 +14,11 @@ pub struct Config {
pub base_url: String, pub base_url: String,
pub model: String, pub model: String,
pub token: String, pub token: String,
/// Optional reasoning effort for OpenAI models only (wins over `thinking` there). /// Optional reasoning effort for OpenAI models ("low", "medium", ...).
#[serde(default)] #[serde(default)]
pub reasoning_effort: Option<String>, pub reasoning_effort: Option<String>,
#[serde(default)] #[serde(default)]
pub max_tokens: Option<u32>, pub max_tokens: Option<u32>,
/// Thinking effort for both APIs: "low", "medium", "high", "xhigh", "max" (`LEVELS`), or "off"/unset for none.
/// Anthropic: adaptive thinking + output_config.effort; OpenAI: reasoning.effort. A level the model rejects is
/// replaced by the nearest accepted one (logged), never by "off".
#[serde(default)]
pub thinking: Option<String>,
}
/// Thinking levels, lowest first.
pub const LEVELS: [&str; 5] = ["low", "medium", "high", "xhigh", "max"];
/// The level nearest to `want` (lower on a tie) among those a 400 error lists in quotes (else all levels), not
/// yet `tried`. None: nothing left, so the request goes out without an effort field (model default).
fn fallback_effort(want: &str, error: &str, tried: &[String]) -> Option<String> {
let listed: Vec<&str> = error.split('\'').skip(1).step_by(2).filter(|w| LEVELS.contains(w)).collect();
let pool = if listed.iter().any(|l| *l != want) { listed } else { LEVELS.to_vec() };
let at = LEVELS.iter().position(|l| *l == want).unwrap_or(1) as i32;
pool.into_iter()
.filter(|l| *l != want && !tried.iter().any(|t| t == l))
.min_by_key(|l| {
let i = LEVELS.iter().position(|x| x == l).unwrap_or(1) as i32;
((i - at).abs(), i > at)
})
.map(str::to_string)
}
#[cfg(test)]
mod tests {
use super::fallback_effort;
#[test]
fn nearest_accepted_level() {
let anthropic = "HTTP 400: output_config.effort: Input should be 'low', 'medium', 'high' or 'max'";
assert_eq!(fallback_effort("xhigh", anthropic, &["xhigh".into()]).as_deref(), Some("high"));
let openai = "HTTP 400: Invalid value: 'ultra'. Supported values are: 'none', 'minimal', 'low', 'medium', 'high', 'xhigh', and 'max'.";
assert_eq!(fallback_effort("ultra", openai, &["ultra".into()]).as_deref(), Some("medium"));
// No list in the error: step to the neighbour, never to off.
assert_eq!(fallback_effort("max", "HTTP 400: effort max not supported", &["max".into()]).as_deref(), Some("xhigh"));
}
} }
#[derive(Serialize, Deserialize, Clone, Debug)] #[derive(Serialize, Deserialize, Clone, Debug)]
@@ -70,7 +32,7 @@ pub enum Entry {
}, },
ToolCall { id: String, name: String, arguments: String }, ToolCall { id: String, name: String, arguments: String },
ToolResult { id: String, output: String }, ToolResult { id: String, output: String },
/// A provider item that only this API shape understands (OpenAI reasoning, Anthropic thinking). /// A provider item that only this API shape understands (OpenAI reasoning).
Opaque { api: String, item: Value }, Opaque { api: String, item: Value },
} }
@@ -81,28 +43,10 @@ pub struct Tool {
} }
pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> { pub async fn complete(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
let want = match cfg.api.as_str() { match cfg.api.as_str() {
"openai-responses" => cfg.reasoning_effort.clone().or_else(|| cfg.thinking.clone()), "openai-responses" => openai(cfg, system, tools, history).await,
_ => cfg.thinking.clone(), "anthropic-messages" => anthropic(cfg, system, tools, history).await,
} other => Err(format!("unknown API shape {other:?}")),
.filter(|t| t != "off");
let mut effort = want.clone();
let mut tried = Vec::new();
loop {
let result = match cfg.api.as_str() {
"openai-responses" => openai(cfg, system, tools, history, effort.as_deref()).await,
"anthropic-messages" => anthropic(cfg, system, tools, history, want.is_some(), effort.as_deref()).await,
other => return Err(format!("unknown API shape {other:?}")),
};
match (result, effort.clone()) {
// The model rejected the effort level: retry with the nearest accepted one (thinking stays on).
(Err(e), Some(current)) if e.starts_with("HTTP 400") && e.contains("effort") => {
tried.push(current.clone());
effort = fallback_effort(want.as_deref().unwrap_or(""), &e, &tried);
eprintln!("agent: {} rejected thinking effort {current:?}; retrying with {effort:?}: {e}", cfg.model);
}
(result, _) => return result,
}
} }
} }
@@ -123,7 +67,7 @@ async fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result<Val
serde_json::from_str(&text).map_err(|e| format!("invalid JSON in response: {e}")) serde_json::from_str(&text).map_err(|e| format!("invalid JSON in response: {e}"))
} }
async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], effort: Option<&str>) -> Result<Vec<Entry>, String> { async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
let mut input = Vec::new(); let mut input = Vec::new();
for e in history { for e in history {
match e { match e {
@@ -160,7 +104,7 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], e
if let Some(max_tokens) = cfg.max_tokens { if let Some(max_tokens) = cfg.max_tokens {
body["max_output_tokens"] = json!(max_tokens); body["max_output_tokens"] = json!(max_tokens);
} }
if let Some(effort) = effort { if let Some(effort) = &cfg.reasoning_effort {
body["reasoning"] = json!({"effort": effort}); body["reasoning"] = json!({"effort": effort});
} }
let url = format!("{}/v1/responses", cfg.base_url); let url = format!("{}/v1/responses", cfg.base_url);
@@ -199,14 +143,7 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], e
Ok(out) Ok(out)
} }
async fn anthropic( async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> {
cfg: &Config,
system: &str,
tools: &[Tool],
history: &[Entry],
thinking: bool,
effort: Option<&str>,
) -> Result<Vec<Entry>, String> {
// Merge consecutive blocks of the same role into one message. // Merge consecutive blocks of the same role into one message.
let mut messages: Vec<(String, Vec<Value>)> = Vec::new(); let mut messages: Vec<(String, Vec<Value>)> = Vec::new();
let mut push = |role: &str, block: Value| match messages.last_mut() { let mut push = |role: &str, block: Value| match messages.last_mut() {
@@ -226,8 +163,6 @@ async fn anthropic(
Entry::ToolResult { id, output } => { Entry::ToolResult { id, output } => {
push("user", json!({"type": "tool_result", "tool_use_id": id, "content": output})) push("user", json!({"type": "tool_result", "tool_use_id": id, "content": output}))
} }
// Thinking blocks go back unchanged (with their signature): required next to tool use.
Entry::Opaque { api, item } if api == "anthropic-messages" && thinking => push("assistant", item.clone()),
Entry::Opaque { .. } => {} Entry::Opaque { .. } => {}
} }
} }
@@ -236,19 +171,13 @@ async fn anthropic(
.iter() .iter()
.map(|t| json!({"name": t.name, "description": t.description, "input_schema": t.parameters})) .map(|t| json!({"name": t.name, "description": t.description, "input_schema": t.parameters}))
.collect(); .collect();
let mut body = json!({ let body = json!({
"model": cfg.model, "model": cfg.model,
"system": system, "system": system,
"messages": messages, "messages": messages,
"tools": tools, "tools": tools,
"max_tokens": cfg.max_tokens.unwrap_or(32000), "max_tokens": cfg.max_tokens.unwrap_or(32000),
}); });
if thinking {
body["thinking"] = json!({"type": "adaptive"});
if let Some(effort) = effort {
body["output_config"] = json!({"effort": effort});
}
}
let url = format!("{}/v1/messages", cfg.base_url); let url = format!("{}/v1/messages", cfg.base_url);
let resp = post( let resp = post(
&url, &url,
@@ -270,9 +199,6 @@ async fn anthropic(
name: block["name"].as_str().unwrap_or_default().to_string(), name: block["name"].as_str().unwrap_or_default().to_string(),
arguments: block["input"].to_string(), arguments: block["input"].to_string(),
}), }),
Some("thinking") | Some("redacted_thinking") => {
out.push(Entry::Opaque { api: "anthropic-messages".into(), item: block })
}
_ => {} _ => {}
} }
} }
-2
View File
@@ -3,5 +3,3 @@ People talk to you through other actors. Your text goes to your subscribers.
Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests. Before writing an actor, read `/actor/broker/public/PROTOCOL.md` for messages, the broker, `cargo`, and tests.
Read the SDK at `/actor/cargo/public/sdk/` before you write an actor. Read the SDK at `/actor/cargo/public/sdk/` before you write an actor.
You can change your own source at `/actor/agent/source`. You can change your own source at `/actor/agent/source`.
The global filesystem is readable by every actor, including every actor's `/data` at `/actor/<id>/data/`. An actor can write only its own `/data` and `/public`. `/data` is not private: any unprotected secret file there is readable by every actor.
-1
View File
@@ -15,7 +15,6 @@ clap = { version = "4", features = ["derive"] }
http = "1" http = "1"
http-body-util = "0.1" http-body-util = "0.1"
include_dir = "0.7" include_dir = "0.7"
notify = "8"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" serde_json = "1"
tokio = { version = "1", features = ["full"] } tokio = { version = "1", features = ["full"] }
+23 -111
View File
@@ -1,12 +1,11 @@
//! The broker: message routing, mailboxes, subscriptions, taps, watches, the //! The broker: message routing, mailboxes, subscriptions, taps, the version
//! version store, and the requests to `broker` (spawn, upgrade, rollback, //! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
//! stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch, //! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
//! 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 broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: at start, the broker boots every //! The world folder is the source of truth: `scan` boots every
//! `actor/<id>/actor.wasm`; afterwards the file watcher (`watch.rs`) reports //! `actor/<id>/actor.wasm` and treats a changed file as an upgrade.
//! changes: a new `actor.wasm` becomes an actor, a changed one an upgrade.
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap};
use std::path::Path; use std::path::Path;
@@ -168,19 +167,14 @@ impl Broker {
/// they are never tapped themselves; a full mailbox drops the copy. /// they are never tapped themselves; a full mailbox drops the copy.
fn tap(&self, tappers: &[String], message: Value) { 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 ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64);
self.inject(tappers, "broker", json!({"type": "tapped", "message": message, "ts": ts})); let body = json!({"type": "tapped", "message": message, "ts": ts});
}
/// Put a message of the host straight into the mailboxes of `to`: not
/// routed (so not tapped), never blocking, dropped if a mailbox is full.
fn inject(&self, to: &[String], from: &str, body: Value) {
let mailboxes = self.mailboxes.lock().unwrap(); let mailboxes = self.mailboxes.lock().unwrap();
for id in to { for tapper in tappers {
if let Some(tx) = mailboxes.get(id) { if let Some(tx) = mailboxes.get(tapper) {
let _ = tx.try_send(Envelope { let _ = tx.try_send(Envelope {
id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)), id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
from: from.into(), from: "broker".into(),
to: id.clone(), to: tapper.clone(),
reply_to: None, reply_to: None,
body: body.clone(), body: body.clone(),
}); });
@@ -412,90 +406,31 @@ impl Broker {
Ok(json!({"type": "removed", "actor": id})) Ok(json!({"type": "removed", "actor": id}))
} }
/// Boot the actors in `actor/`: at start, and as a full pass when the /// Boot the actors in `actor/`. Called at start and on `scan`.
/// file watcher dropped events (`overflowed`).
/// ///
/// - An `actor.wasm` equal to the installed version starts as is. /// - An `actor.wasm` equal to the installed version starts as is.
/// - A new or changed `actor.wasm` is tested where it lies. A new actor /// - A new or changed `actor.wasm` is tested where it lies. A new actor
/// is installed, a changed one is an upgrade. If its tests fail, /// is installed, a changed one is an upgrade. If its tests fail,
/// nothing changes, and the file stays until it changes again. /// nothing changes, and the file stays for the next scan.
pub async fn scan(self: &Arc<Self>) -> Vec<Value> { pub async fn scan(self: &Arc<Self>) -> Vec<Value> {
let mut report = Vec::new(); let mut report = Vec::new();
for id in self.world.actor_ids().unwrap_or_default() { for id in self.world.actor_ids().unwrap_or_default() {
if let Some(r) = self.boot_actor(&id).await { if NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() {
continue;
}
let lock = self.op_lock(&id);
let _guard = lock.lock().await;
if let Some(r) = self.scan_actor(&id).await {
report.push(r); report.push(r);
} }
} }
report report
} }
/// The file watcher saw `actor/<id>/actor.wasm` (or the folder) appear or
/// change: start, install or upgrade `id` as at boot, and log the result.
pub async fn check_actor(self: Arc<Self>, id: String) {
if let Some(result) = self.boot_actor(&id).await {
self.log(&id, &format!("actor.wasm changed: {result}"));
eprintln!("sausage: {result}");
}
}
/// The OS file watcher dropped events: tell every watcher with one
/// `fs_changed {path: "/actor", op: "rescan"}`, then run a full pass.
pub async fn overflowed(self: Arc<Self>) -> Vec<Value> {
self.host_error("the file watcher dropped events; watchers get a rescan, actors a full pass");
let watchers: Vec<String> =
self.metas.lock().unwrap().iter().filter(|(_, m)| !m.watches.is_empty()).map(|(id, _)| id.clone()).collect();
self.inject(&watchers, vfs::ID, json!({"type": "fs_changed", "path": "/actor", "op": "rescan"}));
let report = self.scan().await;
for change in &report {
eprintln!("sausage: {change}");
}
report
}
/// Send `fs_changed` for one coalesced change (from `watch.rs`) to every
/// actor that watches `path` or `to`. Like taps: straight into the
/// mailbox, dropped if it is full.
pub fn notify_watchers(&self, path: &str, op: &str, to: Option<&str>) {
let watchers: Vec<String> = {
let metas = self.metas.lock().unwrap();
metas
.iter()
.filter(|(_, m)| m.watches.iter().any(|w| under(path, w) || to.is_some_and(|t| under(t, w))))
.map(|(id, _)| id.clone())
.collect()
};
if watchers.is_empty() {
return;
}
let mut body = json!({"type": "fs_changed", "path": path, "op": op});
if let Some(to) = to {
body["to"] = json!(to);
}
self.inject(&watchers, vfs::ID, body);
}
/// A problem of the host itself: the broker's `stderr.log` and the terminal.
pub fn host_error(&self, text: &str) {
self.log("broker", text);
eprintln!("sausage: {text}");
}
async fn boot_actor(self: &Arc<Self>, id: &str) -> Option<Value> {
if NATIVE.contains(&id) || !world::valid_id(id) {
return None;
}
let lock = self.op_lock(id);
let _guard = lock.lock().await;
if !self.world.wasm_path(id).is_file() {
return None;
}
self.scan_actor(id).await
}
async fn scan_actor(self: &Arc<Self>, id: &str) -> Option<Value> { async fn scan_actor(self: &Arc<Self>, id: &str) -> Option<Value> {
// The broker's copy wins: actors change theirs (subscribe, ...) without the op lock. let meta = self.world.read_meta(id);
let meta = self.metas.lock().unwrap().entry(id.to_string()).or_insert_with(|| self.world.read_meta(id)).clone();
let installed = meta.version.filter(|v| self.world.version_dir(id, *v).join("actor.wasm").is_file()); let installed = meta.version.filter(|v| self.world.version_dir(id, *v).join("actor.wasm").is_file());
self.metas.lock().unwrap().insert(id.to_string(), meta.clone());
if let Some(v) = installed if let Some(v) = installed
&& !self.is_supervised(id) && !self.is_supervised(id)
{ {
@@ -594,32 +529,14 @@ impl Broker {
let answer = if on { "tapping" } else { "untapped" }; let answer = if on { "tapping" } else { "untapped" };
Ok(json!({"type": answer})) Ok(json!({"type": answer}))
} }
"watch" | "unwatch" => { "scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
let on = ty == "watch";
let path_s = b["path"].as_str().ok_or_else(|| anyhow!("missing \"path\""))?;
let global = self.world.global_path(&self.world.resolve(&env.from, path_s)?);
let path = global.ok_or_else(|| anyhow!("{path_s} is not under /actor"))?;
let path = match path.trim_end_matches('/') {
"" => "/actor".to_string(),
p => p.to_string(),
};
self.update_meta(&env.from, |m| {
if on {
m.watches.insert(path.clone());
} else {
m.watches.remove(&path);
}
})?;
let answer = if on { "watching" } else { "unwatched" };
Ok(json!({"type": answer, "path": path}))
}
"exit" => { "exit" => {
self.log("broker", &format!("exit requested by {}", env.from)); self.log("broker", &format!("exit requested by {}", env.from));
// notify_one keeps the permit if nobody waits yet. // notify_one keeps the permit if nobody waits yet.
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, remove, subscribe, unsubscribe, tap, untap, watch, unwatch, exit"), other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
} }
} }
@@ -646,11 +563,6 @@ fn same_file(a: &Path, b: &Path) -> bool {
matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y) matches!((std::fs::read(a), std::fs::read(b)), (Ok(x), Ok(y)) if x == y)
} }
/// Whether the namespace path `path` is `prefix` or inside it.
fn under(path: &str, prefix: &str) -> bool {
path.strip_prefix(prefix).is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
}
pub fn truncate(s: &str, max: usize) -> String { pub fn truncate(s: &str, max: usize) -> String {
if s.len() <= max { if s.len() <= max {
return s.to_string(); return s.to_string();
+47 -173
View File
@@ -1,9 +1,8 @@
//! The native `cargo` actor: builds Rust source folders into `.wasm` files. //! The native `cargo` actor: builds Rust source folders into `.wasm` files.
//! //!
//! Request: `{"type": "build", "source": "/actor/<id>/public/<folder>"}` //! Request: `{"type": "build", "source": "/actor/<id>/public/<folder>"}`
//! optional `"skip_tests": true` skips native `cargo test` (tests run by default). //! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", ...}`
//! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "tests": {"passed": 1, "failed": 0, "output": "..."}, ...}` //! `{"type": "build_failed", "errors": "..."}`
//! `{"type": "build_failed", "errors": "...", "tests": {"passed": 0, "failed": 1, "output": "..."}}`
//! //!
//! The broker knows nothing about Rust. This actor is one way to get a //! The broker knows nothing about Rust. This actor is one way to get a
//! `.wasm` file. Other actors can add other languages. //! `.wasm` file. Other actors can add other languages.
@@ -33,49 +32,11 @@ pub const TARGET: &str = "wasm32-wasip2";
/// The source of the `sausage-actor` SDK. /// The source of the `sausage-actor` SDK.
static SDK: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk"); static SDK: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../sdk");
#[derive(Clone, Default, Debug)]
pub struct TestResults {
pub passed: u64,
pub failed: u64,
pub output: String,
}
impl TestResults {
fn value(&self) -> Value {
json!({"passed": self.passed, "failed": self.failed, "output": self.output})
}
}
pub struct BuildOutput { pub struct BuildOutput {
/// Host path of the published `.wasm` file. /// Host path of the published `.wasm` file.
pub wasm: PathBuf, pub wasm: PathBuf,
pub seconds: f64, pub seconds: f64,
pub warnings: String, pub warnings: String,
pub tests: TestResults,
}
#[derive(Debug)]
pub struct BuildFailure {
pub reason: String,
pub tests: TestResults,
}
impl std::fmt::Display for BuildFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.tests.failed > 0 && !self.tests.output.is_empty() {
write!(f, "{}:\n{}", self.reason, self.tests.output)
} else {
f.write_str(&self.reason)
}
}
}
impl std::error::Error for BuildFailure {}
impl From<anyhow::Error> for BuildFailure {
fn from(error: anyhow::Error) -> Self {
Self { reason: format!("{error:#}"), tests: TestResults::default() }
}
} }
/// All builds share one target folder, so they run one at a time. /// All builds share one target folder, so they run one at a time.
@@ -90,84 +51,51 @@ impl Cargo {
Ok(Self { world }) Ok(Self { world })
} }
/// Run native `cargo test` unless opted out, then build and publish the /// Build the crate in `source` (a host path) and publish the `.wasm` file
/// crate's `wasm32-wasip2` binary under `/actor/cargo/public/builds/<n>/`. /// under `/actor/cargo/public/builds/<n>/`.
pub async fn build(&self, source: &Path, skip_tests: bool) -> std::result::Result<BuildOutput, BuildFailure> { pub async fn build(&self, source: &Path) -> Result<BuildOutput> {
if !source.join("Cargo.toml").is_file() { if !source.join("Cargo.toml").is_file() {
return Err(BuildFailure { bail!("{} has no Cargo.toml", source.display());
reason: format!("{} has no Cargo.toml", source.display()),
tests: TestResults::default(),
});
} }
let started = Instant::now(); let started = Instant::now();
let data = self.world.data_dir(ID); let data = self.world.data_dir(ID);
let build_dir = data.join("work"); let build_dir = data.join("work");
world::replace_source(source, &build_dir)?; world::replace_source(source, &build_dir)?;
let manifest = build_dir.join("Cargo.toml"); let manifest = build_dir.join("Cargo.toml");
let original = std::fs::read_to_string(&manifest).context("read build manifest")?; std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&self.world.public_dir(ID).join("sdk")))?)?;
let prepared = prepare_manifest(&original, Some(&self.world.public_dir(ID).join("sdk")))?;
std::fs::write(&manifest, prepared).context("write prepared manifest")?;
let target_dir = data.join("target"); let mut cmd = Command::new("cargo");
let tests = if skip_tests { cmd.arg("build")
TestResults { output: "skipped (skip_tests=true)".into(), ..TestResults::default() } .arg("--release")
} else { .arg("--target")
match run_tests(&build_dir, &target_dir).await { .arg(TARGET)
Ok(tests) => tests, .arg("--message-format=json-render-diagnostics")
Err(error) => { .arg("--target-dir")
let output = format!("cannot run cargo tests: {error:#}"); .arg(data.join("target"))
return Err(BuildFailure { .current_dir(&build_dir)
reason: output.clone(), .kill_on_drop(true);
tests: TestResults { failed: 1, output, ..TestResults::default() }, let out = cmd.output().await.context("cannot run cargo (is it on PATH, with the wasm32-wasip2 target?)")?;
}); let stderr = String::from_utf8_lossy(&out.stderr).to_string();
} if !out.status.success() {
} bail!("{}", strip_progress(&stderr));
};
if tests.failed > 0 {
return Err(BuildFailure { reason: "unit tests failed".into(), tests });
} }
let mut artifact = None;
let build_result: Result<BuildOutput> = async { for line in String::from_utf8_lossy(&out.stdout).lines() {
let mut cmd = Command::new("cargo"); if let Ok(v) = serde_json::from_str::<Value>(line)
cmd.arg("build") && v["reason"] == "compiler-artifact"
.arg("--release") && let Some(exe) = v["executable"].as_str()
.arg("--target") {
.arg(TARGET) artifact = Some(PathBuf::from(exe));
.arg("--message-format=json-render-diagnostics")
.arg("--target-dir")
.arg(&target_dir)
.current_dir(&build_dir)
.kill_on_drop(true);
let out = cmd.output().await.context("cannot run cargo (is it on PATH, with the wasm32-wasip2 target?)")?;
let stderr = String::from_utf8_lossy(&out.stderr).to_string();
if !out.status.success() {
bail!("{}", strip_progress(&stderr));
} }
let mut artifact = None;
for line in String::from_utf8_lossy(&out.stdout).lines() {
if let Ok(v) = serde_json::from_str::<Value>(line)
&& v["reason"] == "compiler-artifact"
&& let Some(exe) = v["executable"].as_str()
{
artifact = Some(PathBuf::from(exe));
}
}
let artifact = artifact.ok_or_else(|| anyhow!("cargo built no binary. Does the crate have a src/main.rs?"))?;
let n = next_build_number(&self.world.public_dir(ID).join("builds"))?;
let dest_dir = self.world.public_dir(ID).join("builds").join(n.to_string());
let name = artifact.file_name().context("artifact has no file name")?;
let dest = dest_dir.join(name);
world::write_atomic(&dest, &std::fs::read(&artifact)?)?;
Ok(BuildOutput {
wasm: dest,
seconds: started.elapsed().as_secs_f64(),
warnings: strip_progress(&stderr),
tests: tests.clone(),
})
} }
.await; let artifact = artifact.ok_or_else(|| anyhow!("cargo built no binary. Does the crate have a src/main.rs?"))?;
build_result.map_err(|error| BuildFailure { reason: format!("{error:#}"), tests })
let n = next_build_number(&self.world.public_dir(ID).join("builds"))?;
let dest_dir = self.world.public_dir(ID).join("builds").join(n.to_string());
let name = artifact.file_name().context("artifact has no file name")?;
let dest = dest_dir.join(name);
world::write_atomic(&dest, &std::fs::read(&artifact)?)?;
Ok(BuildOutput { wasm: dest, seconds: started.elapsed().as_secs_f64(), warnings: strip_progress(&stderr) })
} }
/// The native `cargo` actor. Builds run one at a time, in order. /// The native `cargo` actor. Builds run one at a time, in order.
@@ -178,7 +106,7 @@ impl Cargo {
continue; continue;
} }
let body = self.handle(&broker, &env).await.unwrap_or_else(|e| { let body = self.handle(&broker, &env).await.unwrap_or_else(|e| {
json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}"), "tests": TestResults::default().value()}) json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")})
}); });
broker.send(ID, Some(&env.from), None, Some(&env.id), body); broker.send(ID, Some(&env.from), None, Some(&env.id), body);
} }
@@ -189,76 +117,22 @@ impl Cargo {
Some("build") => { Some("build") => {
let source_s = env.body.get("source").and_then(Value::as_str).ok_or_else(|| anyhow!("missing \"source\""))?; let source_s = env.body.get("source").and_then(Value::as_str).ok_or_else(|| anyhow!("missing \"source\""))?;
let source = broker.world.resolve(&env.from, source_s)?; let source = broker.world.resolve(&env.from, source_s)?;
let skip_tests = env.body.get("skip_tests").and_then(Value::as_bool).unwrap_or(false); let out = self.build(&source).await?;
match self.build(&source, skip_tests).await { let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string());
Ok(out) => { Ok(json!({
let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string()); "type": "built",
Ok(json!({ "wasm": wasm,
"type": "built", "source": source_s,
"wasm": wasm, "built_by": ID,
"source": source_s, "seconds": (out.seconds * 10.0).round() / 10.0,
"built_by": ID, "warnings": out.warnings,
"seconds": (out.seconds * 10.0).round() / 10.0, }))
"warnings": out.warnings,
"tests": out.tests.value(),
}))
}
Err(failure) => Ok(json!({
"type": "build_failed",
"source": source_s,
"errors": failure.reason,
"tests": failure.tests.value(),
})),
}
} }
other => bail!("unknown request {other:?}. Known: build"), other => bail!("unknown request {other:?}. Known: build"),
} }
} }
} }
/// Run tests for the host triple: WASI crates' pure-logic tests can use the
/// normal Rust test harness without needing a WASI test runner.
async fn run_tests(build_dir: &Path, target_dir: &Path) -> Result<TestResults> {
let mut cmd = Command::new("cargo");
cmd.arg("test")
.arg("--target-dir")
.arg(target_dir)
.current_dir(build_dir)
.kill_on_drop(true);
let out = cmd.output().await.context("cannot run cargo test")?;
let stdout = String::from_utf8_lossy(&out.stdout).to_string();
let stderr = String::from_utf8_lossy(&out.stderr).to_string();
let combined = match (stdout.is_empty(), stderr.is_empty()) {
(true, true) => String::new(),
(false, true) => stdout,
(true, false) => stderr,
(false, false) => format!("{stdout}\n{stderr}"),
};
let output = strip_progress(&combined);
let passed = test_count(&output, "passed");
let mut failed = test_count(&output, "failed");
// Compilation failures and runner errors have no `test result` summary,
// but still make the default test phase fail.
if !out.status.success() && failed == 0 {
failed = 1;
}
Ok(TestResults { passed, failed, output })
}
/// Count the `passed` or `failed` totals in cargo's per-target test summaries.
fn test_count(output: &str, field: &str) -> u64 {
output
.lines()
.filter_map(|line| line.split_once("test result:").map(|(_, rest)| rest))
.flat_map(|summary| summary.split(';'))
.filter_map(|part| {
let words: Vec<_> = part.split_whitespace().collect();
let index = words.iter().position(|word| *word == field)?;
words.get(index.checked_sub(1)?)?.parse::<u64>().ok()
})
.sum()
}
/// Point a `sausage-actor` dependency at `sdk` (keeping its features), and /// Point a `sausage-actor` dependency at `sdk` (keeping its features), and
/// add an empty `[workspace]`, so the crate never joins a parent workspace. /// add an empty `[workspace]`, so the crate never joins a parent workspace.
fn prepare_manifest(text: &str, sdk: Option<&Path>) -> Result<String> { fn prepare_manifest(text: &str, sdk: Option<&Path>) -> Result<String> {
+1 -2
View File
@@ -3,8 +3,7 @@
//! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends //! - `@<actor> <json>` sends a JSON body. `@<actor> <text>` sends
//! `{"type": "text", "text": ...}`. //! `{"type": "text", "text": ...}`.
//! - A line without `@` goes to `agent` as text. //! - A line without `@` goes to `agent` as text.
//! - `/help`, `/quit`. EOF on stdin also quits (not with `--headless`, which //! - `/help`, `/quit`. EOF on stdin also quits.
//! never reads stdin).
use std::sync::Arc; use std::sync::Arc;
+7 -15
View File
@@ -1,8 +1,7 @@
//! Sausage: grow a system of WASI actors by talking to it. //! Sausage: grow a system of WASI actors by talking to it.
//! //!
//! The host runs the broker, the native actors (`broker`, `cargo`, and the //! The host runs the broker, the native actors (`broker`, `cargo`, and the
//! console), one supervisor per WASM actor, and the file watcher of the //! console), and one supervisor per WASM actor. See `PROTOCOL.md`.
//! world folder (`watch.rs`). See `PROTOCOL.md`.
pub mod broker; pub mod broker;
pub mod candidate; pub mod candidate;
@@ -11,7 +10,6 @@ pub mod console;
pub mod runtime; pub mod runtime;
pub mod supervisor; pub mod supervisor;
pub mod vfs; pub mod vfs;
pub mod watch;
pub mod world; pub mod world;
use std::path::Path; use std::path::Path;
@@ -32,13 +30,12 @@ pub struct Host {
pub world: World, pub world: World,
pub broker: Arc<Broker>, pub broker: Arc<Broker>,
native: Vec<tokio::task::JoinHandle<()>>, native: Vec<tokio::task::JoinHandle<()>>,
watcher: Option<watch::Watcher>,
_lock: std::fs::File, _lock: std::fs::File,
} }
impl Host { impl Host {
/// Start the broker, `cargo`, the file watcher, and every registered /// Start the broker, `cargo`, and every registered actor of the world at
/// actor of the world at `root`. Only one host can run a world at a time. /// `root`. Only one host can run a world at a time.
pub async fn start(root: &Path) -> Result<Host> { pub async fn start(root: &Path) -> Result<Host> {
let world = World::new(std::path::absolute(root)?); let world = World::new(std::path::absolute(root)?);
if !world.exists() { if !world.exists() {
@@ -56,14 +53,10 @@ impl Host {
tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))), tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))),
tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))), tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))),
]; ];
// Watch first, so that no change during the boot pass is lost.
let watcher = watch::start(broker.clone())
.inspect_err(|e| broker.host_error(&format!("no file watching: {e}")))
.ok();
for change in broker.scan().await { for change in broker.scan().await {
eprintln!("sausage: {change}"); eprintln!("sausage: {change}");
} }
Ok(Host { world, broker, native, watcher, _lock: lock }) Ok(Host { world, broker, native, _lock: lock })
} }
/// Register a native actor with the ID `id` and return its mailbox. /// Register a native actor with the ID `id` and return its mailbox.
@@ -71,10 +64,9 @@ impl Host {
self.broker.open_mailbox(id) self.broker.open_mailbox(id)
} }
/// Stop the file watcher, the broker and `cargo` (with the requests they /// Stop the broker and `cargo` (with the requests they still run), then
/// still run), then every actor, gracefully. /// every actor, gracefully.
pub async fn stop(mut self) { pub async fn stop(self) {
drop(self.watcher.take());
for task in &self.native { for task in &self.native {
task.abort(); task.abort();
} }
+9 -55
View File
@@ -5,7 +5,6 @@ use anyhow::{Context, Result, bail};
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use include_dir::{Dir, include_dir}; use include_dir::{Dir, include_dir};
use serde_json::json; use serde_json::json;
use tokio::signal::unix::Signal;
use sausage::world::{ActorMeta, World}; use sausage::world::{ActorMeta, World};
use sausage::{Host, cargo_actor, console}; use sausage::{Host, cargo_actor, console};
@@ -13,18 +12,11 @@ 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 {
/// Do not read stdin, so end of input does not stop the world. It runs
/// until SIGTERM (or SIGINT), or until an actor sends
/// `@broker {"type":"exit"}`. Also set by SAUSAGE_HEADLESS=1.
#[arg(long, global = true)]
headless: bool,
#[command(subcommand)] #[command(subcommand)]
command: Option<Cmd>, command: Cmd,
} }
#[derive(Subcommand)] #[derive(Subcommand)]
@@ -91,47 +83,19 @@ async fn init(dir: PathBuf) -> Result<()> {
std::fs::create_dir_all(&source)?; std::fs::create_dir_all(&source)?;
AGENT_SOURCE.extract(&source).context("extract agent source")?; AGENT_SOURCE.extract(&source).context("extract agent source")?;
println!("Building the agent actor (the first build takes a few minutes) …"); println!("Building the agent actor (the first build takes a few minutes) …");
let out = cargo_actor::Cargo::new(world.clone())?.build(&source, false).await.context("build agent")?; let out = cargo_actor::Cargo::new(world.clone())?.build(&source).await.context("build agent")?;
std::fs::copy(&out.wasm, world.wasm_path("agent"))?; std::fs::copy(&out.wasm, world.wasm_path("agent"))?;
println!("Built in {:.1} s. Start the world with: sausage run {}", out.seconds, dir.display()); println!("Built in {:.1} s. Start the world with: sausage run {}", out.seconds, dir.display());
Ok(()) Ok(())
} }
/// Install the SIGTERM and SIGINT handlers now (not when first polled). async fn run(dir: PathBuf) -> Result<()> {
fn termination_signals() -> Result<(Signal, Signal)> {
use tokio::signal::unix::{SignalKind, signal};
let term = signal(SignalKind::terminate()).context("install SIGTERM handler")?;
let int = signal(SignalKind::interrupt()).context("install SIGINT handler")?;
Ok((term, int))
}
/// Resolves on SIGTERM or SIGINT.
async fn termination((mut term, mut int): (Signal, Signal)) {
tokio::select! {
_ = term.recv() => {}
_ = int.recv() => {}
}
}
async fn run(dir: PathBuf, headless: bool) -> Result<()> {
// Register the signal handlers before the host starts, so a signal that
// arrives early is never the default (kill) action.
let signals = if headless { Some(termination_signals()?) } else { None };
let host = Host::start(&dir).await?; let host = Host::start(&dir).await?;
tokio::spawn(console::run_mailbox(host.mailbox(console::ID))); tokio::spawn(console::run_mailbox(host.mailbox(console::ID)));
let broker = host.broker.clone(); let broker = host.broker.clone();
let exit = match signals { let exit = tokio::select! {
Some(signals) => { _ = console::run_input(host.broker.clone()) => false,
println!("Sausage headless: running until SIGTERM or an exit request."); _ = broker.exit_requested() => true,
tokio::select! {
_ = termination(signals) => false,
_ = broker.exit_requested() => true,
}
}
None => tokio::select! {
_ = console::run_input(host.broker.clone()) => false,
_ = broker.exit_requested() => true,
},
}; };
println!("Stopping actors …"); println!("Stopping actors …");
host.stop().await; host.stop().await;
@@ -147,18 +111,8 @@ async fn run(dir: PathBuf, headless: bool) -> Result<()> {
#[tokio::main] #[tokio::main]
async fn main() -> Result<()> { async fn main() -> Result<()> {
let cli = Cli::parse(); match Cli::parse().command {
let headless = cli.headless || std::env::var("SAUSAGE_HEADLESS").is_ok_and(|v| !v.is_empty() && v != "0"); Cmd::Init { dir } => init(dir).await,
match cli.command { Cmd::Run { dir } => run(dir).await,
Some(Cmd::Init { dir }) => init(dir).await,
Some(Cmd::Run { dir }) => run(dir, headless).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, headless).await
}
} }
} }
+6 -7
View File
@@ -328,15 +328,14 @@ impl VfsView<'_> {
async fn native_directory_entries(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Vec<DirectoryEntry>> { 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 = HostDescriptor::read_directory(&mut self.inner, fd).await?;
let stream_rep = stream.rep();
let mut entries = Vec::new(); let mut entries = Vec::new();
loop { loop {
match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, Resource::new_borrow(stream_rep)).await? { match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream.clone()).await? {
Some(entry) => entries.push(entry), Some(entry) => entries.push(entry),
None => break, None => break,
} }
} }
HostDirectoryEntryStream::drop(&mut self.inner, stream).map_err(|_| ErrorCode::Io)?; HostDirectoryEntryStream::drop(&mut self.inner, stream)?;
Ok(entries) Ok(entries)
} }
@@ -588,10 +587,10 @@ impl HostDescriptor for VfsView<'_> {
None => { None => {
let child_path = self.state.dirs.get(&fd.rep()).map(|base| join(base, &path)); 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?; let opened = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await?;
if let Some(path) = child_path { if let Some(path) = child_path
if HostDescriptor::get_type(&mut self.inner, Resource::new_borrow(opened.rep())).await? == DescriptorType::Directory { && HostDescriptor::get_type(&mut self.inner, opened.clone()).await? == DescriptorType::Directory
self.state.dirs.insert(opened.rep(), path); {
} self.state.dirs.insert(opened.rep(), path);
} }
Ok(opened) Ok(opened)
} }
-256
View File
@@ -1,256 +0,0 @@
//! File watching: the host watches the whole world folder with the OS file
//! watcher (`notify`, inotify on Linux), whoever writes: actors, a 9P mount,
//! a shell or `git` on the host. Nothing outside the folder counts.
//!
//! Events are coalesced per path: once a path is quiet for `QUIET` (or has
//! kept changing for `MAX_DELAY`), it becomes one change
//! `{path, op: write | create | remove | rename, to?}` with a namespace path
//! (`/actor/...`). Then the broker
//!
//! - sends `fs_changed` to the actors that watch the path (`watch` request),
//! - starts or upgrades an actor whose `actor/<id>/actor.wasm` (or folder
//! `actor/<id>`) changed, tested first, as at boot (`Broker::check_actor`).
//!
//! The host's temporary files (`.sausage-tmp-*`) are skipped: a rename of one
//! onto a file is a `write` of that file. If the OS drops events, the broker
//! runs a full pass and tells every watcher (`Broker::overflowed`).
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use notify::event::{EventKind, ModifyKind, RenameMode};
use notify::{RecursiveMode, Watcher as _};
use tokio::sync::mpsc;
use tokio::time::Instant;
use crate::broker::Broker;
use crate::world::TMP_PREFIX;
/// A path that is quiet this long becomes one change.
pub const QUIET: Duration = Duration::from_millis(100);
/// A path that keeps changing becomes a change at least this often.
pub const MAX_DELAY: Duration = Duration::from_secs(1);
/// The running watcher. Dropping it stops watching.
pub struct Watcher {
_os: notify::RecommendedWatcher,
task: tokio::task::JoinHandle<()>,
}
impl Drop for Watcher {
fn drop(&mut self) {
self.task.abort();
}
}
/// Watch the world folder of `broker` recursively. A failure to watch part
/// of it (for example too few inotify watches) is logged, and the rest is
/// watched.
pub fn start(broker: Arc<Broker>) -> notify::Result<Watcher> {
let (tx, rx) = mpsc::unbounded_channel();
let mut os = notify::recommended_watcher(move |event| {
let _ = tx.send(event);
})?;
if let Err(e) = os.watch(&broker.world.root, RecursiveMode::Recursive) {
broker.host_error(&format!("cannot watch all of {}: {e}", broker.world.root.display()));
}
let task = tokio::spawn(run(broker, rx));
Ok(Watcher { _os: os, task })
}
/// A change waiting for its path to be quiet.
struct Pending {
op: &'static str,
to: Option<String>,
first: Instant,
last: Instant,
}
impl Pending {
fn due(&self) -> Instant {
(self.last + QUIET).min(self.first + MAX_DELAY)
}
}
async fn run(broker: Arc<Broker>, mut events: mpsc::UnboundedReceiver<notify::Result<notify::Event>>) {
let root = broker.world.actors_root();
let mut pending: HashMap<String, Pending> = HashMap::new();
// Actor checks and full passes. Dropping the set (when the host stops) aborts them.
let mut tasks = tokio::task::JoinSet::new();
loop {
let due = pending.values().map(Pending::due).min();
tokio::select! {
event = events.recv() => match event {
Some(Ok(event)) => {
if event.need_rescan() {
let broker = broker.clone();
tasks.spawn(async move {
broker.overflowed().await;
});
}
let now = Instant::now();
for (path, op, to) in changes(&root, &event) {
add(&mut pending, path, op, to, now);
}
}
Some(Err(e)) => broker.host_error(&format!("file watcher: {e}")),
None => break,
},
_ = sleep_until(due) => {
let now = Instant::now();
let ready: Vec<String> = pending.iter().filter(|(_, p)| p.due() <= now).map(|(k, _)| k.clone()).collect();
for path in ready {
let p = pending.remove(&path).unwrap();
let op = if p.op == "moved-in" { "create" } else { p.op };
broker.notify_watchers(&path, op, p.to.as_deref());
if let Some(id) = new_wasm(&path, op, p.to.as_deref()) {
tasks.spawn(broker.clone().check_actor(id));
}
}
}
Some(_) = tasks.join_next() => {}
}
}
}
async fn sleep_until(t: Option<Instant>) {
match t {
Some(t) => tokio::time::sleep_until(t).await,
None => std::future::pending().await,
}
}
/// The namespace path (`/actor/...`) of a host path, unless it is outside
/// `actor/` or a temporary file of the host.
fn namespace(root: &Path, path: &Path) -> Option<String> {
let rel = path.strip_prefix(root).ok()?;
let mut out = String::from("/actor");
for part in rel.components() {
let name = part.as_os_str().to_str()?;
if name.starts_with(TMP_PREFIX) {
return None;
}
out.push('/');
out.push_str(name);
}
Some(out)
}
/// One change: namespace path, op, and the new path of a rename.
type Change = (String, &'static str, Option<String>);
/// The changes of one OS event. inotify reports a rename three times:
/// `From` (the old name), `To` (the new name) and `Both`. `To` becomes the
/// internal op `moved-in`, which `Both` replaces.
fn changes(root: &Path, event: &notify::Event) -> Vec<Change> {
let path = |i: usize| event.paths.get(i).and_then(|p| namespace(root, p));
let one = |op: &'static str| -> Vec<Change> { path(0).map(|p| (p, op, None)).into_iter().collect() };
match event.kind {
EventKind::Create(_) => one("create"),
EventKind::Remove(_) => one("remove"),
EventKind::Modify(ModifyKind::Metadata(_)) => vec![],
EventKind::Modify(ModifyKind::Name(RenameMode::From)) => one("remove"),
EventKind::Modify(ModifyKind::Name(RenameMode::To)) => one("moved-in"),
EventKind::Modify(ModifyKind::Name(RenameMode::Both)) => match (path(0), path(1)) {
(Some(from), Some(to)) => vec![(from, "rename", Some(to))],
// Renamed to a temporary name: gone.
(Some(from), None) => vec![(from, "remove", None)],
// A temporary file renamed onto `to`: an atomic write.
(None, Some(to)) => vec![(to, "write", None)],
(None, None) => vec![],
},
EventKind::Modify(_) => one("write"),
_ => vec![],
}
}
/// Coalesce a change into `pending`. The last op wins, except that a write
/// after a create stays a create. A rename replaces the `moved-in` of its
/// new name.
fn add(pending: &mut HashMap<String, Pending>, path: String, op: &'static str, to: Option<String>, now: Instant) {
if let Some(to) = &to
&& pending.get(to).is_some_and(|p| p.op == "moved-in")
{
pending.remove(to);
}
match pending.get_mut(&path) {
Some(p) => {
if !(p.op == "create" && op == "write") {
p.op = op;
p.to = to;
}
p.last = now;
}
None => {
pending.insert(path, Pending { op, to, first: now, last: now });
}
}
}
/// The actor to check when `actor/<id>/actor.wasm` or the folder `actor/<id>`
/// appeared or changed.
fn new_wasm(path: &str, op: &str, to: Option<&str>) -> Option<String> {
let target = match op {
"remove" => return None,
"rename" => to?,
_ => path,
};
let rest = target.strip_prefix("/actor/")?;
let id = rest.strip_suffix("/actor.wasm").unwrap_or(rest);
(!id.is_empty() && !id.contains('/')).then(|| id.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn wasm_and_folder_changes_name_the_actor() {
assert_eq!(new_wasm("/actor/a/actor.wasm", "write", None).as_deref(), Some("a"));
assert_eq!(new_wasm("/actor/a", "create", None).as_deref(), Some("a"));
assert_eq!(new_wasm("/actor/x/y", "rename", Some("/actor/b/actor.wasm")).as_deref(), Some("b"));
assert_eq!(new_wasm("/actor/a/actor.wasm", "remove", None), None);
assert_eq!(new_wasm("/actor/a/versions/1/actor.wasm", "write", None), None);
assert_eq!(new_wasm("/actor/a/public/x", "write", None), None);
assert_eq!(new_wasm("/actor", "write", None), None);
}
#[test]
fn a_rename_is_one_change_and_temp_files_are_skipped() {
let root = Path::new("/w/actor");
let now = Instant::now();
let mut pending = HashMap::new();
let ev = |kind, paths: &[&str]| {
let mut e = notify::Event::new(kind);
for p in paths {
e = e.add_path(std::path::PathBuf::from(*p));
}
e
};
let name = |mode| EventKind::Modify(ModifyKind::Name(mode));
for e in [
ev(name(RenameMode::From), &["/w/actor/a/public/x"]),
ev(name(RenameMode::To), &["/w/actor/a/public/y"]),
ev(name(RenameMode::Both), &["/w/actor/a/public/x", "/w/actor/a/public/y"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/actor/a/.sausage-tmp-host-1-actor.json"]),
ev(name(RenameMode::Both), &["/w/actor/a/.sausage-tmp-host-1-actor.json", "/w/actor/a/actor.json"]),
ev(EventKind::Create(notify::event::CreateKind::File), &["/w/.lock"]),
] {
for (path, op, to) in changes(root, &e) {
add(&mut pending, path, op, to, now);
}
}
let mut got: Vec<(String, &str, Option<String>)> =
pending.into_iter().map(|(k, p)| (k, p.op, p.to)).collect();
got.sort();
assert_eq!(
got,
vec![
("/actor/a/actor.json".to_string(), "write", None),
("/actor/a/public/x".to_string(), "rename", Some("/actor/a/public/y".to_string())),
]
);
}
}
+2 -6
View File
@@ -1,12 +1,11 @@
//! On-disk layout of a Sausage world. The folder is the source of truth: //! On-disk layout of a Sausage world. The folder is the source of truth:
//! `sausage run` boots every actor that it finds in `actor/`, and then //! `sausage run` boots every actor that it finds in `actor/`.
//! watches the folder (`watch.rs`).
//! //!
//! ```text //! ```text
//! <world>/ //! <world>/
//! actor/<id>/ //! actor/<id>/
//! actor.wasm the current version //! actor.wasm the current version
//! actor.json version, running, subscriptions, tapped, watches (broker) //! actor.json version, running, subscriptions, tapped (broker)
//! source/ source of the current version //! source/ source of the current version
//! public/ written by the actor, read-only for others //! public/ written by the actor, read-only for others
//! data/ state, writable only by the actor //! data/ state, writable only by the actor
@@ -47,9 +46,6 @@ pub struct ActorMeta {
/// Whether this actor gets a copy of every routed message (`tap`). /// Whether this actor gets a copy of every routed message (`tap`).
#[serde(default)] #[serde(default)]
pub tapped: bool, pub tapped: bool,
/// Namespace paths whose changes this actor gets as `fs_changed` (`watch`).
#[serde(default)]
pub watches: BTreeSet<String>,
} }
fn yes() -> bool { fn yes() -> bool {
-85
View File
@@ -1,85 +0,0 @@
//! 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());
}
/// Start a headless host on a fresh world with stdin at EOF; returns once it
/// printed its ready line (printed after the signal handlers are installed).
fn start_headless(label: &str) -> (std::process::Child, std::io::Lines<std::io::BufReader<std::process::ChildStdout>>) {
use std::io::{BufRead, BufReader};
let dir = empty_dir(label);
let world = dir.join("world");
sausage::World::create(&world).unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
.args(["run", "--headless"])
.arg(&world)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
let line = lines.next().expect("host exited before it started").unwrap();
assert!(line.contains("headless"), "unexpected first line: {line}");
(child, lines)
}
/// With stdin at EOF (as soon as the host starts), a headless host stays up.
#[test]
fn headless_ignores_stdin_eof() {
let (mut child, _lines) = start_headless("headless-eof");
std::thread::sleep(std::time::Duration::from_millis(500));
let alive = child.try_wait().unwrap().is_none();
let _ = child.kill();
let _ = child.wait();
assert!(alive, "headless host exited on stdin EOF");
}
/// SIGTERM right after the ready line stops the actors and exits with 0.
#[test]
fn headless_stops_cleanly_on_sigterm_right_after_ready() {
let (mut child, lines) = start_headless("headless-term");
let pid = child.id().to_string();
assert!(Command::new("kill").args(["-TERM", &pid]).status().unwrap().success());
let rest: Vec<String> = lines.map(|l| l.unwrap()).collect();
let status = child.wait().unwrap();
assert!(status.success(), "exit status {status:?}, output {rest:?}");
assert!(rest.iter().any(|l| l.contains("Stopping actors")), "output {rest:?}");
}
@@ -1,6 +0,0 @@
[package]
name = "cargo-test-fail"
version = "0.1.0"
edition = "2024"
[workspace]
@@ -1,9 +0,0 @@
fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn deliberate_unit_test_failure() {
assert_eq!(2 + 2, 5);
}
}
-8
View File
@@ -1,9 +1 @@
fn main() {} fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn unit_tests_run_on_the_host_target() {
assert_eq!(2 + 2, 4);
}
}
+15 -189
View File
@@ -108,15 +108,15 @@ async fn routes_replies_publishes_and_reports_undeliverable() {
async fn broker_does_not_answer_messages_with_reply_to() { async fn broker_does_not_answer_messages_with_reply_to() {
let world = create_world("broker-reply-to"); let world = create_world("broker-reply-to");
let (host, mut mailbox) = start(&world).await; let (host, mut mailbox) = start(&world).await;
let untapped = request(&host, &mut mailbox, json!({"type": "untap"})).await; let scanned = request(&host, &mut mailbox, json!({"type": "scan"})).await;
assert_eq!(untapped.body["type"], "untapped"); assert_eq!(scanned.body["type"], "scanned");
host.broker.send( host.broker.send(
"console", "console",
Some("broker"), Some("broker"),
None, None,
Some("upstream-message"), Some("upstream-message"),
json!({"type": "untap"}), json!({"type": "scan"}),
); );
assert!( assert!(
tokio::time::timeout(Duration::from_secs(1), mailbox.recv()) tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
@@ -611,18 +611,14 @@ async fn unknown_broker_request_lists_the_known_ones() {
let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await; let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
assert_eq!(unknown.body["type"], "error"); assert_eq!(unknown.body["type"], "error");
let reason = unknown.body["reason"].as_str().unwrap(); let reason = unknown.body["reason"].as_str().unwrap();
assert!(reason.contains("remove") && reason.contains("tap") && reason.contains("watch"), "{reason}"); assert!(reason.contains("remove") && reason.contains("tap"), "{reason}");
// `scan` is retired: the host watches the world folder instead.
let scan = request(&host, &mut mailbox, json!({"type": "scan"})).await;
assert_eq!(scan.body["type"], "error");
host.stop().await; host.stop().await;
} }
#[tokio::test] #[tokio::test]
async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() { async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
let world = create_world("cargo-actor"); let world = create_world("cargo-actor");
let (host, mut mailbox) = start(&world).await; let (host, mut mailbox) = start(&world).await;
let valid_source = cargo_source("cargo-valid"); let valid_source = cargo_source("cargo-valid");
let valid_id = host.broker.send( let valid_id = host.broker.send(
"console", "console",
@@ -631,42 +627,12 @@ async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() {
None, None,
json!({"type": "build", "source": valid_source.display().to_string()}), json!({"type": "build", "source": valid_source.display().to_string()}),
); );
let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(180)).await; let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(120)).await;
assert_eq!(built.body["type"], "built", "{built:?}"); assert_eq!(built.body["type"], "built");
assert_eq!(built.body["tests"]["passed"], 1);
assert_eq!(built.body["tests"]["failed"], 0);
assert!(built.body["tests"]["output"].as_str().unwrap().contains("unit_tests_run_on_the_host_target"));
let wasm = built.body["wasm"].as_str().unwrap(); let wasm = built.body["wasm"].as_str().unwrap();
assert!(wasm.ends_with(".wasm")); assert!(wasm.ends_with(".wasm"));
assert!(host.world.resolve("console", wasm).unwrap().is_file()); assert!(host.world.resolve("console", wasm).unwrap().is_file());
let failing_test_source = cargo_source("cargo-test-fail");
let failed_test_id = host.broker.send(
"console",
Some("cargo"),
None,
None,
json!({"type": "build", "source": failing_test_source.display().to_string()}),
);
let failed_test = receive_reply(&mut mailbox, &failed_test_id, Duration::from_secs(120)).await;
assert_eq!(failed_test.body["type"], "build_failed", "{failed_test:?}");
assert_eq!(failed_test.body["tests"]["passed"], 0);
assert_eq!(failed_test.body["tests"]["failed"], 1);
assert!(failed_test.body["tests"]["output"].as_str().unwrap().contains("deliberate_unit_test_failure"));
let skipped_id = host.broker.send(
"console",
Some("cargo"),
None,
None,
json!({"type": "build", "source": failing_test_source.display().to_string(), "skip_tests": true}),
);
let skipped = receive_reply(&mut mailbox, &skipped_id, Duration::from_secs(120)).await;
assert_eq!(skipped.body["type"], "built", "{skipped:?}");
assert_eq!(skipped.body["tests"]["passed"], 0);
assert_eq!(skipped.body["tests"]["failed"], 0);
assert_eq!(skipped.body["tests"]["output"], "skipped (skip_tests=true)");
let invalid_source = cargo_source("cargo-fail"); let invalid_source = cargo_source("cargo-fail");
let invalid_id = host.broker.send( let invalid_id = host.broker.send(
"console", "console",
@@ -675,10 +641,14 @@ async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() {
None, None,
json!({"type": "build", "source": invalid_source.display().to_string()}), json!({"type": "build", "source": invalid_source.display().to_string()}),
); );
let invalid = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await; let failed = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
assert_eq!(invalid.body["type"], "build_failed", "{invalid:?}"); assert_eq!(failed.body["type"], "build_failed");
assert_eq!(invalid.body["tests"]["failed"], 1); assert!(
assert!(invalid.body["tests"]["output"].as_str().unwrap().contains("missing_function")); failed.body["errors"]
.as_str()
.unwrap()
.contains("missing_function")
);
host.stop().await; host.stop().await;
} }
@@ -1004,147 +974,3 @@ async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped"); assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
host.stop().await; host.stop().await;
} }
/// Ask `actor` again and again (it may not exist yet) until `want` holds for
/// the answer.
async fn ask_until(
host: &Host,
mailbox: &mut mpsc::Receiver<Envelope>,
actor: &str,
body: Value,
want: impl Fn(&Value) -> bool,
) -> Envelope {
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let reply = ask(host, mailbox, actor, body.clone()).await;
if want(&reply.body) {
return reply;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
})
.await
.unwrap_or_else(|_| panic!("{actor} never gave the wanted answer"))
}
/// The `fs_changed` bodies among `messages` (all from `fs`).
fn fs_events(messages: Vec<Envelope>) -> Vec<Value> {
messages
.into_iter()
.filter(|m| m.body["type"] == "fs_changed")
.map(|m| {
assert_eq!(m.from, "fs");
m.body
})
.collect()
}
#[tokio::test]
async fn a_new_actor_wasm_on_disk_becomes_an_actor() {
let world = create_world("watch-spawn");
let (host, mut mailbox) = start(&world).await;
// Any writer counts: here the test itself, like a shell on the host.
let tests = world.source_dir("dropped").join("tests");
std::fs::create_dir_all(&tests).unwrap();
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
std::fs::copy(toolbox_wasm(), world.wasm_path("dropped")).unwrap();
let echo = json!({"type": "echo", "value": "hi"});
let reply = ask_until(&host, &mut mailbox, "dropped", echo, |b| b["type"] == "echo").await;
assert_eq!(reply.from, "dropped");
assert_eq!(world.read_meta("dropped").version, Some(1));
let meta = std::fs::read_to_string(world.version_dir("dropped", 1).join("meta.json")).unwrap();
assert_eq!(serde_json::from_str::<Value>(&meta).unwrap()["requested_by"], "scan");
host.stop().await;
}
#[tokio::test]
async fn a_changed_actor_wasm_on_disk_is_an_upgrade() {
let world = create_world("watch-upgrade");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let source = passing_source(&world, "source");
spawn_type(&host, &mut mailbox, "changing", &wasm, &source).await;
// The same program plus a custom section: a different file.
let mut changed = std::fs::read(&wasm).unwrap();
changed.extend_from_slice(&[0, 4, 1, b'x', b'v', b'2']);
std::fs::write(world.wasm_path("changing"), changed).unwrap();
let version = json!({"type": "version"});
ask_until(&host, &mut mailbox, "changing", version, |b| b["version"] == "2").await;
assert_eq!(world.read_meta("changing").version, Some(2));
assert_eq!(world.versions("changing"), vec![1, 2]);
host.stop().await;
}
#[tokio::test]
async fn watchers_get_one_event_per_change_from_any_writer() {
let (world, host, mut mailbox) = filesystem_world("watch-events").await;
let watch = json!({"type": "watch", "path": "/actor/writer/public/"});
let watching = request(&host, &mut mailbox, watch).await;
assert_eq!(watching.body, json!({"type": "watching", "path": "/actor/writer/public"}));
assert!(world.read_meta("console").watches.contains("/actor/writer/public"));
drain(&mut mailbox, Duration::from_millis(300)).await;
let quiet = Duration::from_millis(600);
let event = |path: &str, op: &str| json!({"type": "fs_changed", "path": format!("/actor/writer/public/{path}"), "op": op});
// An actor creates a file: one create, not create + write.
let write = json!({"type": "write", "path": "/public/a.txt", "text": "a"});
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "create")]);
// The host (a shell, git, ...) changes it.
std::fs::write(world.public_dir("writer").join("a.txt"), "b").unwrap();
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("a.txt", "write")]);
// A rename is one event with `to`.
let rename = json!({"type": "rename", "path": "/public/a.txt", "to": "/public/b.txt"});
ask_type(&host, &mut mailbox, "writer", rename, "ok").await;
let mut renamed = event("a.txt", "rename");
renamed["to"] = json!("/actor/writer/public/b.txt");
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![renamed]);
std::fs::remove_file(world.public_dir("writer").join("b.txt")).unwrap();
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), vec![event("b.txt", "remove")]);
// Nothing outside the watched path.
let outside = json!({"type": "write", "path": "/data/x", "text": "x"});
ask_type(&host, &mut mailbox, "writer", outside, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
let unwatch = json!({"type": "unwatch", "path": "/actor/writer/public"});
let unwatched = request(&host, &mut mailbox, unwatch).await;
assert_eq!(unwatched.body, json!({"type": "unwatched", "path": "/actor/writer/public"}));
assert!(world.read_meta("console").watches.is_empty());
let write = json!({"type": "write", "path": "/public/c.txt", "text": "c"});
ask_type(&host, &mut mailbox, "writer", write, "ok").await;
assert_eq!(fs_events(drain(&mut mailbox, quiet).await), Vec::<Value>::new());
host.stop().await;
}
#[tokio::test]
async fn a_watcher_overflow_sends_one_rescan_and_runs_a_full_pass() {
let world = create_world("watch-overflow");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let mut bystander = host.mailbox("bystander");
let watch = json!({"type": "watch", "path": "/actor/elsewhere"});
assert_eq!(request(&host, &mut mailbox, watch).await.body["type"], "watching");
// A dropped actor that the full pass finds (before the watcher's quiet time).
let tests = world.source_dir("planted").join("tests");
std::fs::create_dir_all(&tests).unwrap();
let test = json!({"steps": [{"send": {"type": "echo", "value": "candidate"}, "expect": {"type": "echo", "value": "candidate"}}]});
std::fs::write(tests.join("candidate.json"), test.to_string()).unwrap();
std::fs::copy(&wasm, world.wasm_path("planted")).unwrap();
let report = host.broker.clone().overflowed().await;
assert!(report.iter().any(|r| r["type"] == "spawned" && r["actor"] == "planted"), "{report:?}");
let rescan = json!({"type": "fs_changed", "path": "/actor", "op": "rescan"});
assert_eq!(fs_events(drain(&mut mailbox, Duration::from_millis(600)).await), vec![rescan]);
assert!(bystander.try_recv().is_err(), "an actor without watches got the rescan");
host.stop().await;
}
Generated
-16
View File
@@ -1,20 +1,5 @@
{ {
"nodes": { "nodes": {
"crane": {
"locked": {
"lastModified": 1791754810,
"narHash": "sha256-cN0Mgk+sQtCJcGWs9H4M0CnLRAuC4I8aCwiBTcBJs6E=",
"owner": "ipetkov",
"repo": "crane",
"rev": "bd63bf1d0a0f6dcacd7549c714a671ee63b0db6a",
"type": "github"
},
"original": {
"owner": "ipetkov",
"repo": "crane",
"type": "github"
}
},
"flake-utils": { "flake-utils": {
"inputs": { "inputs": {
"systems": "systems" "systems": "systems"
@@ -51,7 +36,6 @@
}, },
"root": { "root": {
"inputs": { "inputs": {
"crane": "crane",
"flake-utils": "flake-utils", "flake-utils": "flake-utils",
"nixpkgs": "nixpkgs", "nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay" "rust-overlay": "rust-overlay"
+13 -101
View File
@@ -6,10 +6,9 @@
flake-utils.url = "github:numtide/flake-utils"; flake-utils.url = "github:numtide/flake-utils";
rust-overlay.url = "github:oxalica/rust-overlay"; rust-overlay.url = "github:oxalica/rust-overlay";
rust-overlay.inputs.nixpkgs.follows = "nixpkgs"; rust-overlay.inputs.nixpkgs.follows = "nixpkgs";
crane.url = "github:ipetkov/crane";
}; };
outputs = { nixpkgs, flake-utils, rust-overlay, crane, ... }: outputs = { nixpkgs, flake-utils, rust-overlay, ... }:
flake-utils.lib.eachSystem flake-utils.lib.allSystems (system: flake-utils.lib.eachSystem flake-utils.lib.allSystems (system:
let let
pkgs = import nixpkgs { pkgs = import nixpkgs {
@@ -21,41 +20,22 @@
targets = [ "wasm32-wasip2" ]; targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ]; extensions = [ "rust-src" ];
}; };
# The runtime cargo actor needs Cargo, rustc (with wasm32-wasip2), and rustPlatform = pkgs.makeRustPlatform { cargo = rust; rustc = rust; };
# the native linker, even when the host itself is started with a bare sausage = rustPlatform.buildRustPackage {
# PATH. Keep this same runtime path in the packaged host and exercise
# the installed binary below with PATH deliberately empty.
runtimePath = pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ];
craneLib = (crane.mkLib pkgs).overrideToolchain rust;
common = {
pname = "sausage"; pname = "sausage";
version = "0.1.0"; version = "0.1.0";
src = ./.; src = ./.;
}; cargoLock.lockFile = ./Cargo.lock;
# The test packages for `checks.tests`. cargoBuildFlags = [ "-p" "sausage" ];
testPackages = "-p sausage -p sausage-actor -p agent";
cargoArtifacts = craneLib.buildDepsOnly (common // {
doCheck = false;
# The dependencies of the host binary and of the tests.
buildPhaseCargoCommand = ''
cargoWithProfile build --locked -p sausage
cargoWithProfile test --locked --no-run ${testPackages}
'';
});
sausage = craneLib.buildPackage (common // {
inherit cargoArtifacts;
doNotRemoveReferencesToRustToolchain = true;
cargoExtraArgs = "--locked -p sausage";
doCheck = false; doCheck = false;
nativeBuildInputs = [ pkgs.makeWrapper ]; nativeBuildInputs = [ pkgs.makeWrapper ];
# The `cargo` actor runs cargo at runtime. Build scripts need a C linker. # The `cargo` actor runs cargo at runtime. Build scripts need a C linker.
postInstall = '' postInstall = ''
wrapProgram $out/bin/sausage --prefix PATH : ${runtimePath} wrapProgram $out/bin/sausage --prefix PATH : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]}
''; '';
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>`
# (without a folder: bare `sausage`, i.e. ./world, created if needed)
# from the public repo, again and again: after a crash or a broker # 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.
@@ -64,10 +44,9 @@
repoUrl = "git+https://gitea.surma.technology/surma/sausage"; repoUrl = "git+https://gitea.surma.technology/surma/sausage";
loop = pkgs.writeShellScriptBin "sausage-loop" '' loop = pkgs.writeShellScriptBin "sausage-loop" ''
url="''${SAUSAGE_FLAKE:-${repoUrl}}" url="''${SAUSAGE_FLAKE:-${repoUrl}}"
# 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" -- "$@" nix run --refresh "$url#default" -- run "$@"
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
@@ -81,82 +60,15 @@
# `nix flake check` runs every test of the host, the SDK, and the # `nix flake check` runs every test of the host, the SDK, and the
# agent. The tests build WASM actors with cargo, offline, from the # agent. The tests build WASM actors with cargo, offline, from the
# vendored crates of Cargo.lock. # vendored crates of Cargo.lock.
# The binary the host loop runs (`nix run`), built from the same checks.tests = sausage.overrideAttrs (old: {
# cached dependencies as the tests. pname = "sausage-tests";
checks.package = sausage; doCheck = true;
checks.tests = craneLib.cargoTest (common // { cargoTestFlags = [ "-p" "sausage" "-p" "sausage-actor" "-p" "agent" ];
inherit cargoArtifacts;
cargoExtraArgs = "--locked ${testPackages}";
# Nothing builds on the test run's target folder: don't store it.
doInstallCargoArtifacts = false;
preCheck = '' preCheck = ''
# Its own lock file pins crates that the vendored set may lack. # Its own lock file pins crates that the vendored set may lack.
rm -f cli/tests/actors/toolbox/Cargo.lock rm -f cli/tests/actors/toolbox/Cargo.lock
''; '';
}); });
# Exercise the installed host exactly as the loop launches it: the
# caller has no Cargo on PATH, so only the packaged runtime toolchain
# can run the native tests and the wasm32-wasip2 build.
checks.runtime-toolchain = pkgs.runCommand "sausage-runtime-toolchain" { } ''
set -eu
root="$TMPDIR/runtime-toolchain"
world="$root/world"
emptyPath="$root/empty-bin"
mkdir -p "$emptyPath" "$root/home" "$root/cargo-home"
for actor in broker console cargo fs; do
mkdir -p "$world/actor/$actor/public" "$world/actor/$actor/data"
done
source="$world/actor/console/public/probe"
mkdir -p "$source/src" "$source/tests"
cat > "$source/Cargo.toml" <<'TOML'
[package]
name = "runtime-probe"
version = "0.1.0"
edition = "2024"
TOML
printf 'fn main() {}\n' > "$source/src/main.rs"
cat > "$source/tests/probe.rs" <<'RUST'
#[test]
fn native_test_harness_runs() {
assert_eq!(2 + 2, 4);
}
RUST
echo "runtime-check: run packaged host with an empty PATH"
coproc host {
${pkgs.coreutils}/bin/env -i PATH="$emptyPath" HOME="$root/home" TMPDIR="$root" \
CARGO_HOME="$root/cargo-home" ${sausage}/bin/sausage run "$world"
}
hostInput=''${host[1]}
hostOutput=''${host[0]}
hostPid=$host_PID
trap 'kill "$hostPid" 2>/dev/null || true' EXIT
printf '%s\n' '@cargo {"type":"build","source":"/actor/console/public/probe"}' >&"$hostInput"
built=0
while IFS= read -r -t 300 line <&"$hostOutput"; do
echo "$line"
case "$line" in
*'"type":"build_failed"'*)
echo "packaged cargo actor build failed with an empty caller PATH" >&2
exit 1
;;
*'"type":"built"'*)
case "$line" in
*'"passed":1'*) built=1 ;;
*) echo "native test did not pass" >&2; exit 1 ;;
esac
break
;;
esac
done
if [ "$built" -ne 1 ]; then
echo "packaged host did not build the probe" >&2
exit 1
fi
printf '%s\n' '/quit' >&"$hostInput"
wait "$hostPid"
mkdir -p "$out"
touch "$out/passed"
'';
apps.default = { type = "app"; program = "${sausage}/bin/sausage"; }; apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; }; apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };