Borrow host descriptors while tracking VFS paths

- 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.

.gitea/workflows/ci.yml

@@ -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
 on: [push, pull_request]
 

PROTOCOL.md

@@ -68,7 +68,7 @@ Each instance sees three folders:
 | `/actor` | read-only | the `actor/` folder of the world |
 
 `/actor/<id>/` holds everything about an actor: `actor.wasm`, `actor.json`
-(installed version, running, subscriptions), `source/`, `public/`, `data/`,
+(installed version, running, subscriptions, tapped), `source/`, `public/`, `data/`,
 `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
@@ -78,12 +78,46 @@ so that readers never see half of it, write a temporary file and `rename` it
 Send large or binary data as a path under `/actor/<id>/public/`, not inside a
 message.
 
+### Actor-served files: `/actor/<id>/fs/`
+
+Files under `/actor/<id>/fs/` are not on disk: the actor `<id>` serves them.
+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`
+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 |
+| --- | --- |
+| `{"type": "fs_read", "path"}` | `{"type": "fs_file", "text"}` (or `"base64"` instead of `"text"`), or `{"type": "fs_dir", "entries": [{"name", "type": "file" \| "dir"}]}` |
+| `{"type": "fs_write", "path", "text"}` (or `"base64"` for non-UTF-8 data) | `{"type": "fs_done"}` |
+| `{"type": "fs_mkdir", "path"}`, `{"type": "fs_remove", "path"}` | `{"type": "fs_done"}` |
+| `{"type": "fs_rename", "path", "to"}` | `{"type": "fs_done"}` |
+
+Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
+`errno` is a `wasi:filesystem` error code: `no-entry`, `exist`,
+`not-directory`, `is-directory`, `not-empty`, `access`, `not-permitted`,
+`invalid`, `unsupported`, `busy`, or `io` (also for anything else).
+
+- Opening a file sends `fs_read` once. Reads come from that copy.
+- After a change, closing (or syncing) the file sends the whole file in one
+  `fs_write`.
+- 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/`
+  waits for itself and fails after the timeout.
+- `fs` is reserved: no actor can have that ID.
+
 ## The broker
 
 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`,
-`rolled_back`, `stopped`, `started`, `subscribed`, `unsubscribed`,
-`scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
+`rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
+`tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
 
 | `type` | Fields | Effect |
 | --- | --- | --- |
@@ -91,12 +125,32 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
 | `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
 | `rollback` | `actor`, `version` | go back to an older version |
 | `stop`, `start` | `actor` | stop or start an actor |
+| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
 | `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
+| `tap`, `untap` | | receive (or not) a copy of every message the broker routes |
 | `scan` | | boot new or changed `actor.wasm` files |
 | `exit` | | answer `exiting`, stop every actor, and end the `sausage` process (a wrapper like `nix run .#loop` starts it again, updated) |
 
 Paths are `/actor/...`, or the sender's `/public/...` or `/data/...`.
 
