Compare commits

...
Author SHA1 Message Date
agent-pool-8 6e07efaaa0 Borrow host descriptors while tracking VFS paths
ci / check (push) Successful in 13m25s
2026-10-16 21:09:59 +00:00
agent-pool-8 14f086423b Keep discovery test file at served root
ci / check (push) Failing after 6m54s
2026-10-16 20:59:09 +00:00
agent-pool-8 e32bfc57e2 Expose advertised actor-served files in directory listings 2026-10-16 20:59:09 +00:00
Sausage Agent d0876626df Public repo: .#loop fetches over public HTTPS, README for outsiders
ci / check (push) Successful in 8s
- flake.nix: .#loop runs git+https://gitea.surma.technology/surma/sausage
  (SAUSAGE_FLAKE overrides it); drop loop-internal / --internal (LAN SSH).
- README: clone step, LLM token requirement, loop URL and SAUSAGE_FLAKE.
- ci.yml: comment without internal host name.
2026-10-16 20:40:26 +00:00
6 changed files with 182 additions and 29 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
# Runs on the Nix runner on nexus (label `nixos`, host executor). # Runs on a self-hosted Nix runner (label `nixos`, host executor).
name: ci name: ci
on: [push, pull_request] on: [push, pull_request]
+7
View 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"}]}` |
+7 -4
View File
@@ -10,9 +10,11 @@ change its own code.
## Start ## Start
You need [Nix](https://nixos.org/download) with flakes. You need [Nix](https://nixos.org/download) with flakes, and an API token for
an LLM (OpenAI Responses or Anthropic Messages API).
```sh ```sh
git clone https://gitea.surma.technology/surma/sausage && cd sausage
nix run -- init # creates ./world and builds the agent (a few minutes) nix run -- init # creates ./world and builds the agent (a few minutes)
nix run -- run # starts the world and a console nix run -- run # starts the world and a console
``` ```
@@ -26,12 +28,13 @@ folder as an argument, for example `nix run -- run my-world`.
To keep a world on the newest version, run it in a loop: To keep a world on the newest version, run it in a loop:
```sh ```sh
nix run .#loop -- my-world # or .#loop-internal (Gitea on the LAN) nix run .#loop -- my-world
``` ```
The loop runs `nix run --refresh git+ssh://…/surma/sausage -- run my-world`. The loop runs `nix run --refresh git+https://gitea.surma.technology/surma/sausage -- run my-world`.
When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop When the host crashes or an actor sends `@broker {"type":"exit"}`, the loop
fetches the newest commit and starts it again. `/quit` ends the loop. fetches the newest commit and starts it again. `/quit` ends the loop. Set
`SAUSAGE_FLAKE` to run another flake URL instead, for example your fork.
In the console: In the console:
+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"] {
+65
View File
@@ -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;
+7 -9
View File
@@ -36,15 +36,14 @@
meta.mainProgram = "sausage"; meta.mainProgram = "sausage";
}; };
# `nix run .#loop -- <world>` runs the newest `sausage run <world>` # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
# from the Gitea repo, again and again: after a crash or a broker # from the public repo, again and again: after a crash or a broker
# `exit` request it refreshes the flake and starts the host again, # `exit` request it refreshes the flake and starts the host again,
# so the world updates itself. `/quit` (exit code 0) ends the loop. # so the world updates itself. `/quit` (exit code 0) ends the loop.
# `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN. # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage"; # e.g. a fork or a local checkout).
internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage"; repoUrl = "git+https://gitea.surma.technology/surma/sausage";
loop = name: url: pkgs.writeShellScriptBin name '' loop = pkgs.writeShellScriptBin "sausage-loop" ''
url="${url}" url="''${SAUSAGE_FLAKE:-${repoUrl}}"
if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
if [ $# -eq 0 ]; then set -- world; fi if [ $# -eq 0 ]; then set -- world; fi
while true; do while true; do
nix run --refresh "$url#default" -- run "$@" nix run --refresh "$url#default" -- run "$@"
@@ -72,8 +71,7 @@
}); });
apps.default = { type = "app"; program = "${sausage}/bin/sausage"; }; apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; }; apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
devShells.default = pkgs.mkShell { devShells.default = pkgs.mkShell {
packages = [ rust pkgs.wasmtime pkgs.wasm-tools ]; packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];