Run world service policy in WASI init with pollable process observation

host/src/harness/mod.rs

@@ -208,7 +208,6 @@ pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow::
     })?;
     let services = crate::process::supervise(runtime.clone());
     let result = run_with_exec(world, Arc::new(runtime)).await;
-    services.abort();
-    let _ = services.await;
+    services.shutdown().await;
     result
 }

host/src/process/mod.rs

@@ -41,11 +41,14 @@ mod bindings {
         with: {
             "wasi:io": wasmtime_wasi::p2::bindings::io,
             "sausage:process/host.child": super::Child,
+            "sausage:process/host.watcher": super::Watcher,
         },
     });
 }
 use bindings::sausage::process::host::{Host, HostChild, Request, Stdio};
 mod services;
+mod watcher;
+pub use watcher::Watcher;
 mod stdio;
 pub use services::supervise;
 use stdio::Io;
@@ -470,7 +473,23 @@ impl Host for State {
         Ok(self.runtime.kill_pid(pid).map_err(|e| e.to_string()))
     }
 }
+#[async_trait::async_trait]
+impl wasmtime_wasi::p2::Pollable for Child {
+    async fn ready(&mut self) {
+        while self.code.borrow_and_update().is_none() {
+            if self.code.changed().await.is_err() {
+                break;
+            }
+        }
+    }
+}
 impl HostChild for State {
+    async fn subscribe(
+        &mut self,
+        child: Resource<Child>,
+    ) -> wasmtime::Result<Resource<wasmtime_wasi::p2::DynPollable>> {
+        wasmtime_wasi::p2::subscribe(&mut self.table, child)
+    }
     async fn pid(&mut self, child: Resource<Child>) -> wasmtime::Result<u32> {
         Ok(self.table.get(&child)?.pid)
     }

host/src/process/services.rs

@@ -1,49 +1,28 @@
-//! Supervises the world's services list, live edits, crash limits, and graceful stops.
-use super::{GroupGuard, Request, Runtime, Started, Stdio};
-use serde::Deserialize;
-use std::{
-    collections::{BTreeMap, VecDeque},
-    sync::Arc,
-    time::Duration,
-};
-use tokio::time::Instant;
-
-#[derive(Clone, Deserialize, PartialEq, Eq)]
-struct Spec {
-    name: String,
-    argv: Vec<String>,
-    #[serde(default = "root")]
-    cwd: String,
-    #[serde(default)]
-    env: BTreeMap<String, String>,
+//! Starts world init, reaps its command tree, and applies host restart and shutdown deadlines.
+use super::{GroupGuard, Request, Runtime, Stdio};
+use std::{sync::Arc, time::Duration};
+pub struct Supervisor {
+    shutdown: tokio::sync::oneshot::Sender<()>,
+    task: tokio::task::JoinHandle<()>,
 }
-fn root() -> String {
-    "/".into()
+impl Supervisor {
+    pub async fn shutdown(self) {
+        let _ = self.shutdown.send(());
+        let _ = self.task.await;
+    }
 }
-struct Live {
-    child: Started,
-    _guard: GroupGuard,
-}
-struct Service {
-    spec: Spec,
-    live: Option<Live>,
-    crashes: VecDeque<Instant>,
-    stopped: bool,
-}
-impl Service {
-    async fn start(&mut self, runtime: &Arc<Runtime>) {
-        match runtime
-            .start(
+/// The host knows only init. Unit policy belongs to the WASI program.
+pub fn supervise(runtime: Arc<Runtime>) -> Supervisor {
+    let (shutdown, mut stop) = tokio::sync::oneshot::channel();
+    let task = tokio::spawn(async move {
+        let mut backoff = Duration::from_secs(1);
+        loop {
+            let start = runtime.start(
                 Request {
-                    program: self.spec.argv.first().cloned().unwrap_or_default(),
-                    argv: self.spec.argv.clone(),
-                    cwd: self.spec.cwd.clone(),
-                    env: self
-                        .spec
-                        .env
-                        .iter()
-                        .map(|(k, v)| (k.clone(), v.clone()))
-                        .collect(),
+                    program: "/bin/init.wasm".into(),
+                    argv: vec!["/bin/init.wasm".into()],
+                    cwd: "/".into(),
+                    env: Vec::new(),
                     group: None,
                     stdin: Stdio::Pipe,
                     stdout: Stdio::Null,
@@ -54,116 +33,56 @@ impl Service {
                 0,
                 None,
                 None,
-            )
-            .await
-        {
-            Ok(child) => {
+            );
+            let child = tokio::select! { _ = &mut stop => return, child = start => child };
+            if let Ok(mut child) = child {
                 let guard = GroupGuard {
                     runtime: runtime.clone(),
                     group: child.pid,
                 };
-                self.live = Some(Live {
-                    child,
-                    _guard: guard,
-                });
+                let exiting = async {
+                    while child.code.borrow_and_update().is_none() {
+                        if child.code.changed().await.is_err() {
+                            break;
+                        }
+                    }
+                };
+                let shutdown = tokio::select! { _ = &mut stop => true, _ = exiting => false };
+                if shutdown {
+                    drop(child.input.take());
+                    let _ = tokio::time::timeout(Duration::from_secs(15), async {
+                        while child.code.borrow_and_update().is_none() {
+                            if child.code.changed().await.is_err() {
+                                break;
+                            }
+                        }
+                    })
+                    .await;
+                    drop(guard);
+                    reaped(&runtime, child.pid).await;
+                    return;
+                }
+                // This guard kills all descendants, including separate process groups.
+                drop(guard);
+                reaped(&runtime, child.pid).await;
             }
-            Err(_) => self.crashed(),
+            tokio::select! { _ = &mut stop => return, _ = tokio::time::sleep(backoff) => () }
+            backoff = (backoff * 2).min(Duration::from_secs(30));
         }
-    }
-    fn crashed(&mut self) {
-        let now = Instant::now();
-        while self
-            .crashes
-            .front()
-            .is_some_and(|t| now.duration_since(*t) >= Duration::from_secs(10))
+    });
+    Supervisor { shutdown, task }
+}
+async fn reaped(runtime: &Runtime, command: u32) {
+    loop {
+        if !runtime
+            .entries
+            .lock()
+            .unwrap()
+            .values()
+            .any(|entry| entry.command == command)
         {
-            self.crashes.pop_front();
-        }
-        self.crashes.push_back(now);
-        self.stopped = self.crashes.len() > 5;
-    }
-    async fn stop(&mut self) {
-        if let Some(mut live) = self.live.take() {
-            drop(live.child.input.take());
-            let _ = tokio::time::timeout(Duration::from_secs(5), async {
-                loop {
-                    if live.child.code.borrow().is_some() {
-                        break;
-                    }
-                    if live.child.code.changed().await.is_err() {
-                        break;
-                    }
-                }
-            })
-            .await;
-            // The guard stops remaining descendants, or the process after its grace period.
+            return;
         }
+        tokio::task::yield_now().await;
     }
 }
-/// Start the supervisor. Aborting its task stops its service processes.
-/// The YAML file contains name, argv, optional cwd, and an optional map of environment variables.
-pub fn supervise(runtime: Arc<Runtime>) -> tokio::task::JoinHandle<()> {
-    tokio::spawn(async move {
-        let mut services: BTreeMap<String, Service> = BTreeMap::new();
-        let mut previous = Vec::new();
-        let mut interval = tokio::time::interval(Duration::from_millis(100));
-        loop {
-            interval.tick().await;
-            let content = match runtime
-                .world
-                .root
-                .open("/config/services.yaml", Default::default())
-                .await
-            {
-                Ok(file) => match file.stat().await {
-                    Ok(stat) => file.read(0, stat.size as usize).await.ok(),
-                    Err(_) => None,
-                },
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(b"[]".to_vec()),
-                Err(_) => None,
-            };
-            if let Some(content) = content.filter(|c| *c != previous)
-                && let Ok(specs) = serde_yaml_ng::from_slice::<Vec<Spec>>(&content)
-            {
-                let specs: BTreeMap<_, _> =
-                    specs.into_iter().map(|s| (s.name.clone(), s)).collect();
-                let removed: Vec<_> = services
-                    .keys()
-                    .filter(|name| !specs.contains_key(*name))
-                    .cloned()
-                    .collect();
-                for name in removed {
-                    services.remove(&name).unwrap().stop().await;
-                }
-                for (name, spec) in specs {
-                    let service = services.entry(name).or_insert_with(|| Service {
-                        spec: spec.clone(),
-                        live: None,
-                        crashes: VecDeque::new(),
-                        stopped: false,
-                    });
-                    if service.spec != spec {
-                        service.stop().await;
-                    }
-                    service.spec = spec;
-                    service.crashes.clear();
-                    service.stopped = false;
-                }
-                previous = content;
-            }
-            for service in services.values_mut() {
-                if service
-                    .live
-                    .as_ref()
-                    .is_some_and(|live| live.child.code.borrow().is_some())
-                {
-                    service.live = None;
-                    service.crashed();
-                }
-                if service.live.is_none() && !service.stopped {
-                    service.start(&runtime).await;
-                }
-            }
-        }
-    })
-}

host/src/process/watcher.rs

@@ -0,0 +1,110 @@
+//! Adapts VFS observation to bounded, pollable world-path change queues.
+use super::{
+    State,
+    bindings::sausage::process::host::{Change, ChangeKind, HostWatcher},
+};
+use sausage_vfs::observe::{ChangeKind as Kind, under};
+use tokio::sync::broadcast::{Receiver, error::TryRecvError};
+use wasmtime::component::Resource;
+use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe};
+
+pub struct Watcher {
+    paths: Vec<String>,
+    events: Receiver<(String, Kind)>,
+    pending: Vec<Change>,
+}
+impl Watcher {
+    fn push(&mut self, path: String, kind: Kind) {
+        if self.paths.iter().any(|base| under(&path, base)) {
+            if self
+                .pending
+                .iter()
+                .any(|c| matches!(c.kind, ChangeKind::Overflow))
+            {
+                return;
+            }
+            if self.pending.len() >= 256 {
+                self.overflow();
+                return;
+            }
+            self.pending.push(Change {
+                path,
+                kind: match kind {
+                    Kind::Created => ChangeKind::Created,
+                    Kind::Modified => ChangeKind::Modified,
+                    Kind::Removed => ChangeKind::Removed,
+                },
+            });
+        }
+    }
+    fn overflow(&mut self) {
+        self.pending.clear();
+        self.pending.push(Change {
+            path: "/".into(),
+            kind: ChangeKind::Overflow,
+        });
+    }
+    fn drain(&mut self) {
+        loop {
+            match self.events.try_recv() {
+                Ok((path, kind)) => self.push(path, kind),
+                Err(TryRecvError::Lagged(_)) => self.overflow(),
+                Err(_) => break,
+            }
+        }
+    }
+}
+#[async_trait::async_trait]
+impl Pollable for Watcher {
+    async fn ready(&mut self) {
+        loop {
+            self.drain();
+            if !self.pending.is_empty() {
+                return;
+            }
+            match self.events.recv().await {
+                Ok((path, kind)) => self.push(path, kind),
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => self.overflow(),
+                Err(_) => return,
+            }
+        }
+    }
+}
+impl HostWatcher for State {
+    async fn new(&mut self, paths: Vec<String>) -> wasmtime::Result<Resource<Watcher>> {
+        let observer = self
+            .runtime
+            .world
+            .root
+            .observer()
+            .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?;
+        Ok(self.table.push(Watcher {
+            paths: paths
+                .iter()
+                .map(|p| {
+                    sausage_vfs::resolve("/", p)
+                        .trim_end_matches('/')
+                        .to_owned()
+                })
+                .map(|p| if p.is_empty() { "/".into() } else { p })
+                .collect(),
+            events: observer.subscribe(),
+            pending: Vec::new(),
+        })?)
+    }
+    async fn subscribe(
+        &mut self,
+        watcher: Resource<Watcher>,
+    ) -> wasmtime::Result<Resource<DynPollable>> {
+        subscribe(&mut self.table, watcher)
+    }
+    async fn changes(&mut self, watcher: Resource<Watcher>) -> wasmtime::Result<Vec<Change>> {
+        let watcher = self.table.get_mut(&watcher)?;
+        watcher.drain();
+        Ok(std::mem::take(&mut watcher.pending))
+    }
+    async fn drop(&mut self, watcher: Resource<Watcher>) -> wasmtime::Result<()> {
+        self.table.delete(watcher)?;
+        Ok(())
+    }
+}

host/tests/init.rs

@@ -0,0 +1,455 @@
+//! Init acceptance tests use real WASI programs, a persistent guest cache, and temporary worlds.
+use sausage::{
+    exec::{Command, Exec, Options, Stream},
+    process::{Runtime, supervise},
+    world::World,
+};
+use sausage_vfs::{Layer, Open};
+use std::{
+    path::{Path, PathBuf},
+    sync::{Arc, OnceLock},
+    time::Duration,
+};
+static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+fn guests() -> &'static Path {
+    static BUILT: OnceLock<PathBuf> = OnceLock::new();
+    BUILT.get_or_init(|| {
+        let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-guests");
+        let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland/Cargo.toml");
+        assert!(
+            std::process::Command::new("cargo")
+                .args([
+                    "build",
+                    "--release",
+                    "--target",
+                    "wasm32-wasip2",
+                    "-p",
+                    "sausage-init",
+                    "--manifest-path"
+                ])
+                .arg(manifest)
+                .env("CARGO_TARGET_DIR", &target)
+                .status()
+                .unwrap()
+                .success()
+        );
+        target.join("wasm32-wasip2/release")
+    })
+}
+fn world() -> (tempfile::TempDir, Arc<Runtime>, Arc<dyn Layer>) {
+    let temp = tempfile::tempdir().unwrap();
+    let bin = temp.path().join("world/bin");
+    std::fs::create_dir_all(&bin).unwrap();
+    for (source, name) in [
+        ("init", "init"),
+        ("systemctl", "systemctl"),
+        ("journalctl", "journalctl"),
+        ("systemd-run", "systemd-run"),
+        ("init-probe", "probe"),
+    ] {
+        std::fs::copy(
+            guests().join(format!("{source}.wasm")),
+            bin.join(format!("{name}.wasm")),
+        )
+        .unwrap();
+    }
+    std::fs::create_dir_all(
+        temp.path()
+            .join("world/etc/systemd/system/default.target.wants"),
+    )
+    .unwrap();
+    let cache = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-components");
+    std::fs::create_dir_all(&cache).unwrap();
+    std::fs::create_dir_all(temp.path().join("host")).unwrap();
+    std::os::unix::fs::symlink(cache, temp.path().join("host/compile-cache")).unwrap();
+    let world = World::open(temp.path()).unwrap();
+    let root = world.root.clone();
+    (temp, Runtime::new(world).unwrap(), root)
+}
+async fn write(root: &Arc<dyn Layer>, path: &str, text: &[u8]) {
+    let file = root
+        .open(
+            path,
+            Open {
+                create: true,
+                write: true,
+                truncate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    file.write(0, text.to_vec()).await.unwrap();
+}
+async fn unit(root: &Arc<dyn Layer>, name: &str, body: &str) {
+    write(
+        root,
+        &format!("/etc/systemd/system/default.target.wants/{name}.service"),
+        b"",
+    )
+    .await;
+    write(
+        root,
+        &format!("/etc/systemd/system/{name}.service"),
+        body.as_bytes(),
+    )
+    .await;
+}
+async fn run(runtime: &Arc<Runtime>, argv: &[&str]) -> (i32, String) {
+    let mut out = Vec::new();
+    let code = runtime
+        .exec(
+            Command::Argv(argv.iter().map(|a| (*a).into()).collect()),
+            Options {
+                timeout: Some(Duration::from_secs(30)),
+                cwd: "/".into(),
+                ..Default::default()
+            },
+            &mut |stream, bytes| {
+                if stream == Stream::Stdout {
+                    out.extend_from_slice(bytes);
+                } else {
+                    eprint!("{}", String::from_utf8_lossy(bytes));
+                }
+            },
+        )
+        .await
+        .unwrap();
+    (code, String::from_utf8_lossy(&out).into_owned())
+}
+async fn wait(mut check: impl FnMut() -> bool) {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        while !check() {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+}
+async fn state(root: &Arc<dyn Layer>, unit: &str, expected: &str) {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            if let Ok(file) = root
+                .open(
+                    &format!("/run/systemd/state/{unit}.service"),
+                    Default::default(),
+                )
+                .await
+                && let Ok(bytes) = file.read(0, 65536).await
+                && serde_json::from_slice::<serde_json::Value>(&bytes)
+                    .is_ok_and(|s| s["state"] == expected)
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+}
+fn lines(path: &Path) -> Vec<String> {
+    std::fs::read_to_string(path)
+        .unwrap_or_default()
+        .lines()
+        .map(str::to_owned)
+        .collect()
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn watcher_and_child_exit_pollable() {
+    let _serial = SERIAL.lock().await;
+    let (_temp, runtime, _) = world();
+    let result = run(&runtime, &["/bin/probe.wasm", "watch"]).await;
+    assert_eq!(result.0, 0, "{}", result.1);
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn dependencies_restart_limit_and_journal() {
+    let _serial = SERIAL.lock().await;
+    let (temp, runtime, root) = world();
+    unit(
+        &root,
+        "prepare",
+        "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm oneshot /order.log\n",
+    )
+    .await;
+    unit(&root, "after", "[Unit]\nAfter=prepare.service\nRequires=prepare.service\n[Service]\nExecStart=/bin/probe.wasm after /after.log\n").await;
+    unit(
+        &root,
+        "fail",
+        "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm exit 2\n",
+    )
+    .await;
+    unit(&root, "blocked", "[Unit]\nAfter=fail.service\nRequires=fail.service\n[Service]\nExecStart=/bin/probe.wasm oneshot /blocked.log\n").await;
+    unit(&root, "crash", "[Service]\nExecStart=/bin/probe.wasm crash /crash.log\nRestart=on-failure\nRestartSec=0.01\n").await;
+    unit(&root, "always", "[Service]\nExecStart=/bin/probe.wasm oneshot /always.log\nRestart=always\nRestartSec=0.01\nStartLimitBurst=2\n").await;
+    unit(
+        &root,
+        "success",
+        "[Service]\nExecStart=/bin/probe.wasm oneshot /success.log\nRestart=on-failure\n",
+    )
+    .await;
+    unit(
+        &root,
+        "no",
+        "[Service]\nExecStart=/bin/probe.wasm crash /no.log\nRestart=no\n",
+    )
+    .await;
+    unit(
+        &root,
+        "linger",
+        "[Service]\nExecStart=/bin/probe.wasm linger /linger.log\nTimeoutStopSec=0.02\n",
+    )
+    .await;
+    let supervisor = supervise(runtime.clone());
+    let path = temp.path().join("world");
+    wait(|| {
+        lines(&path.join("after.log")) == ["start"] && lines(&path.join("crash.log")).len() == 5
+    })
+    .await;
+    state(&root, "always", "failed").await;
+    assert_eq!(lines(&path.join("always.log")).len(), 2);
+    assert_eq!(lines(&path.join("success.log")).len(), 1);
+    assert_eq!(lines(&path.join("no.log")).len(), 1);
+    state(&root, "crash", "failed").await;
+    state(&root, "after", "active").await;
+    assert_eq!(lines(&path.join("order.log")), ["start"]);
+    assert!(!path.join("blocked.log").exists());
+    wait(|| {
+        std::fs::read_to_string(path.join("var/log/journal/after.service.log"))
+            .unwrap_or_default()
+            .contains("line 10")
+    })
+    .await;
+    assert_eq!(
+        run(&runtime, &["/bin/systemctl.wasm", "is-active", "after"])
+            .await
+            .0,
+        0
+    );
+    assert_eq!(
+        run(&runtime, &["/bin/systemctl.wasm", "is-active", "crash"])
+            .await
+            .0,
+        3
+    );
+    assert_eq!(
+        run(&runtime, &["/bin/systemctl.wasm", "is-failed", "crash"])
+            .await
+            .0,
+        0
+    );
+    let (code, output) = run(
+        &runtime,
+        &["/bin/journalctl.wasm", "-u", "after", "-n", "5"],
+    )
+    .await;
+    assert_eq!(code, 0);
+    assert_eq!(output, "line 6\nline 7\nline 8\nline 9\nline 10\n");
+    state(&root, "linger", "active").await;
+    assert_eq!(
+        run(&runtime, &["/bin/systemctl.wasm", "stop", "linger"])
+            .await
+            .0,
+        0
+    );
+    state(&root, "linger", "inactive").await;
+    assert_eq!(lines(&path.join("linger.log")), ["start"]);
+    supervisor.shutdown().await;
+    assert_eq!(lines(&path.join("after.log")), ["start", "stop"]);
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn live_unit_changes_and_binary_replacement() {
+    let _serial = SERIAL.lock().await;
+    let (temp, runtime, root) = world();
+    let supervisor = supervise(runtime.clone());
+    let path = temp.path().join("world");
+    unit(
+        &root,
+        "live",
+        "[Service]\nExecStart=/bin/probe.wasm server /live.log\n",
+    )
+    .await;
+    wait(|| lines(&path.join("live.log")) == ["start"]).await;
+    write(
+        &root,
+        "/etc/systemd/system/live.service",
+        b"[Service]\nExecStart=/bin/probe.wasm server /live.log\nEnvironment=REV=2\n",
+    )
+    .await;
+    wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await;
+    let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap();
+    write(&root, "/bin/replacement.wasm", &bytes).await;
+    root.rename("/bin/replacement.wasm", "/bin/probe.wasm")
+        .await
+        .unwrap();
+    wait(|| lines(&path.join("live.log")).len() >= 5).await;
+    root.remove_file("/etc/systemd/system/live.service")
+        .await
+        .unwrap();
+    wait(|| {
+        lines(&path.join("live.log"))
+            .last()
+            .is_some_and(|l| l == "stop")
+    })
+    .await;
+    supervisor.shutdown().await;
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn enable_persists_but_transient_units_do_not() {
+    let _serial = SERIAL.lock().await;
+    let (temp, runtime, root) = world();
+    write(&root, "/etc/systemd/system/persistent.service", b"[Service]\nExecStart=/bin/probe.wasm server /persistent.log\n[Install]\nWantedBy=default.target\n").await;
+    let supervisor = supervise(runtime.clone());
+    let path = temp.path().join("world");
+    wait(|| path.join("var/log/journal").exists()).await;
+    assert_eq!(
+        run(
+            &runtime,
+            &["/bin/systemctl.wasm", "enable", "--now", "persistent"]
+        )
+        .await
+        .0,
+        0
+    );
+    wait(|| lines(&path.join("persistent.log")) == ["start"]).await;
+    assert_eq!(
+        run(
+            &runtime,
+            &["/bin/systemctl.wasm", "is-enabled", "persistent"]
+        )
+        .await
+        .0,
+        0
+    );
+    assert_eq!(
+        run(
+            &runtime,
+            &[
+                "/bin/systemd-run.wasm",
+                "--unit",
+                "transient",
+                "--wait",
+                "/bin/probe.wasm",
+                "oneshot",
+                "/transient.log"
+            ]
+        )
+        .await
+        .0,
+        0
+    );
+    supervisor.shutdown().await;
+    drop(root);
+    drop(runtime);
+    let world = World::open(temp.path()).unwrap();
+    assert!(
+        world
+            .root
+            .stat("/run/systemd/transient/transient.service", true)
+            .await
+            .is_err()
+    );
+    let runtime = Runtime::new(world).unwrap();
+    let supervisor = supervise(runtime.clone());
+    wait(|| lines(&path.join("persistent.log")) == ["start", "stop", "start"]).await;
+    assert_eq!(lines(&path.join("transient.log")), ["start"]);
+    supervisor.shutdown().await;
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn init_crash_reaps_children_and_restarts_enabled_units() {
+    let _serial = SERIAL.lock().await;
+    let (temp, runtime, root) = world();
+    unit(
+        &root,
+        "server",
+        "[Service]\nExecStart=/bin/probe.wasm server /server.log\n",
+    )
+    .await;
+    let supervisor = supervise(runtime.clone());
+    let path = temp.path().join("world/server.log");
+    wait(|| lines(&path) == ["start"]).await;
+    assert_eq!(run(&runtime, &["/bin/probe.wasm", "kill-init"]).await.0, 0);
+    wait(|| {
+        lines(&path)
+            .iter()
+            .filter(|line| line.as_str() == "start")
+            .count()
+            == 2
+    })
+    .await;
+    let table = root
+        .open("/proc/processes", Default::default())
+        .await
+        .unwrap()
+        .read(0, 65536)
+        .await
+        .unwrap();
+    let table = String::from_utf8(table).unwrap();
+    assert_eq!(
+        table
+            .lines()
+            .filter(|l| l.contains("/bin/probe.wasm server"))
+            .count(),
+        1
+    );
+    supervisor.shutdown().await;
+}
+#[tokio::test(flavor = "multi_thread")]
+async fn journal_follow_observes_new_lines() {
+    let _serial = SERIAL.lock().await;
+    let (_temp, runtime, root) = world();
+    // Init creates the public journal directory.
+    let supervisor = supervise(runtime.clone());
+    tokio::time::timeout(Duration::from_secs(30), async {
+        while root.stat("/var/log/journal", true).await.is_err() {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    write(&root, "/var/log/journal/x.service.log", b"old\n").await;
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+    let executor = runtime.clone();
+    let task = tokio::spawn(async move {
+        executor
+            .exec(
+                Command::Argv(vec![
+                    "/bin/journalctl.wasm".into(),
+                    "-u".into(),
+                    "x".into(),
+                    "-f".into(),
+                ]),
+                Options {
+                    cwd: "/".into(),
+                    ..Default::default()
+                },
+                &mut |_, bytes| {
+                    let _ = tx.send(bytes.to_vec());
+                },
+            )
+            .await
+    });
+    let first = tokio::time::timeout(Duration::from_secs(30), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(first, b"old\n");
+    let file = root
+        .open(
+            "/var/log/journal/x.service.log",
+            Open {
+                write: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    file.append(b"new\n".to_vec()).await.unwrap();
+    let next = tokio::time::timeout(Duration::from_secs(30), rx.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(next, b"new\n");
+    task.abort();
+    let _ = task.await;
+    supervisor.shutdown().await;
+}

host/tests/process.rs

@@ -12,7 +12,7 @@ use std::{
 mod features {
     //! Acceptance coverage for preview1, identity, inherited IO, services, HTTP, and bundle aliases.
     use super::*;
-    use std::{collections::BTreeMap, time::Instant};
+    use std::collections::BTreeMap;
 
     async fn poll(mut condition: impl FnMut() -> bool) {
         tokio::time::timeout(Duration::from_secs(10), async {
@@ -23,13 +23,6 @@ mod features {
         .await
         .unwrap();
     }
-    fn lines(path: &Path) -> Vec<String> {
-        std::fs::read_to_string(path)
-            .unwrap_or_default()
-            .lines()
-            .map(str::to_owned)
-            .collect()
-    }
     async fn prepare(runtime: &Arc<Runtime>, argv: Vec<String>) {
         assert_eq!(
             runtime
@@ -285,61 +278,6 @@ mod features {
         assert_eq!(runtime.compiled_programs(), 1);
     }
     #[tokio::test(flavor = "multi_thread")]
-    async fn services_restart_edit_limit_and_eof() {
-        let _serial = SERIAL.lock().await;
-        let (temp, runtime, processes) = world();
-        prepare(&runtime, vec!["/bin/launcher".into()]).await;
-        let root = temp.path().join("world");
-        std::fs::create_dir_all(root.join("config")).unwrap();
-        let spec = |name: &str, mode: &str, log: &str| serde_json::json!({"name": name, "argv": ["/bin/launcher", mode, log]});
-        let write = |specs: Vec<serde_json::Value>| {
-            std::fs::write(
-                root.join("config/services.yaml"),
-                serde_yaml_ng::to_string(&specs).unwrap(),
-            )
-            .unwrap()
-        };
-        write(vec![
-            spec("a", "server", "/a.log"),
-            spec("crash", "crash", "/crash.log"),
-        ]);
-        let supervisor = sausage::process::supervise(runtime.clone());
-        poll(|| {
-            lines(&root.join("a.log")) == ["start"] && lines(&root.join("crash.log")).len() == 6
-        })
-        .await;
-        prepare(&runtime, vec!["/bin/launcher".into()]).await;
-        assert_eq!(lines(&root.join("crash.log")).len(), 6);
-        assert!(
-            processes
-                .lock()
-                .unwrap()
-                .values()
-                .all(|p| p.conversation.is_none())
-        );
-        write(vec![spec("b", "server", "/b.log")]);
-        poll(|| {
-            lines(&root.join("a.log")) == ["start", "stop"]
-                && lines(&root.join("b.log")) == ["start"]
-        })
-        .await;
-        let mut changed = spec("b", "server", "/b.log");
-        changed["env"] = serde_json::json!({"REV": "new"});
-        write(vec![changed]);
-        poll(|| lines(&root.join("b.log")) == ["start", "stop", "start"]).await;
-        write(vec![]);
-        poll(|| lines(&root.join("b.log")) == ["start", "stop", "start", "stop"]).await;
-        write(vec![spec("linger", "linger", "/linger.log")]);
-        poll(|| lines(&root.join("linger.log")) == ["start"]).await;
-        let stop = Instant::now();
-        write(vec![]);
-        poll(|| processes.lock().unwrap().is_empty()).await;
-        assert!(stop.elapsed() >= Duration::from_secs(5));
-        supervisor.abort();
-        let _ = supervisor.await;
-        empty(&processes).await;
-    }
-    #[tokio::test(flavor = "multi_thread")]
     async fn git_local_and_anonymous_https_clone() {
         let _serial = SERIAL.lock().await;
         let (temp, runtime, _) = world();

host/wit/process.wit

@@ -3,6 +3,14 @@ package sausage:process@0.1.0;
 /// Programs share one world. These resources provide lifecycle and bounded stdio transport.
 interface host {
     use wasi:io/streams@0.2.12.{input-stream, output-stream};
+    use wasi:io/poll@0.2.12.{pollable};
+    enum change-kind { created, modified, removed, overflow }
+    record change { path: string, kind: change-kind }
+    resource watcher {
+        constructor(paths: list<string>);
+        subscribe: func() -> pollable;
+        changes: func() -> list<change>;
+    }
     enum stdio { pipe, inherit, null }
     record request {
         program: string,
@@ -21,6 +29,7 @@ interface host {
         stdin: func() -> output-stream;
         stdout: func() -> input-stream;
         stderr: func() -> input-stream;
+        subscribe: func() -> pollable;
         status: func() -> option<s32>;
         wait: func() -> s32;
         kill: func();

userland/Cargo.lock

@@ -4045,6 +4045,17 @@ dependencies = [
  "walkdir",
 ]
 
+[[package]]
+name = "sausage-init"
+version = "0.1.0"
+dependencies = [
+ "sausage-spawn",
+ "serde",
+ "serde_json",
+ "shlex 1.3.0",
+ "wasi 0.14.7+wasi-0.2.4",
+]
+
 [[package]]
 name = "sausage-procps"
 version = "0.1.0"

userland/Cargo.toml

@@ -1,7 +1,7 @@
 # The default CLIs of a world, built for wasm32-wasip2. Its own workspace, outside the host's.
 [workspace]
 resolver = "3"
-members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools"]
+members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "init"]
 
 # WASI compatibility patches: git/vendor/PATCHES.md.
 [patch.crates-io]

userland/init/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "sausage-init"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+sausage-spawn = { path = "../spawn" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+wasi = "0.14"
+shlex = "1"
+
+[[bin]]
+name = "init-probe"
+path = "src/probe.rs"
+
+[[bin]]
+name = "init"
+path = "src/main.rs"
+
+[[bin]]
+name = "systemctl"
+path = "src/systemctl.rs"
+
+[[bin]]
+name = "journalctl"
+path = "src/journalctl.rs"
+
+[[bin]]
+name = "systemd-run"
+path = "src/systemd-run.rs"

userland/init/README.md

@@ -0,0 +1,75 @@
+# World init
+
+`/bin/init.wasm` owns unit policy. The host starts init without a conversation.
+The host restarts init after an exit and kills its descendants before the next start.
+
+Init reads persistent units from `/etc/systemd/system/*.service`.
+It reads transient units from `/run/systemd/transient/*.service`.
+The host mounts `/run` in memory, so transient units disappear after a host restart.
+
+Unit files use `[Unit]`, `[Service]`, and `[Install]` sections.
+Init supports dependency ordering, simple and oneshot services, environment variables, and restart policies.
+The default start limit permits five starts within ten seconds.
+A stop closes stdin. Init kills the process group after `TimeoutStopSec`, which defaults to ten seconds.
+When init receives EOF, it stops dependents before prerequisites.
+
+Enable records are files under `default.target.wants` or `multi-user.target.wants` in the persistent unit directory.
+The `[Install]` section supplies `WantedBy` for `systemctl enable`.
+Init observes VFS changes and applies unit edits without `daemon-reload`.
+A replacement of a running unit's executable restarts that unit.
+
+## Control protocol
+
+The protocol uses UTF-8 JSON files. Writers create a temporary file, then rename it to the final path.
+Each request uses a unique identifier.
+
+1. The client watches `/run/systemd/replies/<id>` before it submits the request.
+2. The client writes `/run/systemd/requests/<id>.json`.
+3. Init reads the request and applies the action.
+4. Init writes `/run/systemd/replies/<id>` and removes the request.
+5. The client reads and removes the reply.
+
+Request example:
+
+```json
+{"action":"start","unit":"example.service"}
+```
+
+Actions are `start`, `stop`, and `restart`.
+A reply acknowledges the action. It does not imply that the service completed its start or stop.
+The control tools then wait for the requested state. A failed start returns code 1.
+
+Reply example:
+
+```json
+{"code":0,"text":""}
+```
+
+A missing unit returns code 5. An unknown action returns code 1.
+Clients read `/run/systemd/state/<unit>` for the current state:
+
+```json
+{"state":"active","pid":42,"code":null}
+```
+
+States are `inactive`, `activating`, `active`, `deactivating`, and `failed`.
+The `code` field records the most recent exit status.
+`systemctl is-active` returns 3 for an inactive unit.
+`systemctl is-failed` returns 0 for a failed unit and 1 otherwise.
+`daemon-reload` returns 0 without an action.
+
+## Journal
+
+Init drains each unit's stdout and stderr into `/var/log/journal/<unit>.log`.
+The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix.
+`journalctl -u <unit> -n <count>` prints the last lines.
+`journalctl -f` observes the journal file and prints appended bytes.
+
+## Process observation
+
+`sausage-spawn` exposes `Watcher`, `ChangeKind`, and child exit pollables.
+A watcher matches each supplied path and all its descendants.
+Its pollable remains ready while changes remain pending.
+`changes()` returns and clears the pending changes.
+The queue holds at most 256 changes. An `overflow` event requires a rescan.
+Observers report successful VFS mutations, not direct host filesystem writes outside the VFS.

userland/init/src/journalctl.rs

@@ -0,0 +1,66 @@
+//! Reads a unit journal, with optional live output through the world watcher.
+use sausage_init::name;
+use std::{
+    fs,
+    io::{self, Write},
+};
+fn run() -> io::Result<()> {
+    let mut args = std::env::args().skip(1);
+    let (mut unit, mut lines, mut follow) = (String::new(), 10usize, false);
+    while let Some(arg) = args.next() {
+        match arg.as_str() {
+            "-u" => {
+                unit = name(
+                    &args
+                        .next()
+                        .ok_or_else(|| io::Error::other("missing unit"))?,
+                )
+            }
+            "-n" => {
+                lines = args
+                    .next()
+                    .and_then(|n| n.parse().ok())
+                    .ok_or_else(|| io::Error::other("missing count"))?
+            }
+            "-f" => follow = true,
+            "--no-pager" => (),
+            _ => return Err(io::Error::other("unknown argument")),
+        }
+    }
+    let path = format!("/var/log/journal/{unit}.log");
+    let watcher = sausage_spawn::Watcher::new(&[path.clone()]);
+    let bytes = fs::read(&path).unwrap_or_default();
+    let mut starts = vec![0];
+    for (i, b) in bytes.iter().enumerate() {
+        if *b == b'\n' && i + 1 < bytes.len() {
+            starts.push(i + 1);
+        }
+    }
+    let start = if lines == 0 {
+        bytes.len()
+    } else {
+        starts[starts.len().saturating_sub(lines)]
+    };
+    io::stdout().write_all(&bytes[start..])?;
+    io::stdout().flush()?;
+    let mut offset = bytes.len();
+    while follow {
+        let pollable = watcher.subscribe();
+        pollable.block();
+        drop(pollable);
+        watcher.changes();
+        let bytes = fs::read(&path).unwrap_or_default();
+        if bytes.len() < offset {
+            offset = 0;
+        }
+        io::stdout().write_all(&bytes[offset..])?;
+        io::stdout().flush()?;
+        offset = bytes.len();
+    }
+    Ok(())
+}
+fn main() {
+    if run().is_err() {
+        std::process::exit(1);
+    }
+}

userland/init/src/lib.rs

@@ -0,0 +1,188 @@
+//! Unit definitions and the file protocol shared by init and its control tools.
+use serde::{Deserialize, Serialize};
+use std::{collections::BTreeMap, fs, io, time::Duration};
+/// WASI 0.2's legacy exit call loses numeric error codes. Use the standard extended call.
+pub fn exit(code: i32) -> ! {
+    wasi::cli::exit::exit_with_code(code as u8);
+    unreachable!("exit returned")
+}
+pub const RUN: &str = "/run/systemd";
+pub const UNITS: &str = "/etc/systemd/system";
+pub const TRANSIENT: &str = "/run/systemd/transient";
+#[derive(Clone, Debug, PartialEq)]
+pub struct Unit {
+    pub values: BTreeMap<String, String>,
+    pub source: String,
+}
+impl Unit {
+    pub fn parse(source: String) -> io::Result<Self> {
+        let mut values = BTreeMap::new();
+        let mut section = "";
+        for line in source.lines().map(str::trim) {
+            if line.is_empty() || line.starts_with(['#', ';']) {
+                continue;
+            }
+            if line.starts_with('[') && line.ends_with(']') {
+                section = &line[1..line.len() - 1];
+                continue;
+            }
+            let (key, value) = line
+                .split_once('=')
+                .ok_or_else(|| io::Error::other("expected Key=Value"))?;
+            let key = format!("{section}.{}", key.trim());
+            if matches!(
+                key.as_str(),
+                "Unit.After"
+                    | "Unit.Before"
+                    | "Unit.Requires"
+                    | "Unit.Wants"
+                    | "Service.Environment"
+            ) {
+                values
+                    .entry(key)
+                    .and_modify(|v: &mut String| {
+                        v.push(' ');
+                        v.push_str(value.trim());
+                    })
+                    .or_insert(value.trim().into());
+            } else {
+                values.insert(key, value.trim().into());
+            }
+        }
+        if !values.contains_key("Service.ExecStart") {
+            return Err(io::Error::other("missing ExecStart"));
+        }
+        Ok(Self { values, source })
+    }
+    pub fn get(&self, key: &str) -> &str {
+        self.values.get(key).map(String::as_str).unwrap_or("")
+    }
+    pub fn words(&self, key: &str) -> Vec<String> {
+        self.get(key).split_whitespace().map(name).collect()
+    }
+    pub fn seconds(&self, key: &str, default: f64) -> Duration {
+        Duration::from_secs_f64(
+            self.get(key)
+                .trim_end_matches('s')
+                .parse::<f64>()
+                .ok()
+                .filter(|n| n.is_finite() && *n >= 0.0)
+                .unwrap_or(default),
+        )
+    }
+}
+pub fn name(s: &str) -> String {
+    if s.ends_with(".service") || s.ends_with(".target") {
+        s.into()
+    } else {
+        format!("{s}.service")
+    }
+}
+pub fn unit_path(name: &str) -> String {
+    let persistent = format!("{UNITS}/{name}");
+    if std::path::Path::new(&persistent).exists() {
+        persistent
+    } else {
+        format!("{TRANSIENT}/{name}")
+    }
+}
+pub fn enabled(name: &str) -> bool {
+    ["default.target", "multi-user.target"]
+        .iter()
+        .any(|target| std::path::Path::new(&format!("{UNITS}/{target}.wants/{name}")).exists())
+}
+#[derive(Serialize, Deserialize, Default)]
+pub struct Status {
+    pub state: String,
+    pub pid: Option<u32>,
+    pub code: Option<i32>,
+}
+#[derive(Serialize, Deserialize)]
+pub struct Request {
+    pub action: String,
+    pub unit: String,
+}
+#[derive(Serialize, Deserialize)]
+pub struct Reply {
+    pub code: i32,
+    pub text: String,
+}
+pub fn atomic(path: &str, bytes: &[u8]) -> io::Result<()> {
+    let tmp = format!("{path}.tmp");
+    fs::write(&tmp, bytes)?;
+    fs::rename(tmp, path)
+}
+pub fn request(action: &str, unit: &str) -> io::Result<Reply> {
+    let id = format!(
+        "{}",
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    );
+    let reply = format!("{RUN}/replies/{id}");
+    let state = format!("{RUN}/state/{}", name(unit));
+    let watcher = sausage_spawn::Watcher::new(&[reply.clone(), state.clone()]);
+    atomic(
+        &format!("{RUN}/requests/{id}.json"),
+        &serde_json::to_vec(&Request {
+            action: action.into(),
+            unit: name(unit),
+        })?,
+    )?;
+    let deadline = std::time::Instant::now() + Duration::from_secs(60);
+    let mut acknowledged = false;
+    loop {
+        if !acknowledged && let Ok(bytes) = fs::read(&reply) {
+            fs::remove_file(&reply)?;
+            let reply: Reply = serde_json::from_slice(&bytes)?;
+            if reply.code != 0 {
+                return Ok(reply);
+            }
+            acknowledged = true;
+        }
+        if acknowledged && let Ok(bytes) = fs::read(&state) {
+            let status: Status = serde_json::from_slice(&bytes)?;
+            let done = if action == "stop" {
+                status.state == "inactive"
+            } else {
+                status.state == "active"
+                    || status.state == "failed"
+                    || status.state == "inactive" && status.code.is_some()
+            };
+            if done {
+                return Ok(Reply {
+                    code: if action != "stop" && status.state == "failed" {
+                        1
+                    } else {
+                        0
+                    },
+                    text: String::new(),
+                });
+            }
+        }
+        if std::time::Instant::now() >= deadline {
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "init did not reply",
+            ));
+        }
+        // The short timer also handles a missing or failed init.
+        let _ = watcher.changes();
+        std::thread::sleep(Duration::from_millis(20));
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn unit_sections_and_defaults() {
+        let unit = Unit::parse("[Unit]\nAfter=prepare\nRequires=prepare\n[Service]\nType=oneshot\nExecStart=/bin/task.wasm\n".into()).unwrap();
+        assert_eq!(unit.words("Unit.Requires"), ["prepare.service"]);
+        assert_eq!(
+            unit.seconds("Service.TimeoutStopSec", 10.),
+            Duration::from_secs(10)
+        );
+        assert!(Unit::parse("[Unit]\nDescription=missing command".into()).is_err());
+    }
+}

userland/init/src/main.rs

@@ -0,0 +1,421 @@
+//! Runs the world's unit state machine, journal drains, and file-based control requests.
+use sausage_init::*;
+use sausage_spawn::{Child, Command, Stdio};
+use std::{
+    collections::{BTreeMap, BTreeSet, VecDeque},
+    fs,
+    io::{self, Write},
+    time::{Duration, Instant},
+};
+struct Service {
+    unit: Unit,
+    child: Option<Child>,
+    state: String,
+    code: Option<i32>,
+    desired: bool,
+    starts: VecDeque<Instant>,
+    next: Instant,
+    stopping: Option<Instant>,
+}
+impl Service {
+    fn new(unit: Unit) -> Self {
+        Self {
+            unit,
+            child: None,
+            state: "inactive".into(),
+            code: None,
+            desired: false,
+            starts: VecDeque::new(),
+            next: Instant::now(),
+            stopping: None,
+        }
+    }
+    fn stop(&mut self) {
+        self.desired = false;
+        if let Some(child) = &mut self.child {
+            child.stdin.take();
+            self.stopping
+                .get_or_insert(Instant::now() + self.unit.seconds("Service.TimeoutStopSec", 10.));
+            self.state = "deactivating".into();
+        } else {
+            self.state = "inactive".into();
+        }
+    }
+    fn start(&mut self) {
+        let now = Instant::now();
+        let interval = self.unit.seconds("Service.StartLimitIntervalSec", 10.);
+        while self
+            .starts
+            .front()
+            .is_some_and(|t| now.duration_since(*t) >= interval)
+        {
+            self.starts.pop_front();
+        }
+        let burst = self
+            .unit
+            .get("Service.StartLimitBurst")
+            .parse()
+            .unwrap_or(5);
+        if burst != 0 && self.starts.len() >= burst {
+            self.state = "failed".into();
+            self.desired = false;
+            return;
+        }
+        self.starts.push_back(now);
+        let Some(argv) = shlex::split(self.unit.get("Service.ExecStart")).filter(|a| !a.is_empty())
+        else {
+            self.state = "failed".into();
+            self.desired = false;
+            return;
+        };
+        let mut command = Command::new(&argv[0]);
+        command
+            .args(&argv[1..])
+            .new_group()
+            .stdin(Stdio::Pipe)
+            .stdout(Stdio::Pipe)
+            .stderr(Stdio::Pipe);
+        let cwd = self.unit.get("Service.WorkingDirectory");
+        command.current_dir(if cwd.is_empty() { "/" } else { cwd });
+        for assignment in shlex::split(self.unit.get("Service.Environment")).unwrap_or_default() {
+            if let Some((key, value)) = assignment.split_once('=') {
+                command.env(key, value);
+            }
+        }
+        match command.spawn() {
+            Ok(child) => {
+                self.child = Some(child);
+                self.state = if self.unit.get("Service.Type") == "oneshot" {
+                    "activating"
+                } else {
+                    "active"
+                }
+                .into();
+                self.code = None;
+            }
+            Err(_) => {
+                self.code = Some(127);
+                self.exited(127);
+            }
+        }
+    }
+    fn exited(&mut self, code: i32) {
+        self.code = Some(code);
+        if self.stopping.take().is_some() {
+            self.state = if self.desired {
+                "activating"
+            } else {
+                "inactive"
+            }
+            .into();
+            if self.desired {
+                self.code = None;
+            }
+            return;
+        }
+        self.state = if code != 0 {
+            "failed"
+        } else if matches!(
+            self.unit.get("Service.RemainAfterExit"),
+            "yes" | "true" | "1"
+        ) {
+            "active"
+        } else {
+            "inactive"
+        }
+        .into();
+        let restart = self.unit.get("Service.Restart");
+        if restart == "always" || restart == "on-failure" && code != 0 {
+            self.next = Instant::now() + self.unit.seconds("Service.RestartSec", 0.1);
+        } else {
+            self.desired = false;
+        }
+    }
+    fn tick(&mut self, name: &str) {
+        if let Some(child) = &mut self.child {
+            for slot in [&mut child.stdout, &mut child.stderr] {
+                if let Some(stream) = slot {
+                    loop {
+                        match stream.read(8192) {
+                            Ok(bytes) if !bytes.is_empty() => journal(name, &bytes),
+                            Ok(_) => break,
+                            Err(_) => {
+                                *slot = None;
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+            if child.try_wait().is_none() && self.stopping.is_some_and(|t| Instant::now() >= t) {
+                let _ = sausage_spawn::kill(child.id());
+            }
+            if let Some(code) = child.try_wait() {
+                if child.stdout.is_none() && child.stderr.is_none() {
+                    self.child = None;
+                    self.exited(code);
+                }
+            }
+        }
+    }
+    fn publish(&self, name: &str) {
+        let status = Status {
+            state: self.state.clone(),
+            pid: self.child.as_ref().map(Child::id),
+            code: self.code,
+        };
+        let bytes = serde_json::to_vec(&status).unwrap();
+        let path = format!("{RUN}/state/{name}");
+        if fs::read(&path).ok().as_deref() != Some(bytes.as_slice()) {
+            let _ = atomic(&path, &bytes);
+        }
+    }
+}
+fn journal(name: &str, bytes: &[u8]) {
+    let path = format!("/var/log/journal/{name}.log");
+    if fs::metadata(&path).is_ok_and(|m| m.len() + bytes.len() as u64 > 1024 * 1024) {
+        let _ = fs::rename(&path, format!("{path}.1"));
+    }
+    if let Ok(mut f) = fs::OpenOptions::new().create(true).append(true).open(path) {
+        let _ = f.write_all(bytes);
+    }
+}
+fn activate(name: &str, services: &mut BTreeMap<String, Service>, seen: &mut BTreeSet<String>) {
+    if !seen.insert(name.into()) {
+        return;
+    }
+    let Some(service) = services.get(name) else {
+        return;
+    };
+    let deps = [
+        service.unit.words("Unit.Requires"),
+        service.unit.words("Unit.Wants"),
+    ]
+    .concat();
+    for dep in deps {
+        activate(&dep, services, seen);
+    }
+    let service = services.get_mut(name).unwrap();
+    service.desired = true;
+    if service.child.is_none() {
+        service.state = "activating".into();
+        service.code = None;
+    }
+    service.next = Instant::now();
+}
+fn predecessors(name: &str, services: &BTreeMap<String, Service>) -> Vec<String> {
+    let mut deps = services[name].unit.words("Unit.After");
+    deps.extend(services[name].unit.words("Unit.Requires"));
+    for (other, service) in services {
+        if service.unit.words("Unit.Before").iter().any(|s| s == name) {
+            deps.push(other.clone());
+        }
+    }
+    deps
+}
+fn scan(services: &mut BTreeMap<String, Service>) {
+    let mut found = BTreeMap::new();
+    let mut present = BTreeSet::new();
+    for dir in [TRANSIENT, UNITS] {
+        for entry in fs::read_dir(dir).into_iter().flatten().flatten() {
+            let name = entry.file_name().to_string_lossy().into_owned();
+            if !name.ends_with(".service") {
+                continue;
+            }
+            present.insert(name.clone());
+            if let Ok(source) = fs::read_to_string(entry.path()) {
+                if let Ok(unit) = Unit::parse(source) {
+                    found.insert(name, unit);
+                }
+            }
+        }
+    }
+    for (name, service) in services.iter_mut() {
+        if !present.contains(name) {
+            service.stop();
+        }
+    }
+    services.retain(|name, service| present.contains(name) || service.child.is_some());
+    let mut new_enabled = Vec::new();
+    for (name, unit) in found {
+        if !services.contains_key(&name) && enabled(&name) {
+            new_enabled.push(name.clone());
+        }
+        let service = services
+            .entry(name.clone())
+            .or_insert_with(|| Service::new(unit.clone()));
+        if service.unit != unit {
+            let desired = service.desired || service.state == "active";
+            service.stop();
+            service.unit = unit;
+            service.desired = desired;
+            service.starts.clear();
+            service.next = Instant::now();
+        }
+    }
+    for name in new_enabled {
+        activate(&name, services, &mut BTreeSet::new());
+    }
+}
+fn controls(services: &mut BTreeMap<String, Service>) {
+    for entry in fs::read_dir(format!("{RUN}/requests"))
+        .into_iter()
+        .flatten()
+        .flatten()
+    {
+        let filename = entry.file_name().to_string_lossy().into_owned();
+        if !filename.ends_with(".json") {
+            continue;
+        }
+        let Ok(bytes) = fs::read(entry.path()) else {
+            continue;
+        };
+        let Ok(request) = serde_json::from_slice::<Request>(&bytes) else {
+            continue;
+        };
+        let mut reply = Reply {
+            code: 0,
+            text: String::new(),
+        };
+        if !services.contains_key(&request.unit) {
+            reply.code = 5;
+            reply.text = "unit not found".into();
+        } else {
+            match request.action.as_str() {
+                "start" => {
+                    services.get_mut(&request.unit).unwrap().starts.clear();
+                    activate(&request.unit, services, &mut BTreeSet::new());
+                }
+                "stop" => services.get_mut(&request.unit).unwrap().stop(),
+                "restart" => {
+                    let service = services.get_mut(&request.unit).unwrap();
+                    service.stop();
+                    service.desired = true;
+                    if service.child.is_none() {
+                        service.state = "activating".into();
+                        service.code = None;
+                    }
+                    service.starts.clear();
+                    service.next = Instant::now();
+                }
+                _ => {
+                    reply.code = 1;
+                    reply.text = "unknown action".into();
+                }
+            }
+        }
+        if let Some(service) = services.get(&request.unit) {
+            service.publish(&request.unit);
+        }
+        let id = filename.trim_end_matches(".json");
+        let _ = atomic(
+            &format!("{RUN}/replies/{id}"),
+            &serde_json::to_vec(&reply).unwrap(),
+        );
+        let _ = fs::remove_file(entry.path());
+    }
+}
+fn main() -> io::Result<()> {
+    for dir in [
+        UNITS,
+        TRANSIENT,
+        "/run/systemd/requests",
+        "/run/systemd/replies",
+        "/run/systemd/state",
+        "/var/log/journal",
+    ] {
+        fs::create_dir_all(dir)?;
+    }
+    let watcher = sausage_spawn::Watcher::new(&["/".into()]);
+    let stdin = wasi::cli::stdin::get_stdin();
+    let mut services = BTreeMap::new();
+    scan(&mut services);
+    loop {
+        if stdin.read(1).is_err() {
+            break;
+        }
+        let changes = watcher.changes();
+        if !changes.is_empty() {
+            for service in services.values_mut() {
+                let program = shlex::split(service.unit.get("Service.ExecStart"))
+                    .and_then(|a| a.first().cloned())
+                    .unwrap_or_default();
+                if service.child.is_some()
+                    && changes.iter().any(|c| {
+                        c.path == program || matches!(c.kind, sausage_spawn::ChangeKind::Overflow)
+                    })
+                {
+                    service.stop();
+                    service.desired = true;
+                    service.next = Instant::now();
+                }
+            }
+            scan(&mut services);
+        }
+        controls(&mut services);
+        let names: Vec<_> = services.keys().cloned().collect();
+        for name in &names {
+            services.get_mut(name).unwrap().tick(name);
+        }
+        for name in &names {
+            let service = &services[name];
+            if !service.desired
+                || service.child.is_some()
+                || Instant::now() < service.next
+                || service.stopping.is_some()
+            {
+                continue;
+            }
+            let required = service.unit.words("Unit.Requires");
+            if required
+                .iter()
+                .any(|dep| !services.contains_key(dep) || services[dep].state == "failed")
+            {
+                let service = services.get_mut(name).unwrap();
+                service.state = "failed".into();
+                service.desired = false;
+                continue;
+            }
+            if predecessors(name, &services).iter().any(|dep| {
+                services.get(dep).is_some_and(|s| {
+                    s.state == "activating" || s.desired && s.child.is_none() && s.code.is_none()
+                })
+            }) {
+                continue;
+            }
+            services.get_mut(name).unwrap().start();
+        }
+        for (name, service) in &services {
+            service.publish(name);
+        }
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    // Stop dependents before their prerequisites. A cycle stops together.
+    while services.values().any(|s| s.child.is_some()) {
+        let live: Vec<_> = services
+            .iter()
+            .filter(|(_, s)| s.child.is_some())
+            .map(|(n, _)| n.clone())
+            .collect();
+        let mut leaves: Vec<_> = live
+            .iter()
+            .filter(|name| {
+                !live
+                    .iter()
+                    .any(|other| other != *name && predecessors(other, &services).contains(name))
+            })
+            .cloned()
+            .collect();
+        if leaves.is_empty() {
+            leaves = live;
+        }
+        for name in leaves {
+            let service = services.get_mut(&name).unwrap();
+            service.stop();
+            service.tick(&name);
+            service.publish(&name);
+        }
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    Ok(())
+}

userland/init/src/probe.rs

@@ -0,0 +1,96 @@
+//! Acceptance guest for unit lifecycle and process-interface readiness tests.
+use std::{
+    fs,
+    io::{Read, Write},
+};
+fn main() {
+    let args: Vec<_> = std::env::args().collect();
+    match args.get(1).map(String::as_str).unwrap_or("") {
+        "kill-init" => {
+            let table = fs::read_to_string("/proc/processes").unwrap();
+            let row = table
+                .lines()
+                .find(|line| line.ends_with("/bin/init.wasm"))
+                .unwrap();
+            sausage_spawn::kill(row.split('\t').next().unwrap().parse().unwrap()).unwrap();
+        }
+        "watch" => {
+            use sausage_spawn::{ChangeKind, Watcher};
+            let watcher = Watcher::new(&["/run/watched".into()]);
+            let poll = watcher.subscribe();
+            assert!(!poll.ready());
+            fs::create_dir_all("/run/watched").unwrap();
+            poll.block();
+            assert!(
+                watcher
+                    .changes()
+                    .iter()
+                    .any(|c| matches!(c.kind, ChangeKind::Created))
+            );
+            fs::write("/run/watched/a", "a").unwrap();
+            watcher.changes();
+            fs::write("/run/watched/a", "b").unwrap();
+            assert!(
+                watcher
+                    .changes()
+                    .iter()
+                    .any(|c| c.path == "/run/watched/a" && matches!(c.kind, ChangeKind::Modified))
+            );
+            fs::remove_file("/run/watched/a").unwrap();
+            assert!(
+                watcher
+                    .changes()
+                    .iter()
+                    .any(|c| matches!(c.kind, ChangeKind::Removed))
+            );
+            for i in 0..300 {
+                fs::write(format!("/run/watched/{i}"), "").unwrap();
+            }
+            assert!(
+                watcher
+                    .changes()
+                    .iter()
+                    .any(|c| matches!(c.kind, ChangeKind::Overflow))
+            );
+            drop(poll);
+            let mut cmd = sausage_spawn::Command::new("/bin/probe.wasm");
+            cmd.arg("exit").arg("7");
+            let child = cmd.spawn().unwrap();
+            let exit = child.subscribe();
+            exit.block();
+            assert_eq!(child.try_wait(), Some(7));
+        }
+        "exit" => sausage_init::exit(args.get(2).and_then(|a| a.parse().ok()).unwrap_or(0)),
+        "oneshot" | "server" | "crash" | "after" | "linger" => {
+            if args[1] == "after" {
+                assert!(
+                    !fs::read_to_string("/proc/processes")
+                        .unwrap()
+                        .contains("oneshot /order.log")
+                );
+            }
+            let mut log = fs::OpenOptions::new()
+                .create(true)
+                .append(true)
+                .open(&args[2])
+                .unwrap();
+            writeln!(log, "start").unwrap();
+            for n in 1..=10 {
+                println!("line {n}");
+            }
+            if args[1] == "linger" {
+                loop {
+                    std::thread::sleep(std::time::Duration::from_secs(1));
+                }
+            }
+            if args[1] == "crash" {
+                std::process::exit(1);
+            }
+            if args[1] == "server" || args[1] == "after" {
+                let _ = std::io::stdin().read_to_end(&mut Vec::new());
+                writeln!(log, "stop").unwrap();
+            }
+        }
+        _ => panic!("unknown probe mode"),
+    }
+}

userland/init/src/systemctl.rs

@@ -0,0 +1,123 @@
+//! Controls init and reads its public unit state files.
+use sausage_init::*;
+use std::{fs, io};
+fn run() -> io::Result<i32> {
+    let args: Vec<_> = std::env::args().skip(1).collect();
+    let action = args.first().map(String::as_str).unwrap_or("list-units");
+    if action == "daemon-reload" {
+        return Ok(0);
+    }
+    if action == "list-units" {
+        for entry in fs::read_dir(format!("{RUN}/state"))? {
+            let entry = entry?;
+            let status: Status = serde_json::from_slice(&fs::read(entry.path())?)?;
+            println!("{} {}", entry.file_name().to_string_lossy(), status.state);
+        }
+        return Ok(0);
+    }
+    let now = args.iter().any(|a| a == "--now");
+    let units: Vec<_> = args
+        .iter()
+        .skip(1)
+        .filter(|a| !a.starts_with('-'))
+        .map(|a| name(a))
+        .collect();
+    if units.is_empty() {
+        return Ok(1);
+    }
+    let mut result = 0;
+    for unit in units {
+        let status = fs::read(format!("{RUN}/state/{unit}"))
+            .ok()
+            .and_then(|b| serde_json::from_slice::<Status>(&b).ok())
+            .unwrap_or_default();
+        let code = match action {
+            "is-active" => {
+                println!(
+                    "{}",
+                    if status.state.is_empty() {
+                        "unknown"
+                    } else {
+                        &status.state
+                    }
+                );
+                if status.state == "active" { 0 } else { 3 }
+            }
+            "is-failed" => {
+                println!(
+                    "{}",
+                    if status.state.is_empty() {
+                        "unknown"
+                    } else {
+                        &status.state
+                    }
+                );
+                if status.state == "failed" { 0 } else { 1 }
+            }
+            "is-enabled" => {
+                println!(
+                    "{}",
+                    if enabled(&unit) {
+                        "enabled"
+                    } else {
+                        "disabled"
+                    }
+                );
+                if enabled(&unit) { 0 } else { 1 }
+            }
+            "status" => {
+                println!("{unit}: {}", status.state);
+                if status.state == "active" {
+                    0
+                } else if status.state.is_empty() {
+                    4
+                } else {
+                    3
+                }
+            }
+            "cat" => {
+                print!("{}", fs::read_to_string(unit_path(&unit))?);
+                0
+            }
+            "enable" | "disable" => {
+                let spec = Unit::parse(fs::read_to_string(unit_path(&unit))?)?;
+                let targets: Vec<_> = spec
+                    .get("Install.WantedBy")
+                    .split_whitespace()
+                    .filter(|t| matches!(*t, "default.target" | "multi-user.target"))
+                    .collect();
+                if action == "enable" && targets.is_empty() {
+                    return Ok(1);
+                }
+                for target in ["default.target", "multi-user.target"] {
+                    let dir = format!("{UNITS}/{target}.wants");
+                    let path = format!("{dir}/{unit}");
+                    if action == "enable" && targets.contains(&target) {
+                        fs::create_dir_all(dir)?;
+                        fs::write(path, b"")?;
+                    } else if action == "disable" {
+                        let _ = fs::remove_file(path);
+                    }
+                }
+                if now {
+                    request(if action == "enable" { "start" } else { "stop" }, &unit)?.code
+                } else {
+                    0
+                }
+            }
+            "start" | "stop" | "restart" => {
+                let reply = request(action, &unit)?;
+                if !reply.text.is_empty() {
+                    println!("{}", reply.text);
+                }
+                reply.code
+            }
+            _ => 1,
+        };
+        result = result.max(code);
+    }
+    Ok(result)
+}
+fn main() {
+    sausage_init::exit(run().unwrap_or(1));
+}

userland/init/src/systemd-run.rs

@@ -0,0 +1,82 @@
+//! Creates transient unit files and optionally waits for their final state.
+use sausage_init::*;
+use std::{fs, io, time::Duration};
+fn run() -> io::Result<i32> {
+    let mut args = std::env::args().skip(1).peekable();
+    let mut unit = format!(
+        "run-{}",
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    );
+    let (mut wait, mut properties, mut command) = (false, Vec::new(), Vec::new());
+    while let Some(arg) = args.next() {
+        match arg.as_str() {
+            "--unit" => {
+                unit = args
+                    .next()
+                    .ok_or_else(|| io::Error::other("missing unit"))?
+            }
+            "--wait" => wait = true,
+            "-p" => properties.push(
+                args.next()
+                    .ok_or_else(|| io::Error::other("missing property"))?,
+            ),
+            "--working-directory" => properties.push(format!(
+                "WorkingDirectory={}",
+                args.next()
+                    .ok_or_else(|| io::Error::other("missing directory"))?
+            )),
+            "-E" => properties.push(format!(
+                "Environment={}",
+                args.next()
+                    .ok_or_else(|| io::Error::other("missing environment"))?
+            )),
+            "--" => {
+                command.extend(args);
+                break;
+            }
+            _ if arg.starts_with("--unit=") => unit = arg[7..].into(),
+            _ if arg.starts_with('-') => return Ok(1),
+            _ => {
+                command.push(arg);
+                command.extend(args);
+                break;
+            }
+        }
+    }
+    if command.is_empty() {
+        return Ok(1);
+    }
+    let unit = name(&unit);
+    let mut text = format!(
+        "[Service]\nExecStart={}\n",
+        shlex::try_join(command.iter().map(String::as_str)).map_err(io::Error::other)?
+    );
+    for property in properties {
+        text.push_str(&property);
+        text.push('\n');
+    }
+    fs::create_dir_all(TRANSIENT)?;
+    atomic(&format!("{TRANSIENT}/{unit}"), text.as_bytes())?;
+    let reply = request("start", &unit)?;
+    if reply.code != 0 {
+        return Ok(reply.code);
+    }
+    if wait {
+        loop {
+            if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) {
+                let status: Status = serde_json::from_slice(&bytes)?;
+                if let Some(code) = status.code {
+                    return Ok(code);
+                }
+            }
+            std::thread::sleep(Duration::from_millis(20));
+        }
+    }
+    Ok(0)
+}
+fn main() {
+    sausage_init::exit(run().unwrap_or(1));
+}

userland/spawn/src/lib.rs

@@ -7,7 +7,10 @@ mod bindings {
 }
 use bindings::sausage::process::host;
 pub use bindings::wasi::io::streams::{InputStream, OutputStream};
-pub use host::Stdio;
+pub use host::{Stdio, Watcher, Change, ChangeKind};
+pub use bindings::wasi::io::poll::{Pollable, poll};
+/// Hard-stop a process group by a live process ID.
+pub fn kill(pid: u32) -> io::Result<()> { host::kill(pid).map_err(io::Error::other) }
 
 /// A prepared child process.
 pub struct Command {
@@ -63,6 +66,7 @@ pub struct Child {
 }
 impl Child {
     pub fn id(&self) -> u32 { self.child.pid() }
+    pub fn subscribe(&self) -> Pollable { self.child.subscribe() }
     pub fn try_wait(&self) -> Option<i32> { self.child.status() }
     pub fn wait(&self) -> i32 { self.child.wait() }
     /// Stop every process in this child's group.

vfs/src/lib.rs

@@ -67,6 +67,10 @@ pub trait File: Send + Sync {
 /// Implementations must retain open file identity and return portable IO errors.
 #[async_trait]
 pub trait Layer: Send + Sync {
+    /// Shared observation of mutations in this namespace, when available.
+    fn observer(&self) -> Option<observe::Observer> {
+        None
+    }
     async fn stat(&self, path: &str, follow: bool) -> Result<Stat>;
     /// Return the target of a symbolic link, or InvalidInput for another node.
     async fn readlink(&self, _path: &str) -> Result<String> {

vfs/src/mounts.rs

@@ -6,15 +6,24 @@ struct Mount {
     readonly: bool,
 }
 /// A per-process mount namespace. Mount the root before use.
-#[derive(Default)]
 pub struct Mounts {
     mounts: BTreeMap<String, Mount>,
+    observer: observe::Observer,
+}
+impl Default for Mounts {
+    fn default() -> Self {
+        Self {
+            mounts: BTreeMap::new(),
+            observer: observe::Observer::new(|_| {}),
+        }
+    }
 }
 impl Mounts {
     /// Add or replace a mount. Read-only mounts reject every mutation.
     pub fn mount(&mut self, path: &str, layer: Arc<dyn Layer>, readonly: bool) {
-        self.mounts
-            .insert(resolve("/", path), Mount { layer, readonly });
+        let path = resolve("/", path);
+        let layer = self.observer.layer(layer, &path, Process::default());
+        self.mounts.insert(path, Mount { layer, readonly });
     }
     pub(crate) async fn metadata_file(&self, path: &str, follow: bool) -> Result<Arc<dyn File>> {
         let (m, p) = self.route(path, true)?;
@@ -49,6 +58,9 @@ impl Mounts {
 }
 #[async_trait]
 impl Layer for Mounts {
+    fn observer(&self) -> Option<observe::Observer> {
+        Some(self.observer.clone())
+    }
     async fn stat(&self, path: &str, follow: bool) -> Result<Stat> {
         let (m, p) = self.route(path, false)?;
         m.layer.stat(&p, follow).await

vfs/src/observe.rs

@@ -6,6 +6,14 @@ use std::{
 };
 use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};
 
+/// The successful namespace or content mutation observed at a world path.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ChangeKind {
+    Created,
+    Modified,
+    Removed,
+}
+
 type Namespace = Mutex<BTreeMap<String, Weak<RwLock<()>>>>;
 enum Held {
     Read { _guard: OwnedRwLockReadGuard<()> },
@@ -20,6 +28,7 @@ type Callback = dyn Fn(&str) + Send + Sync;
 #[derive(Clone)]
 pub struct Observer {
     changed: Arc<Callback>,
+    events: tokio::sync::broadcast::Sender<(String, ChangeKind)>,
     paths: Arc<Mutex<Vec<Weak<TrackedPath>>>>,
     // Only open-registration versus rename/remove needs ordering. Lock the affected
     // subtrees, not all IO: a blocked open must not stall unrelated namespaces.
@@ -29,10 +38,19 @@ impl Observer {
     pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self {
         Self {
             changed: Arc::new(changed),
+            events: tokio::sync::broadcast::channel(256).0,
             paths: Arc::default(),
             namespace: Arc::default(),
         }
     }
+    /// Subscribe before the first mutation of interest. Lag means callers must rescan.
+    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> {
+        self.events.subscribe()
+    }
+    fn notify(&self, path: &str, kind: ChangeKind) {
+        (self.changed)(path);
+        let _ = self.events.send((path.to_owned(), kind));
+    }
     /// Wrap a layer at its canonical namespace prefix (e.g. `/run`).
     pub fn layer(&self, layer: Arc<dyn Layer>, prefix: &str, process: Process) -> Arc<dyn Layer> {
         Arc::new(Observed {
@@ -149,7 +167,7 @@ impl Handle {
                 let result = f.await?;
                 let path = path.lock().unwrap().clone();
                 if let Some(path) = path {
-                    (observer.changed)(&path);
+                    observer.notify(&path, ChangeKind::Modified);
                 }
                 Ok(result)
             })
@@ -178,6 +196,7 @@ impl Layer for Observed {
         self.process
             .operation(async move {
                 let order = observer.order(&[(&path, false)]).await;
+                let existed = !options.create || layer.stat(&local, options.follow).await.is_ok();
                 let file = layer.open(&local, options).await?;
                 let handle = Arc::new(Handle {
                     file,
@@ -187,7 +206,14 @@ impl Layer for Observed {
                 });
                 drop(order);
                 if options.create || options.truncate {
-                    (observer.changed)(&path);
+                    observer.notify(
+                        &path,
+                        if existed {
+                            ChangeKind::Modified
+                        } else {
+                            ChangeKind::Created
+                        },
+                    );
                 }
                 Ok(handle as Arc<dyn File>)
             })
@@ -203,7 +229,7 @@ impl Layer for Observed {
         self.process
             .operation(async move {
                 layer.mkdir(&local).await?;
-                (observer.changed)(&path);
+                observer.notify(&path, ChangeKind::Created);
                 Ok(())
             })
             .await
@@ -226,13 +252,22 @@ impl Layer for Observed {
         self.process
             .operation(async move {
                 let order = observer.order(&[(&from, true), (&to, true)]).await;
+                let replaced = layer.stat(&b, false).await.is_ok();
                 layer.rename(&a, &b).await?;
-                if from != to {
-                    observer.moved(&from, Some(&to));
+                if from == to {
+                    return Ok(());
                 }
+                observer.moved(&from, Some(&to));
                 drop(order);
-                (observer.changed)(&from);
-                (observer.changed)(&to);
+                observer.notify(&from, ChangeKind::Removed);
+                observer.notify(
+                    &to,
+                    if replaced {
+                        ChangeKind::Modified
+                    } else {
+                        ChangeKind::Created
+                    },
+                );
                 Ok(())
             })
             .await
@@ -256,7 +291,7 @@ impl Observed {
                 }
                 observer.moved(&path, None);
                 drop(order);
-                (observer.changed)(&path);
+                observer.notify(&path, ChangeKind::Removed);
                 Ok(())
             })
             .await