Keep discovery test file at served root

{"type":"remove","actor"} shuts the actor down normally, drops its
mailbox, supervisor and actor.json entry (so also its subscriptions and
tap), and deletes /actor/<id>/. Native actors cannot be removed.
Re-applied from card-1-broker-remove (3c91747) onto card-2-broker-tap.

PROTOCOL.md

@@ -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`
 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"}]}` |
@@ -109,7 +116,7 @@ Any request can be answered with `{"type": "fs_error", "errno": "no-entry"}`.
 
 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`,
+`rolled_back`, `stopped`, `started`, `removed`, `subscribed`, `unsubscribed`,
 `tapping`, `untapped`, `scanned`, `exiting`), or `{"type": "error", "reason": "..."}`.
 
 | `type` | Fields | Effect |
@@ -118,6 +125,7 @@ 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 |

cli/src/broker.rs

@@ -1,7 +1,7 @@
 //! The broker: message routing, mailboxes, subscriptions, taps, the version
 //! store, and the requests to `broker` (spawn, upgrade, rollback, stop, start,
-//! subscribe, unsubscribe, tap, untap, scan, exit). `supervisor.rs` runs each
-//! WASM actor.
+//! 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
@@ -377,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.
@@ -481,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";
@@ -506,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, tap, untap, scan, exit"),
+            other => bail!("unknown broker request {other:?}. Known: spawn, upgrade, rollback, stop, start, remove, subscribe, unsubscribe, tap, untap, scan, exit"),
         }
     }
 

cli/src/vfs.rs

@@ -4,7 +4,8 @@
 //! 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. 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_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
 //! 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::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, Weak};
@@ -41,13 +42,14 @@ pub struct VfsState {
     broker: Weak<Broker>,
     /// The actor of this instance, for notes in its `stderr.log`.
     me: String,
-    /// Descriptors of the `/actor` preopen.
-    roots: HashSet<u32>,
+    /// 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, roots: HashSet::new() }
+        Self { broker, me, dirs: HashMap::new() }
     }
 }
 
@@ -199,6 +201,33 @@ fn join(base: &str, rel: &str) -> String {
     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_,
@@ -266,6 +295,14 @@ impl VfsView<'_> {
         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()
     }
@@ -282,16 +319,26 @@ impl VfsView<'_> {
         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 if self.state.roots.contains(&fd.rep()) => {
-                let full = join("/", path);
-                let mut parts = full.splitn(4, '/').skip(1);
-                let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or(""));
-                (fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}")))
+            None => {
+                let base = self.state.dirs.get(&fd.rep())?;
+                actor_fs_path(&join(base, path))
             }
-            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 mut entries = Vec::new();
+        loop {
+            match HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream.clone()).await? {
+                Some(entry) => entries.push(entry),
+                None => break,
+            }
+        }
+        HostDirectoryEntryStream::drop(&mut self.inner, stream)?;
+        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()))
@@ -343,7 +390,7 @@ impl preopens::Host for VfsView<'_> {
         let dirs = preopens::Host::get_directories(&mut self.inner)?;
         for (fd, name) in &dirs {
             if name == "/actor" {
-                self.state.roots.insert(fd.rep());
+                self.state.dirs.insert(fd.rep(), "/actor".to_string());
             }
         }
         Ok(dirs)
@@ -453,7 +500,19 @@ impl HostDescriptor for VfsView<'_> {
                 Node::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 +584,16 @@ impl HostDescriptor for VfsView<'_> {
     ) -> FsResult<Resource<types::Descriptor>> {
         match self.target(&fd, &path) {
             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
+                    && HostDescriptor::get_type(&mut self.inner, opened.clone()).await? == DescriptorType::Directory
+                {
+                    self.state.dirs.insert(opened.rep(), path);
+                }
+                Ok(opened)
+            }
         }
     }
 
@@ -538,7 +606,7 @@ impl HostDescriptor for VfsView<'_> {
             }
             return Ok(());
         }
-        self.state.roots.remove(&fd.rep());
+        self.state.dirs.remove(&fd.rep());
         HostDescriptor::drop(&mut self.inner, fd)
     }
 
@@ -698,6 +766,17 @@ fn base64_decode(text: &str) -> Option<Vec<u8>> {
 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"] {

cli/tests/host.rs

@@ -246,6 +246,71 @@ async fn actors_serve_files_under_fs() {
     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;
@@ -494,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");