Compare commits

..
Author SHA1 Message Date
Sausage agent 8a533f0158 Run without a clone: bare sausage creates ./world if needed and runs it
ci / check (push) Successful in 14m24s
- cli: the subcommand is optional; without one, sausage runs ./world and
  runs init first when ./world is not a world yet (test: cli/tests/cli.rs).
- flake: .#loop without a folder runs bare sausage.
- README: nix run git+https://gitea.surma.technology/surma/sausage(#loop),
  clone only for working on Sausage.
2026-10-16 20:56:59 +00:00
18 changed files with 157 additions and 1303 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"
+12 -59
View File
@@ -68,7 +68,7 @@ Each instance sees three folders:
| `/actor` | read-only | the `actor/` folder of the world | | `/actor` | read-only | the `actor/` folder of the world |
`/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
@@ -85,13 +85,6 @@ Every actor can read and write them. The host turns each file operation into a
message from `fs` to `<id>` and waits up to 10 seconds for the reply. `path` message from `fs` to `<id>` and waits up to 10 seconds for the reply. `path`
is relative to `fs/` and starts with `/` (the folder itself is `/`). is relative to `fs/` and starts with `/` (the folder itself is `/`).
To make this virtual directory discoverable, an actor that serves it declares
`Actor-served files` and `fs_read` in its leading `//!` documentation in
`source/src/main.rs`. The host uses that declaration when listing
`/actor/<id>/`; it does not probe every actor (which could stall on actors that
do not handle filesystem messages). This is discovery metadata only: opening
and reading the path still requires a valid `fs_read` reply.
| Request | Reply | | Request | Reply |
| --- | --- | | --- | --- |
| `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` | | `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` |
@@ -110,15 +103,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 |
| --- | --- | --- | | --- | --- | --- |
@@ -129,7 +121,7 @@ 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 (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) |
Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`. Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
@@ -152,31 +144,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>/`
@@ -215,15 +182,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
@@ -231,19 +194,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
+14 -15
View File
@@ -18,35 +18,35 @@ nix run git+https://gitea.surma.technology/surma/sausage
``` ```
This creates a world in `./world` the first time (it builds the agent, which 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 takes a few minutes) and then starts it with a console. Later runs start the
existing world. The two steps on their own: existing world. The two steps also exist on their own:
```sh ```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 -- init # create ./world
nix run git+https://gitea.surma.technology/surma/sausage -- run # starts the world and a console nix run git+https://gitea.surma.technology/surma/sausage -- run # start it
``` ```
`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`. `init` and `run` take
folder as an argument, for example `... -- run my-world`. another folder as an argument, for example `-- run my-world`.
To keep a world on the newest version, run it in a loop: To keep a world on the newest version, run it in a loop:
```sh ```sh
nix run git+https://gitea.surma.technology/surma/sausage#loop # ./world, created if needed nix run git+https://gitea.surma.technology/surma/sausage#loop # ./world
nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world # an existing world 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 the newest `sausage` from that URL (`nix run --refresh`).
When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop
fetches the newest commit and starts it again. `/quit` ends the loop. 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 To work on Sausage itself, clone it
same commands are `nix run`, `nix run -- init`, `nix run -- run` and (`git clone https://gitea.surma.technology/surma/sausage`) and use `nix run`,
`nix run .#loop`. `nix run .#loop` and `nix develop` in the checkout.
In the console: In the console:
@@ -72,9 +72,8 @@ 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 })
}
_ => {} _ => {}
} }
} }
-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> {
+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();
} }
+1 -1
View File
@@ -85,7 +85,7 @@ 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(())
+15 -95
View File
@@ -4,8 +4,7 @@
//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`. //! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
//! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the //! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the
//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for //! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
//! the reply. Other paths go to `wasmtime-wasi` unchanged except that an //! the reply. Every other path goes to `wasmtime-wasi` unchanged.
//! actor directory listing advertises `fs/` when its source docs declare it.
//! //!
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}` //! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`, //! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
@@ -15,7 +14,7 @@
//! A file is read once when it is opened. After a change, it is written back //! A file is read once when it is opened. After a change, it is written back
//! in one `fs_write` when it is synced or closed. //! in one `fs_write` when it is synced or closed.
use std::collections::HashMap; use std::collections::HashSet;
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak}; use std::sync::{Arc, Mutex, Weak};
@@ -42,14 +41,13 @@ pub struct VfsState {
broker: Weak<Broker>, broker: Weak<Broker>,
/// The actor of this instance, for notes in its `stderr.log`. /// The actor of this instance, for notes in its `stderr.log`.
me: String, me: String,
/// Native directory descriptors and their absolute paths. Keeping every /// Descriptors of the `/actor` preopen.
/// `/actor` child (not only the preopen) lets `fs/` work after traversal. roots: HashSet<u32>,
dirs: HashMap<u32, String>,
} }
impl VfsState { impl VfsState {
pub fn new(broker: Weak<Broker>, me: String) -> Self { pub fn new(broker: Weak<Broker>, me: String) -> Self {
Self { broker, me, dirs: HashMap::new() } Self { broker, me, roots: HashSet::new() }
} }
} }
@@ -201,33 +199,6 @@ fn join(base: &str, rel: &str) -> String {
format!("/{}", parts.join("/")) format!("/{}", parts.join("/"))
} }
/// Split a global path under an actor's virtual `fs/` into its owner and
/// actor-relative path.
fn actor_fs_path(full: &str) -> Option<(String, String)> {
let rest = full.strip_prefix("/actor/")?;
let (actor, rest) = rest.split_once('/')?;
let (fs, path) = rest.split_once('/').unwrap_or((rest, ""));
(fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{path}")))
}
fn actor_root(path: &str) -> Option<&str> {
let actor = path.strip_prefix("/actor/")?;
(!actor.is_empty() && !actor.contains('/')).then_some(actor)
}
/// An actor advertises its virtual filesystem in its leading `//!` docs.
/// This is discovery metadata; actual reads still go through `fs_read`.
fn declares_actor_fs(source: &str) -> bool {
let docs = source
.lines()
.take_while(|line| line.trim().starts_with("//!"))
.map(|line| line.trim().trim_start_matches("//!").trim())
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase();
docs.contains("actor-served files") && docs.contains("fs_read")
}
fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat { fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
types::DescriptorStat { types::DescriptorStat {
type_, type_,
@@ -295,14 +266,6 @@ impl VfsView<'_> {
Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() } Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
} }
fn actor_serves_fs(&self, actor: &str) -> bool {
self.state
.broker
.upgrade()
.and_then(|broker| std::fs::read_to_string(broker.world.source_dir(actor).join("src/main.rs")).ok())
.is_some_and(|source| declares_actor_fs(&source))
}
fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> { fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned() self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
} }
@@ -319,27 +282,16 @@ impl VfsView<'_> {
match self.virt(fd) { match self.virt(fd) {
Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))), Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))), Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
None => { None if self.state.roots.contains(&fd.rep()) => {
let base = self.state.dirs.get(&fd.rep())?; let full = join("/", path);
actor_fs_path(&join(base, path)) let mut parts = full.splitn(4, '/').skip(1);
let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or(""));
(fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}")))
} }
None => None,
} }
} }
async fn native_directory_entries(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Vec<DirectoryEntry>> {
let stream = HostDescriptor::read_directory(&mut self.inner, fd).await?;
let stream_rep = stream.rep();
let mut entries = Vec::new();
loop {
match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, Resource::new_borrow(stream_rep)).await? {
Some(entry) => entries.push(entry),
None => break,
}
}
HostDirectoryEntryStream::drop(&mut self.inner, stream).map_err(|_| ErrorCode::Io)?;
Ok(entries)
}
fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> { fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
let r = self.inner.table.push(v)?; let r = self.inner.table.push(v)?;
Ok(Resource::new_own(r.rep())) Ok(Resource::new_own(r.rep()))
@@ -391,7 +343,7 @@ impl preopens::Host for VfsView<'_> {
let dirs = preopens::Host::get_directories(&mut self.inner)?; let dirs = preopens::Host::get_directories(&mut self.inner)?;
for (fd, name) in &dirs { for (fd, name) in &dirs {
if name == "/actor" { if name == "/actor" {
self.state.dirs.insert(fd.rep(), "/actor".to_string()); self.state.roots.insert(fd.rep());
} }
} }
Ok(dirs) Ok(dirs)
@@ -501,19 +453,7 @@ impl HostDescriptor for VfsView<'_> {
Node::File(_) => Err(ErrorCode::NotDirectory.into()), Node::File(_) => Err(ErrorCode::NotDirectory.into()),
}, },
Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()), Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
None => { None => HostDescriptor::read_directory(&mut self.inner, fd).await,
let actor = self.state.dirs.get(&fd.rep()).and_then(|path| actor_root(path)).map(str::to_string);
if actor.as_deref().is_some_and(|id| self.actor_serves_fs(id)) {
let mut entries = self.native_directory_entries(fd).await?;
if !entries.iter().any(|entry| entry.name == "fs") {
entries.push(DirectoryEntry { type_: DescriptorType::Directory, name: "fs".to_string() });
}
let r = self.inner.table.push(VDir(entries.into_iter()))?;
Ok(Resource::new_own(r.rep()))
} else {
HostDescriptor::read_directory(&mut self.inner, fd).await
}
}
} }
} }
@@ -585,16 +525,7 @@ impl HostDescriptor for VfsView<'_> {
) -> FsResult<Resource<types::Descriptor>> { ) -> FsResult<Resource<types::Descriptor>> {
match self.target(&fd, &path) { match self.target(&fd, &path) {
Some((actor, path)) => self.open(actor, path, oflags, flags).await, Some((actor, path)) => self.open(actor, path, oflags, flags).await,
None => { None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await,
let child_path = self.state.dirs.get(&fd.rep()).map(|base| join(base, &path));
let opened = HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await?;
if let Some(path) = child_path {
if HostDescriptor::get_type(&mut self.inner, Resource::new_borrow(opened.rep())).await? == DescriptorType::Directory {
self.state.dirs.insert(opened.rep(), path);
}
}
Ok(opened)
}
} }
} }
@@ -607,7 +538,7 @@ impl HostDescriptor for VfsView<'_> {
} }
return Ok(()); return Ok(());
} }
self.state.dirs.remove(&fd.rep()); self.state.roots.remove(&fd.rep());
HostDescriptor::drop(&mut self.inner, fd) HostDescriptor::drop(&mut self.inner, fd)
} }
@@ -767,17 +698,6 @@ fn base64_decode(text: &str) -> Option<Vec<u8>> {
mod tests { mod tests {
use super::*; use super::*;
#[test]
fn actor_fs_paths_and_discovery_docs() {
assert_eq!(actor_fs_path("/actor/brain/fs"), Some(("brain".to_string(), "/".to_string())));
assert_eq!(actor_fs_path("/actor/brain/fs/docs/readme.md"), Some(("brain".to_string(), "/docs/readme.md".to_string())));
assert_eq!(actor_fs_path("/actor/brain/public/fs/file"), None);
assert_eq!(actor_root("/actor/brain"), Some("brain"));
assert_eq!(actor_root("/actor/brain/fs"), None);
assert!(declares_actor_fs("//! brain\n//! Actor-served files include fs_read.\nfn main() {}"));
assert!(!declares_actor_fs("//! actor without fs support\nfn main() {}"));
}
#[test] #[test]
fn base64_round_trip() { fn base64_round_trip() {
for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] { for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
-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 {
@@ -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 -254
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())
@@ -246,71 +246,6 @@ async fn actors_serve_files_under_fs() {
host.stop().await; host.stop().await;
} }
#[tokio::test]
async fn actor_fs_is_discoverable_only_when_its_source_docs_advertise_it() {
let world = create_world("actor-fs-discovery");
let wasm = toolbox_wasm().clone();
let (host, mut mailbox) = start(&world).await;
let advertised = passing_source(&world, "advertised-fs");
std::fs::create_dir_all(advertised.join("src")).unwrap();
std::fs::write(
advertised.join("src/main.rs"),
"//! test actor\n//! Actor-served files: implements fs_read and fs_dir.\n",
)
.unwrap();
spawn_type(&host, &mut mailbox, "served", &wasm, &advertised).await;
let undocumented = passing_source(&world, "undocumented-fs");
std::fs::create_dir_all(undocumented.join("src")).unwrap();
std::fs::write(undocumented.join("src/main.rs"), "//! test actor without a virtual filesystem\n").unwrap();
spawn_type(&host, &mut mailbox, "plain", &wasm, &undocumented).await;
let reader_source = passing_source(&world, "fs-reader");
spawn_type(&host, &mut mailbox, "reader", &wasm, &reader_source).await;
let served_entries = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "list", "path": "/actor/served"}),
"entries",
)
.await;
assert!(served_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
let plain_entries = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "list", "path": "/actor/plain"}),
"entries",
)
.await;
assert!(!plain_entries.body["names"].as_array().unwrap().iter().any(|name| name.as_str() == Some("fs")));
let path = "/actor/served/fs/readme.txt";
ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "write", "path": path, "text": "virtual document"}),
"ok",
)
.await;
let document = ask_type(
&host,
&mut mailbox,
"reader",
json!({"type": "read", "path": path}),
"content",
)
.await;
assert_eq!(document.body["text"], "virtual document");
host.stop().await;
}
#[tokio::test] #[tokio::test]
async fn files_are_plain_posix_files() { async fn files_are_plain_posix_files() {
let (_, host, mut mailbox) = filesystem_world("posix").await; let (_, host, mut mailbox) = filesystem_world("posix").await;
@@ -611,18 +546,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 +562,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 +576,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 +909,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"
+11 -97
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,39 +20,21 @@
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) # (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
@@ -81,82 +62,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"; };