+**Taps.** After `tap`, the sender gets a copy of every message that the
+broker routes (direct messages, replies, and publishes, between any actors,
+including its own), for debugging, logging, or a live view of the world:
+
+```json
+{"type": "tapped", "message": {"id": "m-17", "from": "console", "to": "agent", "reply_to": null, "body": {...}}, "ts": 1792132579067}
+```
+
+- `to` is `null` for a publish: one copy per publish, not per subscriber.
+- `ts` is the time of routing, in milliseconds since the Unix epoch.
+- Copies come from `broker`. They are not routed themselves, so they are
+  never tapped, and each message is tapped once per tapper.
+- A copy that does not fit into the tapper's mailbox is dropped; routing
+  never waits for a tapper, and nobody gets `undeliverable` for a copy.
+- Never answer a copy: the answer would be tapped again, forever.
+- Like subscriptions, a tap is saved in `actor.json` (`"tapped": true`) and
+  survives restarts and upgrades. `untap` ends it.
+
 **Versions.** A new version (from `spawn`, `upgrade`, or a changed
 `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>/`

README.md

@@ -10,9 +10,11 @@ change its own code.
 
 ## 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
+git clone https://gitea.surma.technology/surma/sausage && cd sausage
 nix run -- init     # creates ./world and builds the agent (a few minutes)
 nix run -- run      # starts the world and a console
 ```
@@ -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:
 
 ```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
-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:
 

cli/src/broker.rs

@@ -1,6 +1,7 @@
-//! The broker: message routing, mailboxes, subscriptions, the version store,
-//! and the requests to `broker` (spawn, upgrade, rollback, stop, start,
-//! subscribe, unsubscribe, scan, exit). `supervisor.rs` runs each WASM actor.
+//! The broker: message routing, mailboxes, subscriptions, taps, the version
+//! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
+//! remove, subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs`
+//! runs each WASM actor.
 //!
 //! The broker accepts only `.wasm` files. It does not know any source language.
 //! The world folder is the source of truth: `scan` boots every
@@ -20,6 +21,7 @@ use wasmtime::component::Component;
 use crate::candidate;
 use crate::runtime::{InstanceConfig, Mount, Runtime};
 use crate::supervisor::{Control, Reply, Supervisor};
+use crate::vfs;
 use crate::world::{self, ActorMeta, NATIVE, VersionMeta, World};
 
 pub const MAILBOX_SIZE: usize = 1000;
@@ -60,12 +62,15 @@ pub struct Broker {
     seq: AtomicU64,
     /// Set by the `exit` request: the host process should stop.
     exit: tokio::sync::Notify,
+    /// Requests of the host from `fs` (see `vfs.rs`) that wait for a reply, by message ID.
+    asks: Mutex<HashMap<String, oneshot::Sender<Value>>>,
+    me: std::sync::Weak<Broker>,
 }
 
 impl Broker {
     pub fn new(world: World, runtime: Arc<Runtime>) -> Result<Arc<Self>> {
         let metas = world.actor_ids()?.into_iter().map(|id| (id.clone(), world.read_meta(&id))).collect();
-        Ok(Arc::new(Self {
+        Ok(Arc::new_cyclic(|me| Self {
             world,
             runtime,
             mailboxes: Mutex::new(HashMap::new()),
@@ -74,9 +79,27 @@ impl Broker {
             op_locks: Mutex::new(HashMap::new()),
             seq: AtomicU64::new(1),
             exit: tokio::sync::Notify::new(),
+            asks: Mutex::new(HashMap::new()),
+            me: me.clone(),
         }))
     }
 
+    /// Send `body` from `fs` to `to` now. The future resolves to the reply
+    /// body, or `None` after `vfs::ASK_TIMEOUT`.
+    pub fn ask(self: &Arc<Self>, to: &str, body: Value) -> impl Future<Output = Option<Value>> + Send + 'static {
+        let own = format!("q{}", self.seq.fetch_add(1, Ordering::Relaxed));
+        let id = format!("{}/{own}", vfs::ID);
+        let (tx, rx) = oneshot::channel();
+        self.asks.lock().unwrap().insert(id.clone(), tx);
+        self.send(vfs::ID, Some(to), Some(&own), None, body);
+        let this = self.clone();
+        async move {
+            let reply = tokio::time::timeout(vfs::ASK_TIMEOUT, rx).await;
+            this.asks.lock().unwrap().remove(&id);
+            reply.ok()?.ok()
+        }
+    }
+
     /// Wait until an actor sends `{"type": "exit"}` to `broker`. The `sausage`
     /// binary then stops all actors and exits, so that a wrapper such as
     /// `nix run .#loop` can start a fresh (updated) host.
@@ -111,17 +134,23 @@ impl Broker {
             Some(own) => format!("{from}/{own}"),
             None => format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
         };
+        let (recipients, tappers) = {
+            let metas = self.metas.lock().unwrap();
             let recipients: Vec<String> = match to {
                 Some(to) => vec![to.to_string()],
-            None => self
-                .metas
-                .lock()
-                .unwrap()
+                None => metas
                     .iter()
                     .filter(|(_, m)| m.subscriptions.contains(from))
                     .map(|(id, _)| id.clone())
                     .collect(),
             };
+            let tappers: Vec<String> = metas.iter().filter(|(_, m)| m.tapped).map(|(id, _)| id.clone()).collect();
+            (recipients, tappers)
+        };
+        if !tappers.is_empty() {
+            let message = json!({"id": id, "from": from, "to": to, "reply_to": reply_to, "body": body});
+            self.tap(&tappers, message);
+        }
         for to in recipients {
             self.deliver(Envelope {
                 id: id.clone(),
@@ -134,6 +163,25 @@ impl Broker {
         id
     }
 
+    /// Give every tapper a copy of one routed message. Copies skip `send`, so
+    /// they are never tapped themselves; a full mailbox drops the copy.
+    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 body = json!({"type": "tapped", "message": message, "ts": ts});
+        let mailboxes = self.mailboxes.lock().unwrap();
+        for tapper in tappers {
+            if let Some(tx) = mailboxes.get(tapper) {
+                let _ = tx.try_send(Envelope {
+                    id: format!("m-{}", self.seq.fetch_add(1, Ordering::Relaxed)),
+                    from: "broker".into(),
+                    to: tapper.clone(),
+                    reply_to: None,
+                    body: body.clone(),
+                });
+            }
+        }
+    }
+
     /// Route one stdout line of a WASM actor.
     pub fn route_line(&self, from: &str, line: &str) {
         let line = line.trim();
@@ -155,6 +203,13 @@ impl Broker {
     }
 
     fn deliver(&self, env: Envelope) {
+        if env.to == vfs::ID {
+            let waiting = env.reply_to.as_ref().and_then(|r| self.asks.lock().unwrap().remove(r));
+            if let Some(tx) = waiting {
+                let _ = tx.send(env.body);
+            }
+            return;
+        }
         let tx = self.mailboxes.lock().unwrap().get(&env.to).cloned();
         let error = match tx {
             None => format!("no actor with ID {}", env.to),
@@ -201,6 +256,8 @@ impl Broker {
             network: scratch.is_none(),
             stderr_log: own.join("stderr.log"),
             env,
+            actor: id.to_string(),
+            broker: self.me.clone(),
         }
     }
 
@@ -320,6 +377,35 @@ impl Broker {
         Ok(result)
     }
 
+    /// Stop an actor (normal shutdown) and delete its folder `actor/<id>/`.
+    /// Afterwards the ID is unknown: messages to it are undeliverable, and
+    /// it can be spawned again from scratch.
+    pub async fn remove_actor(&self, id: &str) -> Result<Value> {
+        if NATIVE.contains(&id) {
+            bail!("{id} is part of the host and cannot be removed");
+        }
+        if !world::valid_id(id) {
+            bail!("invalid actor ID {id:?}");
+        }
+        let lock = self.op_lock(id);
+        let _guard = lock.lock().await;
+        let dir = self.world.actor_dir(id);
+        if !self.is_supervised(id) && !dir.is_dir() {
+            bail!("no actor with ID {id}");
+        }
+        if self.is_supervised(id) {
+            self.control(id, |reply| Control::Shutdown { reply }).await?;
+        }
+        // Without its mailbox, messages to `id` are undeliverable.
+        self.controls.lock().unwrap().remove(id);
+        self.mailboxes.lock().unwrap().remove(id);
+        self.metas.lock().unwrap().remove(id);
+        if dir.exists() {
+            std::fs::remove_dir_all(&dir).with_context(|| format!("delete {}", dir.display()))?;
+        }
+        Ok(json!({"type": "removed", "actor": id}))
+    }
+
     /// Boot the actors in `actor/`. Called at start and on `scan`.
     ///
     /// - An `actor.wasm` equal to the installed version starts as is.
@@ -424,6 +510,7 @@ impl Broker {
             }
             "stop" => self.control(&actor()?, |reply| Control::Stop { reply }).await,
             "start" => self.control(&actor()?, |reply| Control::Start { reply }).await,
+            "remove" => self.remove_actor(&actor()?).await,
             "subscribe" | "unsubscribe" => {
                 let publisher = actor()?;
                 let on = ty == "subscribe";
@@ -436,6 +523,12 @@ impl Broker {
                 })?;
                 Ok(json!({"type": format!("{ty}d"), "actor": publisher}))
             }
+            "tap" | "untap" => {
+                let on = ty == "tap";
+                self.update_meta(&env.from, |m| m.tapped = on)?;
+                let answer = if on { "tapping" } else { "untapped" };
+                Ok(json!({"type": answer}))
+            }
             "scan" => Ok(json!({"type": "scanned", "changes": self.scan().await})),
             "exit" => {
                 self.log("broker", &format!("exit requested by {}", env.from));
@@ -443,7 +536,7 @@ impl Broker {
                 self.exit.notify_one();
                 Ok(json!({"type": "exiting"}))
             }
-            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, subscribe, unsubscribe, scan, exit"),
+            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
         }
     }
 

cli/src/lib.rs

@@ -9,6 +9,7 @@ pub mod cargo_actor;
 pub mod console;
 pub mod runtime;
 pub mod supervisor;
+pub mod vfs;
 pub mod world;
 
 use std::path::Path;

cli/src/runtime.rs

@@ -4,7 +4,7 @@
 //! NDJSON lines to its stdin and reads NDJSON lines from its stdout.
 
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 use std::time::Duration;
 
 use anyhow::{Context, Result};
@@ -14,10 +14,15 @@ use tokio::task::JoinHandle;
 use wasmtime::component::{Component, Linker, ResourceTable};
 use wasmtime::{Config, Engine, Store};
 use wasmtime_wasi::cli::{AsyncStdinStream, AsyncStdoutStream};
+use wasmtime_wasi::filesystem::WasiFilesystemCtxView;
 use wasmtime_wasi::p2::bindings::Command;
+use wasmtime_wasi::p2::bindings::filesystem;
 use wasmtime_wasi::{FsPerms, I32Exit, WasiCtx, WasiCtxView, WasiView};
 use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpHooks, WasiHttpView};
 
+use crate::broker::Broker;
+use crate::vfs::{Vfs, VfsState, VfsView};
+
 /// A host folder that the guest sees at `guest`.
 #[derive(Clone, Debug)]
 pub struct Mount {
@@ -65,6 +70,11 @@ pub struct ActorState {
     table: ResourceTable,
     http: WasiHttpCtx,
     hooks: HttpHooks,
+    vfs: VfsState,
+}
+
+fn vfs_view(s: &mut ActorState) -> VfsView<'_> {
+    VfsView { inner: WasiFilesystemCtxView { ctx: s.ctx.filesystem(), table: &mut s.table }, state: &mut s.vfs }
 }
 
 impl WasiView for ActorState {
@@ -111,6 +121,10 @@ pub struct InstanceConfig {
     pub network: bool,
     pub stderr_log: PathBuf,
     pub env: Vec<(String, String)>,
+    /// The actor that runs in the instance.
+    pub actor: String,
+    /// Asks the actors that serve `/actor/<id>/fs/` (see `vfs.rs`).
+    pub broker: Weak<Broker>,
 }
 
 /// A running guest. Dropping it kills the guest.
@@ -224,6 +238,11 @@ impl Runtime {
         let mut linker: Linker<ActorState> = Linker::new(&engine);
         wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
         wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?;
+        // Replace `wasi:filesystem` with the wrapper that serves `/actor/<id>/fs/`.
+        linker.allow_shadowing(true);
+        filesystem::types::add_to_linker::<ActorState, Vfs>(&mut linker, vfs_view)?;
+        filesystem::preopens::add_to_linker::<ActorState, Vfs>(&mut linker, vfs_view)?;
+        linker.allow_shadowing(false);
 
         // Time slices: every 10 ms each running guest yields to the scheduler.
         let ticker = engine.clone();
@@ -274,6 +293,7 @@ impl Runtime {
             table: ResourceTable::new(),
             http: WasiHttpCtx::new(),
             hooks: HttpHooks { allow: cfg.network },
+            vfs: VfsState::new(cfg.broker.clone(), cfg.actor.clone()),
         };
 
         let (line_tx, line_rx) = mpsc::channel(1024);

cli/src/vfs.rs

@@ -0,0 +1,796 @@
+//! Actor-served files: `/actor/<id>/fs/...` (see PROTOCOL.md, "Actor-served
+//! files").
+//!
+//! 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
+//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
+//! the reply. Other paths go to `wasmtime-wasi` unchanged except that an
+//! actor directory listing advertises `fs/` when its source docs declare it.
+//!
+//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
+//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
+//!   `fs_rename {path, to}` -> `fs_done`
+//! - any of them -> `fs_error {errno}` (a `wasi:filesystem` error-code name)
+//!
+//! 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.
+
+use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+
+use bytes::Bytes;
+use serde_json::{Value, json};
+use wasmtime::component::{HasData, Resource};
+use wasmtime_wasi::filesystem::WasiFilesystemCtxView;
+use wasmtime_wasi::p2::bindings::filesystem::types::{
+    self, DescriptorFlags, DescriptorType, DirectoryEntry, ErrorCode, HostDescriptor, HostDirectoryEntryStream, OpenFlags,
+};
+use wasmtime_wasi::p2::bindings::filesystem::preopens;
+use wasmtime_wasi::p2::pipe::MemoryInputPipe;
+use wasmtime_wasi::p2::{DynInputStream, DynOutputStream, FsError, FsResult, OutputStream, Pollable, StreamResult};
+
+use crate::broker::Broker;
+
+/// The `from` of the host's requests. Replies to it go back to the waiting call.
+pub const ID: &str = "fs";
+pub const ASK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+/// Per-instance state.
+pub struct VfsState {
+    broker: Weak<Broker>,
+    /// The actor of this instance, for notes in its `stderr.log`.
+    me: String,
+    /// Native directory descriptors and their absolute paths. Keeping every
+    /// `/actor` child (not only the preopen) lets `fs/` work after traversal.
+    dirs: HashMap<u32, String>,
+}
+
+impl VfsState {
+    pub fn new(broker: Weak<Broker>, me: String) -> Self {
+        Self { broker, me, dirs: HashMap::new() }
+    }
+}
+
+/// [`HasData`] for the linker: the filesystem functions get a [`VfsView`].
+pub struct Vfs;
+
+impl HasData for Vfs {
+    type Data<'a> = VfsView<'a>;
+}
+
+pub struct VfsView<'a> {
+    pub inner: WasiFilesystemCtxView<'a>,
+    pub state: &'a mut VfsState,
+}
+
+/// A descriptor of an actor-served file or folder (in the resource table in
+/// place of a `wasmtime-wasi` descriptor).
+#[derive(Clone)]
+enum Virt {
+    Dir { actor: String, path: String },
+    File(VFile),
+}
+
+#[derive(Clone)]
+struct VFile {
+    actor: String,
+    path: String,
+    data: Arc<Mutex<Vec<u8>>>,
+    dirty: Arc<AtomicBool>,
+    write: bool,
+}
+
+/// A directory-entry-stream of an actor-served folder.
+struct VDir(std::vec::IntoIter<DirectoryEntry>);
+
+enum Node {
+    File(Vec<u8>),
+    Dir(Vec<DirectoryEntry>),
+}
+
+/// Sends the requests. Owned, so that no borrow of the view lives across an await.
+struct Asker {
+    broker: Option<Arc<Broker>>,
+    me: String,
+}
+
+impl Asker {
+    async fn ask(&self, actor: &str, body: Value) -> Result<Value, ErrorCode> {
+        let broker = self.broker.as_ref().ok_or(ErrorCode::Io)?;
+        let what = format!("{} {} to {actor}", body["type"].as_str().unwrap_or(""), body["path"].as_str().unwrap_or(""));
+        match broker.ask(actor, body).await {
+            None => {
+                broker.log(&self.me, &format!("{what}: no reply within {} s", ASK_TIMEOUT.as_secs()));
+                Err(ErrorCode::Io)
+            }
+            Some(reply) => match reply["type"].as_str() {
+                Some("fs_error") => Err(errno(&reply["errno"])),
+                Some("undeliverable") => Err(ErrorCode::NoEntry),
+                _ => Ok(reply),
+            },
+        }
+    }
+
+    async fn read(&self, actor: &str, path: &str) -> Result<Node, ErrorCode> {
+        let reply = self.ask(actor, json!({"type": "fs_read", "path": path})).await?;
+        match reply["type"].as_str() {
+            Some("fs_file") => content(&reply).map(Node::File),
+            Some("fs_dir") => Ok(Node::Dir(
+                reply["entries"]
+                    .as_array()
+                    .into_iter()
+                    .flatten()
+                    .filter_map(|e| {
+                        let type_ = if e["type"] == "dir" { DescriptorType::Directory } else { DescriptorType::RegularFile };
+                        Some(DirectoryEntry { type_, name: e["name"].as_str()?.to_string() })
+                    })
+                    .collect(),
+            )),
+            _ => {
+                if let Some(b) = &self.broker {
+                    b.log(&self.me, &format!("fs_read {path} to {actor}: unexpected reply {reply}"));
+                }
+                Err(ErrorCode::Io)
+            }
+        }
+    }
+
+    async fn change(&self, actor: &str, body: Value) -> FsResult<()> {
+        self.ask(actor, body).await.map(|_| ()).map_err(FsError::from)
+    }
+
+    async fn flush(&self, f: &VFile) -> FsResult<()> {
+        if f.dirty.swap(false, Ordering::SeqCst)
+            && let Err(e) = self.ask(&f.actor, write_body(f)).await
+        {
+            f.dirty.store(true, Ordering::SeqCst);
+            return Err(e.into());
+        }
+        Ok(())
+    }
+}
+
+fn errno(v: &Value) -> ErrorCode {
+    match v.as_str().unwrap_or("") {
+        "no-entry" => ErrorCode::NoEntry,
+        "exist" => ErrorCode::Exist,
+        "not-directory" => ErrorCode::NotDirectory,
+        "is-directory" => ErrorCode::IsDirectory,
+        "not-empty" => ErrorCode::NotEmpty,
+        "access" => ErrorCode::Access,
+        "not-permitted" => ErrorCode::NotPermitted,
+        "invalid" => ErrorCode::Invalid,
+        "unsupported" => ErrorCode::Unsupported,
+        "busy" => ErrorCode::Busy,
+        _ => ErrorCode::Io,
+    }
+}
+
+fn content(v: &Value) -> Result<Vec<u8>, ErrorCode> {
+    if let Some(text) = v["text"].as_str() {
+        return Ok(text.as_bytes().to_vec());
+    }
+    match v["base64"].as_str() {
+        Some(b) => base64_decode(b).ok_or(ErrorCode::IllegalByteSequence),
+        None => Ok(Vec::new()),
+    }
+}
+
+fn write_body(f: &VFile) -> Value {
+    let data = f.data.lock().unwrap().clone();
+    match String::from_utf8(data) {
+        Ok(text) => json!({"type": "fs_write", "path": f.path, "text": text}),
+        Err(e) => json!({"type": "fs_write", "path": f.path, "base64": base64_encode(e.as_bytes())}),
+    }
+}
+
+/// `rel` resolved against the folder `base`, as an absolute path.
+fn join(base: &str, rel: &str) -> String {
+    let mut parts: Vec<&str> = base.split('/').filter(|p| !p.is_empty()).collect();
+    for p in rel.split('/') {
+        match p {
+            "" | "." => {}
+            ".." => {
+                parts.pop();
+            }
+            p => parts.push(p),
+        }
+    }
+    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 {
+    types::DescriptorStat {
+        type_,
+        link_count: 1,
+        size,
+        data_access_timestamp: None,
+        data_modification_timestamp: None,
+        status_change_timestamp: None,
+    }
+}
+
+fn node_stat(node: &Node) -> types::DescriptorStat {
+    match node {
+        Node::File(data) => stat(DescriptorType::RegularFile, data.len() as u64),
+        Node::Dir(_) => stat(DescriptorType::Directory, 0),
+    }
+}
+
+fn hash(actor: &str, path: &str) -> types::MetadataHashValue {
+    let mut h = std::collections::hash_map::DefaultHasher::new();
+    (actor, path).hash(&mut h);
+    types::MetadataHashValue { lower: h.finish(), upper: 0 }
+}
+
+fn write_at(data: &mut Vec<u8>, at: usize, bytes: &[u8]) {
+    if data.len() < at + bytes.len() {
+        data.resize(at + bytes.len(), 0);
+    }
+    data[at..at + bytes.len()].copy_from_slice(bytes);
+}
+
+/// An output stream into an actor-served file (`pos: None` appends).
+struct VWriter {
+    file: VFile,
+    pos: Option<usize>,
+}
+
+#[wasmtime_wasi::async_trait]
+impl Pollable for VWriter {
+    async fn ready(&mut self) {}
+}
+
+#[wasmtime_wasi::async_trait]
+impl OutputStream for VWriter {
+    fn write(&mut self, bytes: Bytes) -> StreamResult<()> {
+        let mut data = self.file.data.lock().unwrap();
+        let at = self.pos.unwrap_or(data.len());
+        write_at(&mut data, at, &bytes);
+        if let Some(p) = &mut self.pos {
+            *p += bytes.len();
+        }
+        self.file.dirty.store(true, Ordering::SeqCst);
+        Ok(())
+    }
+    fn flush(&mut self) -> StreamResult<()> {
+        Ok(())
+    }
+    fn check_write(&mut self) -> StreamResult<usize> {
+        Ok(64 * 1024)
+    }
+}
+
+impl VfsView<'_> {
+    fn asker(&self) -> Asker {
+        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> {
+        self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
+    }
+
+    fn file(&mut self, fd: &Resource<types::Descriptor>) -> Option<FsResult<VFile>> {
+        match self.virt(fd)? {
+            Virt::File(f) => Some(Ok(f)),
+            Virt::Dir { .. } => Some(Err(ErrorCode::IsDirectory.into())),
+        }
+    }
+
+    /// The actor and its path, if `path` relative to `fd` is actor-served.
+    fn target(&mut self, fd: &Resource<types::Descriptor>, path: &str) -> Option<(String, String)> {
+        match self.virt(fd) {
+            Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
+            Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
+            None => {
+                let base = self.state.dirs.get(&fd.rep())?;
+                actor_fs_path(&join(base, path))
+            }
+        }
+    }
+
+    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>> {
+        let r = self.inner.table.push(v)?;
+        Ok(Resource::new_own(r.rep()))
+    }
+
+    async fn open(&mut self, actor: String, path: String, oflags: OpenFlags, flags: DescriptorFlags) -> FsResult<Resource<types::Descriptor>> {
+        let write = flags.contains(DescriptorFlags::WRITE);
+        let create = oflags.contains(OpenFlags::CREATE);
+        let exclusive = create && oflags.contains(OpenFlags::EXCLUSIVE);
+        let virt = match self.asker().read(&actor, &path).await {
+            Ok(_) if exclusive => return Err(ErrorCode::Exist.into()),
+            Ok(Node::Dir(_)) if write => return Err(ErrorCode::IsDirectory.into()),
+            Ok(Node::Dir(_)) => Virt::Dir { actor, path },
+            Ok(Node::File(_)) if oflags.contains(OpenFlags::DIRECTORY) => return Err(ErrorCode::NotDirectory.into()),
+            Ok(Node::File(data)) => {
+                let truncate = oflags.contains(OpenFlags::TRUNCATE);
+                let data = if truncate { Vec::new() } else { data };
+                Virt::File(VFile { actor, path, data: Arc::new(Mutex::new(data)), dirty: Arc::new(AtomicBool::new(truncate)), write })
+            }
+            Err(ErrorCode::NoEntry) if create && !oflags.contains(OpenFlags::DIRECTORY) => Virt::File(VFile {
+                actor,
+                path,
+                data: Arc::default(),
+                dirty: Arc::new(AtomicBool::new(true)),
+                write,
+            }),
+            Err(e) => return Err(e.into()),
+        };
+        self.push(virt)
+    }
+
+    /// Send the changed file without waiting (the descriptor is gone). The
+    /// request leaves now, so later requests of this instance come after it.
+    fn write_back_later(&self, f: &VFile) {
+        let Some(broker) = self.state.broker.upgrade() else { return };
+        let reply = broker.ask(&f.actor, write_body(f));
+        let (me, what) = (self.state.me.clone(), format!("fs_write {} to {}", f.path, f.actor));
+        tokio::spawn(async move {
+            match reply.await {
+                Some(r) if r["type"] != "fs_error" && r["type"] != "undeliverable" => {}
+                r => broker.log(&me, &format!("{what} failed: {}", r.map_or("no reply".to_string(), |r| r.to_string()))),
+            }
+        });
+    }
+}
+
+impl preopens::Host for VfsView<'_> {
+    fn get_directories(&mut self) -> wasmtime::Result<Vec<(Resource<types::Descriptor>, String)>> {
+        let dirs = preopens::Host::get_directories(&mut self.inner)?;
+        for (fd, name) in &dirs {
+            if name == "/actor" {
+                self.state.dirs.insert(fd.rep(), "/actor".to_string());
+            }
+        }
+        Ok(dirs)
+    }
+}
+
+impl types::Host for VfsView<'_> {
+    fn convert_error_code(&mut self, err: FsError) -> wasmtime::Result<ErrorCode> {
+        types::Host::convert_error_code(&mut self.inner, err)
+    }
+
+    fn filesystem_error_code(&mut self, err: Resource<wasmtime::Error>) -> wasmtime::Result<Option<ErrorCode>> {
+        types::Host::filesystem_error_code(&mut self.inner, err)
+    }
+}
+
+impl HostDescriptor for VfsView<'_> {
+    async fn advise(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize, len: types::Filesize, advice: types::Advice) -> FsResult<()> {
+        if self.virt(&fd).is_some() {
+            return Ok(());
+        }
+        HostDescriptor::advise(&mut self.inner, fd, offset, len, advice).await
+    }
+
+    async fn sync_data(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
+        match self.virt(&fd) {
+            Some(Virt::File(f)) => self.asker().flush(&f).await,
+            Some(Virt::Dir { .. }) => Ok(()),
+            None => HostDescriptor::sync_data(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn get_flags(&mut self, fd: Resource<types::Descriptor>) -> FsResult<DescriptorFlags> {
+        match self.virt(&fd) {
+            Some(Virt::File(f)) if f.write => Ok(DescriptorFlags::READ | DescriptorFlags::WRITE),
+            Some(Virt::File(_)) => Ok(DescriptorFlags::READ),
+            Some(Virt::Dir { .. }) => Ok(DescriptorFlags::READ | DescriptorFlags::MUTATE_DIRECTORY),
+            None => HostDescriptor::get_flags(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn get_type(&mut self, fd: Resource<types::Descriptor>) -> FsResult<DescriptorType> {
+        match self.virt(&fd) {
+            Some(Virt::File(_)) => Ok(DescriptorType::RegularFile),
+            Some(Virt::Dir { .. }) => Ok(DescriptorType::Directory),
+            None => HostDescriptor::get_type(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn set_size(&mut self, fd: Resource<types::Descriptor>, size: types::Filesize) -> FsResult<()> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                if !f.write {
+                    return Err(ErrorCode::NotPermitted.into());
+                }
+                f.data.lock().unwrap().resize(size as usize, 0);
+                f.dirty.store(true, Ordering::SeqCst);
+                Ok(())
+            }
+            None => HostDescriptor::set_size(&mut self.inner, fd, size).await,
+        }
+    }
+
+    async fn set_times(&mut self, fd: Resource<types::Descriptor>, atim: types::NewTimestamp, mtim: types::NewTimestamp) -> FsResult<()> {
+        if self.virt(&fd).is_some() {
+            return Ok(());
+        }
+        HostDescriptor::set_times(&mut self.inner, fd, atim, mtim).await
+    }
+
+    async fn read(&mut self, fd: Resource<types::Descriptor>, len: types::Filesize, offset: types::Filesize) -> FsResult<(Vec<u8>, bool)> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                let data = f.data.lock().unwrap();
+                let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len());
+                let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(data.len());
+                Ok((data[start..end].to_vec(), start == end))
+            }
+            None => HostDescriptor::read(&mut self.inner, fd, len, offset).await,
+        }
+    }
+
+    async fn write(&mut self, fd: Resource<types::Descriptor>, buf: Vec<u8>, offset: types::Filesize) -> FsResult<types::Filesize> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                if !f.write {
+                    return Err(ErrorCode::NotPermitted.into());
+                }
+                write_at(&mut f.data.lock().unwrap(), offset as usize, &buf);
+                f.dirty.store(true, Ordering::SeqCst);
+                Ok(buf.len() as u64)
+            }
+            None => HostDescriptor::write(&mut self.inner, fd, buf, offset).await,
+        }
+    }
+
+    async fn read_directory(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Resource<types::DirectoryEntryStream>> {
+        match self.virt(&fd) {
+            Some(Virt::Dir { actor, path }) => match self.asker().read(&actor, &path).await? {
+                Node::Dir(entries) => {
+                    let r = self.inner.table.push(VDir(entries.into_iter()))?;
+                    Ok(Resource::new_own(r.rep()))
+                }
+                Node::File(_) => Err(ErrorCode::NotDirectory.into()),
+            },
+            Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
+            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
+                }
+            }
+        }
+    }
+
+    async fn sync(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
+        match self.virt(&fd) {
+            Some(Virt::File(f)) => self.asker().flush(&f).await,
+            Some(Virt::Dir { .. }) => Ok(()),
+            None => HostDescriptor::sync(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn create_directory_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_mkdir", "path": path})).await,
+            None => HostDescriptor::create_directory_at(&mut self.inner, fd, path).await,
+        }
+    }
+
+    async fn stat(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::DescriptorStat> {
+        match self.virt(&fd) {
+            Some(Virt::File(f)) => Ok(stat(DescriptorType::RegularFile, f.data.lock().unwrap().len() as u64)),
+            Some(Virt::Dir { .. }) => Ok(stat(DescriptorType::Directory, 0)),
+            None => HostDescriptor::stat(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn stat_at(&mut self, fd: Resource<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::DescriptorStat> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => Ok(node_stat(&self.asker().read(&actor, &path).await?)),
+            None => HostDescriptor::stat_at(&mut self.inner, fd, path_flags, path).await,
+        }
+    }
+
+    async fn set_times_at(
+        &mut self,
+        fd: Resource<types::Descriptor>,
+        path_flags: types::PathFlags,
+        path: String,
+        atim: types::NewTimestamp,
+        mtim: types::NewTimestamp,
+    ) -> FsResult<()> {
+        if self.target(&fd, &path).is_some() {
+            return Ok(());
+        }
+        HostDescriptor::set_times_at(&mut self.inner, fd, path_flags, path, atim, mtim).await
+    }
+
+    async fn link_at(
+        &mut self,
+        fd: Resource<types::Descriptor>,
+        old_path_flags: types::PathFlags,
+        old_path: String,
+        new_descriptor: Resource<types::Descriptor>,
+        new_path: String,
+    ) -> FsResult<()> {
+        if self.target(&fd, &old_path).is_some() || self.target(&new_descriptor, &new_path).is_some() {
+            return Err(ErrorCode::Unsupported.into());
+        }
+        HostDescriptor::link_at(&mut self.inner, fd, old_path_flags, old_path, new_descriptor, new_path).await
+    }
+
+    async fn open_at(
+        &mut self,
+        fd: Resource<types::Descriptor>,
+        path_flags: types::PathFlags,
+        path: String,
+        oflags: OpenFlags,
+        flags: DescriptorFlags,
+    ) -> FsResult<Resource<types::Descriptor>> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => self.open(actor, 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)
+            }
+        }
+    }
+
+    fn drop(&mut self, fd: Resource<types::Descriptor>) -> wasmtime::Result<()> {
+        if self.virt(&fd).is_some() {
+            if let Virt::File(f) = self.inner.table.delete(Resource::<Virt>::new_own(fd.rep()))?
+                && f.dirty.load(Ordering::SeqCst)
+            {
+                self.write_back_later(&f);
+            }
+            return Ok(());
+        }
+        self.state.dirs.remove(&fd.rep());
+        HostDescriptor::drop(&mut self.inner, fd)
+    }
+
+    async fn readlink_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<String> {
+        if self.target(&fd, &path).is_some() {
+            return Err(ErrorCode::Invalid.into());
+        }
+        HostDescriptor::readlink_at(&mut self.inner, fd, path).await
+    }
+
+    async fn remove_directory_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await,
+            None => HostDescriptor::remove_directory_at(&mut self.inner, fd, path).await,
+        }
+    }
+
+    async fn rename_at(&mut self, fd: Resource<types::Descriptor>, old_path: String, new_fd: Resource<types::Descriptor>, new_path: String) -> FsResult<()> {
+        match (self.target(&fd, &old_path), self.target(&new_fd, &new_path)) {
+            (None, None) => HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await,
+            (Some((a, from)), Some((b, to))) if a == b => {
+                self.asker().change(&a, json!({"type": "fs_rename", "path": from, "to": to})).await
+            }
+            _ => Err(ErrorCode::Unsupported.into()),
+        }
+    }
+
+    async fn symlink_at(&mut self, fd: Resource<types::Descriptor>, src_path: String, dest_path: String) -> FsResult<()> {
+        if self.target(&fd, &dest_path).is_some() {
+            return Err(ErrorCode::Unsupported.into());
+        }
+        HostDescriptor::symlink_at(&mut self.inner, fd, src_path, dest_path).await
+    }
+
+    async fn unlink_file_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await,
+            None => HostDescriptor::unlink_file_at(&mut self.inner, fd, path).await,
+        }
+    }
+
+    fn read_via_stream(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynInputStream>> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                let data = f.data.lock().unwrap();
+                let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len());
+                let stream: DynInputStream = Box::new(MemoryInputPipe::new(data[start..].to_vec()));
+                std::mem::drop(data);
+                Ok(self.inner.table.push(stream)?)
+            }
+            None => HostDescriptor::read_via_stream(&mut self.inner, fd, offset),
+        }
+    }
+
+    fn write_via_stream(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynOutputStream>> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                if !f.write {
+                    return Err(ErrorCode::NotPermitted.into());
+                }
+                let stream: DynOutputStream = Box::new(VWriter { file: f, pos: Some(offset as usize) });
+                Ok(self.inner.table.push(stream)?)
+            }
+            None => HostDescriptor::write_via_stream(&mut self.inner, fd, offset),
+        }
+    }
+
+    fn append_via_stream(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Resource<DynOutputStream>> {
+        match self.file(&fd) {
+            Some(f) => {
+                let f = f?;
+                if !f.write {
+                    return Err(ErrorCode::NotPermitted.into());
+                }
+                let stream: DynOutputStream = Box::new(VWriter { file: f, pos: None });
+                Ok(self.inner.table.push(stream)?)
+            }
+            None => HostDescriptor::append_via_stream(&mut self.inner, fd),
+        }
+    }
+
+    async fn is_same_object(&mut self, a: Resource<types::Descriptor>, b: Resource<types::Descriptor>) -> wasmtime::Result<bool> {
+        let key = |v: Virt| match v {
+            Virt::Dir { actor, path } => (actor, path),
+            Virt::File(f) => (f.actor, f.path),
+        };
+        match (self.virt(&a), self.virt(&b)) {
+            (None, None) => HostDescriptor::is_same_object(&mut self.inner, a, b).await,
+            (Some(x), Some(y)) => Ok(key(x) == key(y)),
+            _ => Ok(false),
+        }
+    }
+
+    async fn metadata_hash(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::MetadataHashValue> {
+        match self.virt(&fd) {
+            Some(Virt::Dir { actor, path }) => Ok(hash(&actor, &path)),
+            Some(Virt::File(f)) => Ok(hash(&f.actor, &f.path)),
+            None => HostDescriptor::metadata_hash(&mut self.inner, fd).await,
+        }
+    }
+
+    async fn metadata_hash_at(&mut self, fd: Resource<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::MetadataHashValue> {
+        match self.target(&fd, &path) {
+            Some((actor, path)) => Ok(hash(&actor, &path)),
+            None => HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await,
+        }
+    }
+}
+
+impl HostDirectoryEntryStream for VfsView<'_> {
+    async fn read_directory_entry(&mut self, stream: Resource<types::DirectoryEntryStream>) -> FsResult<Option<DirectoryEntry>> {
+        if let Some(dir) = self.inner.table.get_any_mut(stream.rep()).ok().and_then(|a| a.downcast_mut::<VDir>()) {
+            return Ok(dir.0.next());
+        }
+        HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream).await
+    }
+
+    fn drop(&mut self, stream: Resource<types::DirectoryEntryStream>) -> wasmtime::Result<()> {
+        let ours = self.inner.table.get_any_mut(stream.rep()).ok().is_some_and(|a| a.is::<VDir>());
+        if ours {
+            self.inner.table.delete(Resource::<VDir>::new_own(stream.rep()))?;
+            return Ok(());
+        }
+        HostDirectoryEntryStream::drop(&mut self.inner, stream)
+    }
+}
+
+const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+
+fn base64_encode(bytes: &[u8]) -> String {
+    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
+    for chunk in bytes.chunks(3) {
+        let n = chunk.iter().enumerate().fold(0u32, |n, (i, b)| n | (*b as u32) << (16 - 8 * i));
+        for i in 0..4 {
+            out.push(if i <= chunk.len() { B64[(n >> (18 - 6 * i) & 63) as usize] as char } else { '=' });
+        }
+    }
+    out
+}
+
+fn base64_decode(text: &str) -> Option<Vec<u8>> {
+    let (mut out, mut n, mut bits) = (Vec::new(), 0u32, 0);
+    for c in text.bytes().filter(|c| !c.is_ascii_whitespace() && *c != b'=') {
+        n = (n << 6 | B64.iter().position(|b| *b == c)? as u32) & 0xFF_FFFF;
+        bits += 6;
+        if bits >= 8 {
+            bits -= 8;
+            out.push((n >> bits) as u8);
+        }
+    }
+    Some(out)
+}
+
+#[cfg(test)]
+mod tests {
+    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]
+    fn base64_round_trip() {
+        for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
+            let enc = base64_encode(s.as_bytes());
+            assert_eq!(base64_decode(&enc).unwrap(), s.as_bytes());
+        }
+        assert_eq!(base64_encode(b"hello"), "aGVsbG8=");
+    }
+
+    #[test]
+    fn join_resolves_dots() {
+        assert_eq!(join("/a/b", "../c/./d"), "/a/c/d");
+        assert_eq!(join("/", "x/fs"), "/x/fs");
+        assert_eq!(join("/", ".."), "/");
+    }
+}

