Compare commits

...
Author SHA1 Message Date
Sausage Agent 6923ca3e60 preserve the packaged runtime Rust toolchain
ci / check (push) Successful in 6s
2026-10-17 16:36:08 +00:00
Sausage Agent c504532513 drive packaged host runtime check over pipes
ci / check (push) Failing after 1m16s
2026-10-17 16:33:31 +00:00
Sausage Agent eb35a79698 trace the packaged host runtime check
ci / check (push) Failing after 1m47s
2026-10-17 16:29:02 +00:00
Sausage Agent bfcd71e2f4 test packaged host cargo runtime with an empty PATH
ci / check (push) Failing after 1m33s
2026-10-17 16:25:14 +00:00
agent-pool-12 172a776eec cargo: annotate test summary count parsing
ci / check (push) Successful in 5s
2026-10-17 16:00:54 +00:00
agent-pool-12 6a9d35ceb8 cargo: preserve build failures in result wrapper
ci / check (push) Failing after 6m15s
2026-10-17 15:52:37 +00:00
agent-pool-12 22590c608d cargo: run host unit tests before builds
ci / check (push) Failing after 8m51s
2026-10-17 15:50:09 +00:00
Sausage Agent 3bc462ffd2 agent seed: thinking levels low|medium|high|xhigh|max|off on both APIs (card #128)
ci / check (push) Successful in 5s
New config field `thinking` (OpenAI: reasoning_effort still wins): Anthropic adaptive thinking +
output_config.effort (thinking blocks kept and sent back), OpenAI reasoning.effort. A 400 that rejects
the effort is retried with the nearest accepted level (from the error's list, lower on a tie), logged,
never dropped to off.
2026-10-17 15:28:44 +00:00
Sausage Agent f008d0cc60 nix: lock crane; don't store the test run's target folder
ci / check (push) Successful in 39s
2026-10-17 08:02:08 +00:00
Sausage Agent 81daddfd70 nix: build with crane, dependencies cached separately
ci / check (push) Successful in 10m50s
buildDepsOnly builds the dependencies of the host and the tests once
(keyed only on Cargo.toml/Cargo.lock); packages.default and checks.tests
reuse them, so a change to our crates no longer recompiles every crate.
checks.package builds the binary the host loop runs.
2026-10-17 07:49:54 +00:00
agent-pool-9 e8df22fe06 host tests: reuse the existing drain helper
ci / check (push) Successful in 8s
2026-10-17 07:23:30 +00:00
agent-pool-9 0995824c01 Watch the world folder: auto spawn/upgrade, broker watch/unwatch, retire scan
ci / check (push) Failing after 7m36s
The host watches the whole world folder recursively with the notify crate
(any writer counts, .sausage-tmp-* skipped, ~100 ms coalescing per path).
A new or changed actor/<id>/actor.wasm (or new folder actor/<id>) is
booted as at start once quiet: new = spawn (tested first), changed =
upgrade. The broker `scan` request is gone; the boot pass stays, plus a
full pass when the OS watcher overflows.

New broker requests watch/unwatch {path} (saved in actor.json `watches`)
deliver fs_changed {path, op: write|create|remove|rename, to?} from `fs`,
not routed, dropped when the mailbox is full; overflow sends every watcher
one fs_changed {path:"/actor", op:"rescan"}.

Card #91 (multi-host step 1).
2026-10-17 07:12:04 +00:00
agent-pool-8 6a136c1b17 Borrow host descriptors while tracking VFS paths
ci / check (push) Successful in 19s
2026-10-16 21:47:17 +00:00
agent-pool-8 3e90cd73d3 Keep discovery test file at served root 2026-10-16 21:47:17 +00:00
agent-pool-8 845c961f4f Expose advertised actor-served files in directory listings 2026-10-16 21:47:17 +00:00
Sausage agent d6d3c9a2f2 Bare sausage creates ./world if needed and runs it (default app)
ci / check (push) Successful in 5s
- 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).
  So `nix run git+https://gitea.surma.technology/surma/sausage` works.
- flake: .#loop without a folder runs bare sausage.
- README: mention the bare command and the bare loop.
2026-10-16 21:09:06 +00:00
19 changed files with 1362 additions and 153 deletions
Generated
+97
View File
@@ -678,6 +678,15 @@ 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"
@@ -1066,6 +1075,26 @@ 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"
@@ -1165,6 +1194,26 @@ 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"
@@ -1259,6 +1308,7 @@ 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",
] ]
@@ -1269,6 +1319,33 @@ 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"
@@ -1580,6 +1657,15 @@ 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"
@@ -1590,6 +1676,7 @@ dependencies = [
"http", "http",
"http-body-util", "http-body-util",
"include_dir", "include_dir",
"notify",
"serde", "serde",
"serde_json", "serde_json",
"tokio", "tokio",
@@ -2075,6 +2162,16 @@ 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"
+59 -12
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), `source/`, `public/`, `data/`, (installed version, running, subscriptions, tapped, watches), `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,6 +85,13 @@ 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"}]}` |
@@ -103,14 +110,15 @@ 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. - `fs` is reserved: no actor can have that ID. It also sends `fs_changed`
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`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`. `tapping`, `untapped`, `watching`, `unwatched`, `exiting`), or `{"type": "error", "reason": "..."}`.
| `type` | Fields | Effect | | `type` | Fields | Effect |
| --- | --- | --- | | --- | --- | --- |
@@ -121,7 +129,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 |
| `scan` | | boot new or changed `actor.wasm` files | | `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 |
| `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/...`.
@@ -144,6 +152,31 @@ 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>/`
@@ -182,11 +215,15 @@ 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, and on `scan`, the broker boots every **The folder is the truth.** At start, the broker boots every
`actor/<id>/actor.wasm`. If the file differs from the installed version, it `actor/<id>/actor.wasm`. Afterwards, the file watcher reports a new or
is a new version, tested with the tests in `actor/<id>/source/`. If its changed `actor/<id>/actor.wasm` (or a new folder `actor/<id>/`) once it is
tests fail, the file stays where it is, the reason goes to `stderr.log`, and quiet for about 100 ms: a new file becomes a new actor, a changed one an
the next scan tries again. upgrade, tested first with the tests in `actor/<id>/source/`. A file equal
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
@@ -194,9 +231,19 @@ the next scan tries again.
{"type": "build", "source": "/actor/agent/public/drafts/notes"} {"type": "build", "source": "/actor/agent/public/drafts/notes"}
``` ```
The reply is `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "built_by": "cargo", ...}` The build request may include `"skip_tests": true` to skip Rust unit tests.
or `{"type": "build_failed", "errors": "..."}`. Pass `wasm`, `source`, and By default, `cargo` first runs `cargo test` for the host target (not WASI),
`built_by` on to `spawn` or `upgrade`. so pure-logic unit tests in a `wasm32-wasip2` actor crate can run without a
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
+16 -5
View File
@@ -14,7 +14,15 @@ You need [Nix](https://nixos.org/download) with flakes, and an API token for
an LLM (OpenAI Responses or Anthropic Messages API). No clone needed: an LLM (OpenAI Responses or Anthropic Messages API). No clone needed:
```sh ```sh
nix run git+https://gitea.surma.technology/surma/sausage -- init # creates ./world and builds the agent (a few minutes) nix run git+https://gitea.surma.technology/surma/sausage
```
This creates a world in `./world` the first time (it builds the agent, which
takes a few minutes), then starts it with a console. Later runs start the
existing world. The two steps on their own:
```sh
nix run git+https://gitea.surma.technology/surma/sausage -- init # creates ./world and builds the agent
nix run git+https://gitea.surma.technology/surma/sausage -- run # starts the world and a console nix run git+https://gitea.surma.technology/surma/sausage -- run # starts the world and a console
``` ```
@@ -27,7 +35,8 @@ 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 -- my-world nix run git+https://gitea.surma.technology/surma/sausage#loop # ./world, created if needed
nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world # an existing world
``` ```
The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`. The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
@@ -36,7 +45,8 @@ 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 From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
same commands are `nix run -- init`, `nix run -- run` and `nix run .#loop`. same commands are `nix run`, `nix run -- init`, `nix run -- run` and
`nix run .#loop`.
In the console: In the console:
@@ -62,8 +72,9 @@ 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. Then restart the world or with the tests in `source/tests/`) into its folder; the host watches the
send `@broker {"type":"scan"}`. The broker tests the file before it uses it. world folder and picks the file up once it stops changing. The broker tests
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
+84 -10
View File
@@ -14,11 +14,49 @@ 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 ("low", "medium", ...). /// Optional reasoning effort for OpenAI models only (wins over `thinking` there).
#[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)]
@@ -32,7 +70,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). /// A provider item that only this API shape understands (OpenAI reasoning, Anthropic thinking).
Opaque { api: String, item: Value }, Opaque { api: String, item: Value },
} }
@@ -43,10 +81,28 @@ 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> {
match cfg.api.as_str() { let want = match cfg.api.as_str() {
"openai-responses" => openai(cfg, system, tools, history).await, "openai-responses" => cfg.reasoning_effort.clone().or_else(|| cfg.thinking.clone()),
"anthropic-messages" => anthropic(cfg, system, tools, history).await, _ => cfg.thinking.clone(),
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,
}
} }
} }
@@ -67,7 +123,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]) -> Result<Vec<Entry>, String> { async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry], effort: Option<&str>) -> 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 {
@@ -104,7 +160,7 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -
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) = &cfg.reasoning_effort { if let Some(effort) = 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);
@@ -143,7 +199,14 @@ async fn openai(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -
Ok(out) Ok(out)
} }
async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]) -> Result<Vec<Entry>, String> { async fn anthropic(
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() {
@@ -163,6 +226,8 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
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 { .. } => {}
} }
} }
@@ -171,13 +236,19 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
.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 body = json!({ let mut 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,
@@ -199,6 +270,9 @@ async fn anthropic(cfg: &Config, system: &str, tools: &[Tool], history: &[Entry]
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,6 +15,7 @@ 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"] }
+111 -23
View File
@@ -1,11 +1,12 @@
//! The broker: message routing, mailboxes, subscriptions, taps, the version //! The broker: message routing, mailboxes, subscriptions, taps, watches, the
//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start, //! version store, and the requests to `broker` (spawn, upgrade, rollback,
//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` //! stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch,
//! runs each WASM actor. //! exit). `supervisor.rs` runs each WASM actor.
//! //!
//! The broker accepts only `.wasm` files. It does not know any source language. //! The broker accepts only `.wasm` files. It does not know any source language.
//! The world folder is the source of truth: `scan` boots every //! The world folder is the source of truth: at start, the broker boots every
//! `actor/<id>/actor.wasm` and treats a changed file as an upgrade. //! `actor/<id>/actor.wasm`; afterwards the file watcher (`watch.rs`) reports
//! 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;
@@ -167,14 +168,19 @@ 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);
let body = json!({"type": "tapped", "message": message, "ts": ts}); self.inject(tappers, "broker", 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 tapper in tappers { for id in to {
if let Some(tx) = mailboxes.get(tapper) { if let Some(tx) = mailboxes.get(id) {
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: "broker".into(), from: from.into(),
to: tapper.clone(), to: id.clone(),
reply_to: None, reply_to: None,
body: body.clone(), body: body.clone(),
}); });
@@ -406,31 +412,90 @@ impl Broker {
Ok(json!({"type": "removed", "actor": id})) Ok(json!({"type": "removed", "actor": id}))
} }
/// Boot the actors in `actor/`. Called at start and on `scan`. /// Boot the actors in `actor/`: at start, and as a full pass when the
/// 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 for the next scan. /// nothing changes, and the file stays until it changes again.
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 NATIVE.contains(&id.as_str()) || !self.world.wasm_path(&id).is_file() { if let Some(r) = self.boot_actor(&id).await {
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> {
let meta = self.world.read_meta(id); // The broker's copy wins: actors change theirs (subscribe, ...) without the op lock.
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)
{ {
@@ -529,14 +594,32 @@ impl Broker {
let answer = if on { "tapping" } else { "untapped" }; let answer = if on { "tapping" } else { "untapped" };
Ok(json!({"type": answer})) Ok(json!({"type": answer}))
} }
"scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})), "watch" | "unwatch" => {
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, scan, exit"), other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, watch, unwatch, exit"),
} }
} }
@@ -563,6 +646,11 @@ 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();
+173 -47
View File
@@ -1,8 +1,9 @@
//! 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>"}`
//! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", ...}` //! optional `"skip_tests": true` skips native `cargo test` (tests run by default).
//! `{"type": "build_failed", "errors": "..."}` //! Replies: `{"type": "built", "wasm": "/actor/cargo/public/builds/<n>/<name>.wasm", "tests": {"passed": 1, "failed": 0, "output": "..."}, ...}`
//! `{"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.
@@ -32,11 +33,49 @@ 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.
@@ -51,51 +90,84 @@ impl Cargo {
Ok(Self { world }) Ok(Self { world })
} }
/// Build the crate in `source` (a host path) and publish the `.wasm` file /// Run native `cargo test` unless opted out, then build and publish the
/// under `/actor/cargo/public/builds/<n>/`. /// crate's `wasm32-wasip2` binary under `/actor/cargo/public/builds/<n>/`.
pub async fn build(&self, source: &Path) -> Result<BuildOutput> { pub async fn build(&self, source: &Path, skip_tests: bool) -> std::result::Result<BuildOutput, BuildFailure> {
if !source.join("Cargo.toml").is_file() { if !source.join("Cargo.toml").is_file() {
bail!("{} has no Cargo.toml", source.display()); return Err(BuildFailure {
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");
std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&self.world.public_dir(ID).join("sdk")))?)?; let original = std::fs::read_to_string(&manifest).context("read build manifest")?;
let prepared = prepare_manifest(&original, Some(&self.world.public_dir(ID).join("sdk")))?;
std::fs::write(&manifest, prepared).context("write prepared manifest")?;
let mut cmd = Command::new("cargo"); let target_dir = data.join("target");
cmd.arg("build") let tests = if skip_tests {
.arg("--release") TestResults { output: "skipped (skip_tests=true)".into(), ..TestResults::default() }
.arg("--target") } else {
.arg(TARGET) match run_tests(&build_dir, &target_dir).await {
.arg("--message-format=json-render-diagnostics") Ok(tests) => tests,
.arg("--target-dir") Err(error) => {
.arg(data.join("target")) let output = format!("cannot run cargo tests: {error:#}");
.current_dir(&build_dir) return Err(BuildFailure {
.kill_on_drop(true); reason: output.clone(),
let out = cmd.output().await.context("cannot run cargo (is it on PATH, with the wasm32-wasip2 target?)")?; tests: TestResults { failed: 1, output, ..TestResults::default() },
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));
} }
};
if tests.failed > 0 {
return Err(BuildFailure { reason: "unit tests failed".into(), tests });
} }
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 build_result: Result<BuildOutput> = async {
let dest_dir = self.world.public_dir(ID).join("builds").join(n.to_string()); let mut cmd = Command::new("cargo");
let name = artifact.file_name().context("artifact has no file name")?; cmd.arg("build")
let dest = dest_dir.join(name); .arg("--release")
world::write_atomic(&dest, &std::fs::read(&artifact)?)?; .arg("--target")
Ok(BuildOutput { wasm: dest, seconds: started.elapsed().as_secs_f64(), warnings: strip_progress(&stderr) }) .arg(TARGET)
.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;
build_result.map_err(|error| BuildFailure { reason: format!("{error:#}"), tests })
} }
/// The native `cargo` actor. Builds run one at a time, in order. /// The native `cargo` actor. Builds run one at a time, in order.
@@ -106,7 +178,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:#}")}) json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}"), "tests": TestResults::default().value()})
}); });
broker.send(ID, Some(&env.from), None, Some(&env.id), body); broker.send(ID, Some(&env.from), None, Some(&env.id), body);
} }
@@ -117,22 +189,76 @@ 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 out = self.build(&source).await?; let skip_tests = env.body.get("skip_tests").and_then(Value::as_bool).unwrap_or(false);
let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string()); match self.build(&source, skip_tests).await {
Ok(json!({ Ok(out) => {
"type": "built", let wasm = self.world.global_path(&out.wasm).unwrap_or_else(|| out.wasm.display().to_string());
"wasm": wasm, Ok(json!({
"source": source_s, "type": "built",
"built_by": ID, "wasm": wasm,
"seconds": (out.seconds * 10.0).round() / 10.0, "source": source_s,
"warnings": out.warnings, "built_by": ID,
})) "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> {
+15 -7
View File
@@ -1,7 +1,8 @@
//! 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), and one supervisor per WASM actor. See `PROTOCOL.md`. //! console), one supervisor per WASM actor, and the file watcher of the
//! world folder (`watch.rs`). See `PROTOCOL.md`.
pub mod broker; pub mod broker;
pub mod candidate; pub mod candidate;
@@ -10,6 +11,7 @@ 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;
@@ -30,12 +32,13 @@ 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`, and every registered actor of the world at /// Start the broker, `cargo`, the file watcher, and every registered
/// `root`. Only one host can run a world at a time. /// actor of the world at `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() {
@@ -53,10 +56,14 @@ 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, _lock: lock }) Ok(Host { world, broker, native, watcher, _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.
@@ -64,9 +71,10 @@ impl Host {
self.broker.open_mailbox(id) self.broker.open_mailbox(id)
} }
/// Stop the broker and `cargo` (with the requests they still run), then /// Stop the file watcher, the broker and `cargo` (with the requests they
/// every actor, gracefully. /// still run), then every actor, gracefully.
pub async fn stop(self) { pub async fn stop(mut self) {
drop(self.watcher.take());
for task in &self.native { for task in &self.native {
task.abort(); task.abort();
} }
+14 -4
View File
@@ -12,11 +12,13 @@ use sausage::{Host, cargo_actor, console};
/// The source of the agent actor. `init` builds it like any other actor. /// The source of the agent actor. `init` builds it like any other actor.
static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent"); static AGENT_SOURCE: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/../agent");
// Without a command, `sausage` runs the world in `./world` and creates it
// first (like `init`) if it does not exist yet.
#[derive(Parser)] #[derive(Parser)]
#[command(version, about)] #[command(version, about)]
struct Cli { struct Cli {
#[command(subcommand)] #[command(subcommand)]
command: Cmd, command: Option<Cmd>,
} }
#[derive(Subcommand)] #[derive(Subcommand)]
@@ -83,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).await.context("build agent")?; let out = cargo_actor::Cargo::new(world.clone())?.build(&source, false).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(())
@@ -112,7 +114,15 @@ async fn run(dir: PathBuf) -> Result<()> {
#[tokio::main] #[tokio::main]
async fn main() -> Result<()> { async fn main() -> Result<()> {
match Cli::parse().command { match Cli::parse().command {
Cmd::Init { dir } => init(dir).await, Some(Cmd::Init { dir }) => init(dir).await,
Cmd::Run { dir } => run(dir).await, Some(Cmd::Run { dir }) => run(dir).await,
None => {
let dir = PathBuf::from("world");
if !dir.join("actor").is_dir() {
println!("No world in ./world yet, creating it (same as `sausage init`).");
init(dir.clone()).await?;
}
run(dir).await
}
} }
} }
+95 -15
View File
@@ -4,7 +4,8 @@
//! 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. Every other path goes to `wasmtime-wasi` unchanged. //! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
//! 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}`,
@@ -14,7 +15,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::HashSet; use std::collections::HashMap;
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};
@@ -41,13 +42,14 @@ 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,
/// Descriptors of the `/actor` preopen. /// Native directory descriptors and their absolute paths. Keeping every
roots: HashSet<u32>, /// `/actor` child (not only the preopen) lets `fs/` work after traversal.
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, roots: HashSet::new() } Self { broker, me, dirs: HashMap::new() }
} }
} }
@@ -199,6 +201,33 @@ 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_,
@@ -266,6 +295,14 @@ 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()
} }
@@ -282,16 +319,27 @@ 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 if self.state.roots.contains(&fd.rep()) => { None => {
let full = join("/", path); let base = self.state.dirs.get(&fd.rep())?;
let mut parts = full.splitn(4, '/').skip(1); actor_fs_path(&join(base, path))
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()))
@@ -343,7 +391,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.roots.insert(fd.rep()); self.state.dirs.insert(fd.rep(), "/actor".to_string());
} }
} }
Ok(dirs) Ok(dirs)
@@ -453,7 +501,19 @@ 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 => HostDescriptor::read_directory(&mut self.inner, fd).await, None => {
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
}
}
} }
} }
@@ -525,7 +585,16 @@ 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 => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await, None => {
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)
}
} }
} }
@@ -538,7 +607,7 @@ impl HostDescriptor for VfsView<'_> {
} }
return Ok(()); return Ok(());
} }
self.state.roots.remove(&fd.rep()); self.state.dirs.remove(&fd.rep());
HostDescriptor::drop(&mut self.inner, fd) HostDescriptor::drop(&mut self.inner, fd)
} }
@@ -698,6 +767,17 @@ 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
@@ -0,0 +1,256 @@
//! 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())),
]
);
}
}
+6 -2
View File
@@ -1,11 +1,12 @@
//! 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/`. //! `sausage run` boots every actor that it finds in `actor/`, and then
//! 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 (broker) //! actor.json version, running, subscriptions, tapped, watches (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
@@ -46,6 +47,9 @@ 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 {
+41
View File
@@ -0,0 +1,41 @@
//! Tests of the `sausage` command line.
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::time::{SystemTime, UNIX_EPOCH};
fn empty_dir(label: &str) -> PathBuf {
let nonce = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.join("target/test-cli")
.join(format!("{label}-{}-{nonce}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
dir
}
/// Bare `sausage` (as in `nix run <flake>`) without a world starts `init`
/// for ./world instead of printing a usage error.
#[test]
fn bare_command_without_a_world_starts_init() {
let dir = empty_dir("bare");
let output = Command::new(env!("CARGO_BIN_EXE_sausage"))
.current_dir(&dir)
.env_remove("SAUSAGE_API")
.env_remove("SAUSAGE_BASE_URL")
.env_remove("SAUSAGE_MODEL")
.env_remove("SAUSAGE_API_TOKEN")
.stdin(Stdio::null())
.output()
.unwrap();
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
// With no answers on stdin, init stops at the first required value.
assert!(!output.status.success(), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(stdout.contains("No world in ./world yet"), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(stderr.contains("SAUSAGE_BASE_URL"), "stdout:\n{stdout}\nstderr:\n{stderr}");
assert!(!stderr.contains("Usage:"), "stderr:\n{stderr}");
// It asks before it creates anything.
assert!(!dir.join("world").exists());
}
@@ -0,0 +1,6 @@
[package]
name = "cargo-test-fail"
version = "0.1.0"
edition = "2024"
[workspace]
@@ -0,0 +1,9 @@
fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn deliberate_unit_test_failure() {
assert_eq!(2 + 2, 5);
}
}
+8
View File
@@ -1 +1,9 @@
fn main() {} fn main() {}
#[cfg(test)]
mod tests {
#[test]
fn unit_tests_run_on_the_host_target() {
assert_eq!(2 + 2, 4);
}
}
+254 -15
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 scanned = request(&host, &mut mailbox, json!({"type": "scan"})).await; let untapped = request(&host, &mut mailbox, json!({"type": "untap"})).await;
assert_eq!(scanned.body["type"], "scanned"); assert_eq!(untapped.body["type"], "untapped");
host.broker.send( host.broker.send(
"console", "console",
Some("broker"), Some("broker"),
None, None,
Some("upstream-message"), Some("upstream-message"),
json!({"type": "scan"}), json!({"type": "untap"}),
); );
assert!( assert!(
tokio::time::timeout(Duration::from_secs(1), mailbox.recv()) tokio::time::timeout(Duration::from_secs(1), mailbox.recv())
@@ -246,6 +246,71 @@ 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;
@@ -546,14 +611,18 @@ 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}"); assert!(reason.contains("remove") && reason.contains("tap") && reason.contains("watch"), "{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_builds_valid_crates_and_reports_compiler_errors() { async fn cargo_actor_runs_unit_tests_by_default_and_can_skip_them() {
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",
@@ -562,12 +631,42 @@ async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
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(120)).await; let built = receive_reply(&mut mailbox, &valid_id, Duration::from_secs(180)).await;
assert_eq!(built.body["type"], "built"); assert_eq!(built.body["type"], "built", "{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",
@@ -576,14 +675,10 @@ async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
None, None,
json!({"type": "build", "source": invalid_source.display().to_string()}), json!({"type": "build", "source": invalid_source.display().to_string()}),
); );
let failed = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await; let invalid = receive_reply(&mut mailbox, &invalid_id, Duration::from_secs(120)).await;
assert_eq!(failed.body["type"], "build_failed"); assert_eq!(invalid.body["type"], "build_failed", "{invalid:?}");
assert!( assert_eq!(invalid.body["tests"]["failed"], 1);
failed.body["errors"] assert!(invalid.body["tests"]["output"].as_str().unwrap().contains("missing_function"));
.as_str()
.unwrap()
.contains("missing_function")
);
host.stop().await; host.stop().await;
} }
@@ -909,3 +1004,147 @@ 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,5 +1,20 @@
{ {
"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"
@@ -36,6 +51,7 @@
}, },
"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"
+101 -13
View File
@@ -6,9 +6,10 @@
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, ... }: outputs = { nixpkgs, flake-utils, rust-overlay, crane, ... }:
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 {
@@ -20,22 +21,41 @@
targets = [ "wasm32-wasip2" ]; targets = [ "wasm32-wasip2" ];
extensions = [ "rust-src" ]; extensions = [ "rust-src" ];
}; };
rustPlatform = pkgs.makeRustPlatform { cargo = rust; rustc = rust; }; # The runtime cargo actor needs Cargo, rustc (with wasm32-wasip2), and
sausage = rustPlatform.buildRustPackage { # the native linker, even when the host itself is started with a bare
# 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; };
cargoBuildFlags = [ "-p" "sausage" ]; # The test packages for `checks.tests`.
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 : ${pkgs.lib.makeBinPath [ rust pkgs.stdenv.cc ]} wrapProgram $out/bin/sausage --prefix PATH : ${runtimePath}
''; '';
meta.mainProgram = "sausage"; meta.mainProgram = "sausage";
}; });
# `nix run .#loop -- <world>` runs the newest `sausage run <world>` # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
# (without a folder: bare `sausage`, i.e. ./world, created if needed)
# from the public repo, again and again: after a crash or a broker # from the public repo, again and again: after a crash or a broker
# `exit` request it refreshes the flake and starts the host again, # `exit` request it refreshes the flake and starts the host again,
# so the world updates itself. `/quit` (exit code 0) ends the loop. # so the world updates itself. `/quit` (exit code 0) ends the loop.
@@ -44,9 +64,10 @@
repoUrl = "git+https://gitea.surma.technology/surma/sausage"; repoUrl = "git+https://gitea.surma.technology/surma/sausage";
loop = pkgs.writeShellScriptBin "sausage-loop" '' loop = pkgs.writeShellScriptBin "sausage-loop" ''
url="''${SAUSAGE_FLAKE:-${repoUrl}}" url="''${SAUSAGE_FLAKE:-${repoUrl}}"
if [ $# -eq 0 ]; then set -- world; fi # No folder: bare `sausage` (creates ./world first if needed).
if [ $# -gt 0 ]; then set -- run "$@"; fi
while true; do while true; do
nix run --refresh "$url#default" -- run "$@" nix run --refresh "$url#default" -- "$@"
code=$? code=$?
if [ "$code" -eq 0 ]; then exit 0; fi if [ "$code" -eq 0 ]; then exit 0; fi
echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2 echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2
@@ -60,15 +81,82 @@
# `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.
checks.tests = sausage.overrideAttrs (old: { # The binary the host loop runs (`nix run`), built from the same
pname = "sausage-tests"; # cached dependencies as the tests.
doCheck = true; checks.package = sausage;
cargoTestFlags = [ "-p" "sausage" "-p" "sausage-actor" "-p" "agent" ]; checks.tests = craneLib.cargoTest (common // {
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"; };