cli/src/world.rs

@@ -5,7 +5,7 @@
 //! <world>/
 //!   actor/<id>/
 //!     actor.wasm        the current version
-//!     actor.json        version, running, subscriptions (broker)
+//!     actor.json        version, running, subscriptions, tapped (broker)
 //!     source/           source of the current version
 //!     public/           written by the actor, read-only for others
 //!     data/             state, writable only by the actor
@@ -22,8 +22,9 @@ use serde::{Deserialize, Serialize};
 /// Name prefix of the host's temporary files and folders. Copies skip them.
 pub const TMP_PREFIX: &str = ".sausage-tmp-";
 
-/// IDs of the actors that the host implements natively.
-pub const NATIVE: &[&str] = &["broker", "console", "cargo"];
+/// IDs of the actors that the host implements natively. `fs` sends the
+/// requests for actor-served files (`vfs.rs`).
+pub const NATIVE: &[&str] = &["broker", "console", "cargo", "fs"];
 
 #[derive(Clone)]
 pub struct World {
@@ -42,6 +43,9 @@ pub struct ActorMeta {
     /// The actors whose published messages this actor receives.
     #[serde(default)]
     pub subscriptions: BTreeSet<String>,
+    /// Whether this actor gets a copy of every routed message (`tap`).
+    #[serde(default)]
+    pub tapped: bool,
 }
 
 fn yes() -> bool {

cli/tests/actors/toolbox/src/main.rs

@@ -39,6 +39,8 @@ fn main() {
     }
 
     let mut open = HashMap::<String, std::fs::File>::new();
+    // Files that this actor serves at /actor/<id>/fs/ (path -> text).
+    let mut served = std::collections::BTreeMap::<String, String>::new();
     let stdin = io::stdin();
     let mut out = io::stdout().lock();
     let mut emit_shutdown = false;
@@ -73,6 +75,40 @@ fn main() {
                 break;
             }
             Some("echo") => json!({"type": "echo", "value": body["value"]}),
+            Some("fs_read") => {
+                let dir = if path == "/" { "/".to_string() } else { format!("{path}/") };
+                let mut names: Vec<String> = served
+                    .keys()
+                    .filter_map(|k| k.strip_prefix(&dir))
+                    .map(|rest| match rest.split_once('/') {
+                        Some((first, _)) => format!("{first}/"),
+                        None => rest.to_string(),
+                    })
+                    .collect();
+                names.dedup();
+                if let Some(text) = served.get(&path) {
+                    json!({"type": "fs_file", "text": text})
+                } else if path == "/" || !names.is_empty() {
+                    let entries: Vec<Value> = names
+                        .iter()
+                        .map(|n| match n.strip_suffix('/') {
+                            Some(d) => json!({"name": d, "type": "dir"}),
+                            None => json!({"name": n, "type": "file"}),
+                        })
+                        .collect();
+                    json!({"type": "fs_dir", "entries": entries})
+                } else {
+                    json!({"type": "fs_error", "errno": "no-entry"})
+                }
+            }
+            Some("fs_write") => {
+                served.insert(path.clone(), text.to_string());
+                json!({"type": "fs_done"})
+            }
+            Some("fs_remove") => match served.remove(&path) {
+                Some(_) => json!({"type": "fs_done"}),
+                None => json!({"type": "fs_error", "errno": "no-entry"}),
+            },
             Some("publish") => {
                 send(
                     &mut out,

cli/tests/host.rs

@@ -216,6 +216,101 @@ async fn actors_can_read_but_not_write_other_actor_data() {
     host.stop().await;
 }
 
+#[tokio::test]
+async fn actors_serve_files_under_fs() {
+    // "writer" serves its fs/ from memory (toolbox fs_read / fs_write / fs_remove).
+    let (_, host, mut mailbox) = filesystem_world("actor-fs").await;
+    let file = "/actor/writer/fs/notes/a.txt";
+    let write = json!({"type": "write", "path": file, "text": "hello"});
+    ask_type(&host, &mut mailbox, "reader", write, "ok").await;
+    // The write reached the owner as one fs_write message.
+    let direct = json!({"type": "fs_read", "path": "/notes/a.txt"});
+    let served = ask_type(&host, &mut mailbox, "writer", direct, "fs_file").await;
+    assert_eq!(served.body["text"], "hello");
+
+    let read = json!({"type": "read", "path": file});
+    assert_eq!(ask_type(&host, &mut mailbox, "reader", read, "content").await.body["text"], "hello");
+    let list = json!({"type": "list", "path": "/actor/writer/fs/notes"});
+    assert_eq!(ask_type(&host, &mut mailbox, "reader", list, "entries").await.body["names"], json!(["a.txt"]));
+    let root = json!({"type": "list", "path": "/actor/writer/fs"});
+    assert_eq!(ask_type(&host, &mut mailbox, "reader", root, "entries").await.body["names"], json!(["notes"]));
+
+    // fs_error and undeliverable become errors of the file operation.
+    for path in ["/actor/writer/fs/missing.txt", "/actor/nobody/fs/x"] {
+        ask_type(&host, &mut mailbox, "reader", json!({"type": "read", "path": path}), "error").await;
+    }
+    // An actor that does not answer (here: it is busy reading itself) fails after the timeout.
+    let id = host.broker.send("console", Some("reader"), None, None, json!({"type": "read", "path": "/actor/reader/fs/x"}));
+    let timed_out = receive_reply(&mut mailbox, &id, Duration::from_secs(20)).await;
+    assert_eq!(timed_out.body["type"], "error");
+    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]
 async fn files_are_plain_posix_files() {
     let (_, host, mut mailbox) = filesystem_world("posix").await;
@@ -464,6 +559,62 @@ async fn explicit_rollback_keeps_current_data() {
     host.stop().await;
 }
 
+#[tokio::test]
+async fn remove_shuts_an_actor_down_and_deletes_its_folder() {
+    let world = create_world("remove");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await;
+    let write = json!({"type": "write", "path": "/data/state", "text": "x"});
+    ask_type(&host, &mut mailbox, "doomed", write, "ok").await;
+    ask_type(&host, &mut mailbox, "doomed", json!({"type": "arm-shutdown"}), "armed").await;
+
+    // Normal shutdown: the actor gets "shutdown" and its last message is routed.
+    let id = host.broker.send("console", Some("broker"), None, None, json!({"type": "remove", "actor": "doomed"}));
+    let (mut removed, mut last_words) = (false, false);
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !removed || !last_words {
+            let message = mailbox.recv().await.expect("console mailbox closed");
+            if message.reply_to.as_deref() == Some(id.as_str()) {
+                assert_eq!(message.body, json!({"type": "removed", "actor": "doomed"}));
+                removed = true;
+            } else if message.reply_to.as_deref() == Some("broker/shutdown") {
+                assert_eq!(message.body["type"], "shutdown-reply");
+                last_words = true;
+            }
+        }
+    })
+    .await
+    .expect("remove did not answer or did not shut the actor down");
+    assert!(!world.actor_dir("doomed").exists());
+
+    let gone = ask(&host, &mut mailbox, "doomed", json!({"type": "echo", "value": "hi"})).await;
+    assert_eq!(gone.body["type"], "undeliverable");
+    let again = request(&host, &mut mailbox, json!({"type": "remove", "actor": "doomed"})).await;
+    assert_eq!(again.body["type"], "error");
+    let broker = request(&host, &mut mailbox, json!({"type": "remove", "actor": "broker"})).await;
+    assert_eq!(broker.body["type"], "error");
+    assert!(world.actor_dir("broker").exists());
+
+    // The ID is free again: a new spawn starts at version 1 with empty data.
+    assert_eq!(spawn_type(&host, &mut mailbox, "doomed", &wasm, &source).await.body["version"], 1);
+    let read = json!({"type": "read", "path": "/data/state"});
+    ask_type(&host, &mut mailbox, "doomed", read, "error").await;
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn unknown_broker_request_lists_the_known_ones() {
+    let world = create_world("unknown-request");
+    let (host, mut mailbox) = start(&world).await;
+    let unknown = request(&host, &mut mailbox, json!({"type": "frobnicate"})).await;
+    assert_eq!(unknown.body["type"], "error");
+    let reason = unknown.body["reason"].as_str().unwrap();
+    assert!(reason.contains("remove") && reason.contains("tap"), "{reason}");
+    host.stop().await;
+}
+
 #[tokio::test]
 async fn cargo_actor_builds_valid_crates_and_reports_compiler_errors() {
     let world = create_world("cargo-actor");
@@ -712,3 +863,114 @@ async fn running_actors_have_sockets_and_candidates_do_not() {
     assert_eq!(answer.body, json!({"type": "tcp", "line": "echo: hi"}));
     host.stop().await;
 }
+
+/// Every message that arrives until the mailbox is quiet for `quiet`.
+async fn drain(mailbox: &mut mpsc::Receiver<Envelope>, quiet: Duration) -> Vec<Envelope> {
+    let mut messages = Vec::new();
+    while let Ok(Some(message)) = tokio::time::timeout(quiet, mailbox.recv()).await {
+        messages.push(message);
+    }
+    messages
+}
+
+#[tokio::test]
+async fn tap_copies_every_routed_message_once() {
+    let world = create_world("tap");
+    let wasm = toolbox_wasm().clone();
+    let (host, mut mailbox) = start(&world).await;
+    let source = passing_source(&world, "source");
+    spawn_type(&host, &mut mailbox, "tapped", &wasm, &source).await;
+    let subscribe = json!({"type": "subscribe", "actor": "tapped"});
+    assert_eq!(request(&host, &mut mailbox, subscribe).await.body["type"], "subscribed");
+    assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
+
+    let echo = json!({"type": "echo", "value": "hi"});
+    let echo_id = host.broker.send("console", Some("tapped"), None, None, echo.clone());
+    let publish = json!({"type": "publish", "value": "news"});
+    host.broker.send("console", Some("tapped"), None, None, publish);
+    let messages = drain(&mut mailbox, Duration::from_secs(2)).await;
+    let copies: Vec<&Value> = messages
+        .iter()
+        .filter(|m| m.from == "broker" && m.body["type"] == "tapped")
+        .map(|m| &m.body)
+        .collect();
+    // 4 copies (2 requests, 1 reply, 1 publish) + the reply + the publish itself.
+    assert_eq!((copies.len(), messages.len()), (4, 6), "{messages:?}");
+    for copy in &copies {
+        assert!(copy["ts"].as_u64().unwrap() > 0, "{copy}");
+    }
+    let count = |want: Value| copies.iter().filter(|c| c["message"] == want).count();
+    let request_copy = json!({"id": echo_id, "from": "console", "to": "tapped", "reply_to": null, "body": echo});
+    assert_eq!(count(request_copy), 1, "{copies:?}");
+    let reply_copies = copies
+        .iter()
+        .filter(|c| c["message"]["reply_to"] == echo_id.as_str() && c["message"]["from"] == "tapped" && c["message"]["to"] == "console")
+        .count();
+    assert_eq!(reply_copies, 1, "{copies:?}");
+    let publish_copy = copies.iter().find(|c| c["message"]["body"]["type"] == "published").expect("publish not tapped");
+    assert_eq!(publish_copy["message"]["from"], "tapped");
+    assert_eq!(publish_copy["message"]["to"], Value::Null);
+
+    assert_eq!(request(&host, &mut mailbox, json!({"type": "untap"})).await.body["type"], "untapped");
+    let echo_id = host.broker.send("console", Some("tapped"), None, None, json!({"type": "echo", "value": "quiet"}));
+    let messages = drain(&mut mailbox, Duration::from_secs(1)).await;
+    assert_eq!(messages.len(), 1, "{messages:?}");
+    assert_eq!(messages[0].reply_to.as_deref(), Some(echo_id.as_str()));
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn tap_survives_a_restart() {
+    let world = create_world("tap-restart");
+    let (host, mut mailbox) = start(&world).await;
+    assert_eq!(request(&host, &mut mailbox, json!({"type": "tap"})).await.body["type"], "tapping");
+    host.stop().await;
+    assert!(world.read_meta("console").tapped);
+
+    let (host, mut mailbox) = start(&world).await;
+    let id = host.broker.send("console", Some("nobody"), None, None, json!({"type": "hello"}));
+    let copy = tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            let m = mailbox.recv().await.expect("console mailbox closed");
+            if m.body["type"] == "tapped" && m.body["message"]["id"] == id.as_str() {
+                return m;
+            }
+        }
+    })
+    .await
+    .expect("no tapped copy after the restart");
+    assert_eq!(copy.body["message"]["to"], "nobody");
+    host.stop().await;
+}
+
+#[tokio::test]
+async fn a_full_tapper_mailbox_drops_copies_without_blocking() {
+    let world = create_world("tap-full");
+    let (host, mut mailbox) = start(&world).await;
+    // "ghost" taps and never reads its mailbox.
+    let mut ghost = host.mailbox("ghost");
+    let mut sink = host.mailbox("sink");
+    host.broker.send("ghost", Some("broker"), None, None, json!({"type": "tap"}));
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !world.read_meta("ghost").tapped {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("tap was not saved");
+
+    let total = sausage::broker::MAILBOX_SIZE + 500;
+    for n in 0..total {
+        host.broker.send("console", Some("sink"), None, None, json!({"type": "n", "n": n}));
+        assert_eq!(sink.try_recv().unwrap().body["n"], n);
+    }
+    assert!(mailbox.try_recv().is_err(), "a dropped copy reported something to the sender");
+    let mut held = 0;
+    while ghost.try_recv().is_ok() {
+        held += 1;
+    }
+    assert_eq!(held, sausage::broker::MAILBOX_SIZE);
+    host.broker.send("console", Some("sink"), None, None, json!({"type": "after"}));
+    assert_eq!(ghost.try_recv().unwrap().body["type"], "tapped");
+    host.stop().await;
+}

flake.nix

@@ -36,15 +36,14 @@
           meta.mainProgram = "sausage";
         };
         # `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,
         # so the world updates itself. `/quit` (exit code 0) ends the loop.
-        # `--internal` (or `.#loop-internal`) uses Gitea's SSH port on the LAN.
-        publicUrl = "git+ssh://containeruser@gitea.surma.technology:2222/surma/sausage";
-        internalUrl = "git+ssh://containeruser@10.0.0.2:2222/surma/sausage";
-        loop = name: url: pkgs.writeShellScriptBin name ''
-          url="${url}"
-          if [ "''${1:-}" = "--internal" ]; then url="${internalUrl}"; shift; fi
+        # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
+        # e.g. a fork or a local checkout).
+        repoUrl = "git+https://gitea.surma.technology/surma/sausage";
+        loop = pkgs.writeShellScriptBin "sausage-loop" ''
+          url="''${SAUSAGE_FLAKE:-${repoUrl}}"
           if [ $# -eq 0 ]; then set -- world; fi
           while true; do
             nix run --refresh "$url#default" -- run "$@"
@@ -72,8 +71,7 @@
         });
 
         apps.default = { type = "app"; program = "${sausage}/bin/sausage"; };
-        apps.loop = { type = "app"; program = "${loop "sausage-loop" publicUrl}/bin/sausage-loop"; };
-        apps.loop-internal = { type = "app"; program = "${loop "sausage-loop-internal" internalUrl}/bin/sausage-loop-internal"; };
+        apps.loop = { type = "app"; program = "${loop}/bin/sausage-loop"; };
 
         devShells.default = pkgs.mkShell {
           packages = [ rust pkgs.wasmtime pkgs.wasm-tools ];