Add Membrane RPC bootstrap with epoch-scoped capabilities

Cargo.toml

@@ -45,6 +45,7 @@ wasmtime-wasi = "40.0.0"
 wasmtime-wasi-io = "40.0.0"
 capnp = "0.23.2"
 capnp-rpc = "0.23.0"
+stem = { path = "../stem/crates/stem" }
 
 [[bin]]
 name = "ww"

build.rs

@@ -6,7 +6,9 @@ fn main() {
     let manifest_dir = env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set");
     let target_dir = Path::new(&manifest_dir).join("target");
     let cid_file = target_dir.join("default-config.cid");
-    let capnp_file = Path::new(&manifest_dir).join("capnp").join("peer.capnp");
+    let capnp_dir = Path::new(&manifest_dir).join("capnp");
+    let capnp_file = capnp_dir.join("peer.capnp");
+    let membrane_file = capnp_dir.join("membrane.capnp");
 
     // Read CID from the generated .cid file in target directory
     let cid_value = if cid_file.exists() {
@@ -39,7 +41,10 @@ fn main() {
 
     capnpc::CompilerCommand::new()
         .file(&capnp_file)
+        .file(&membrane_file)
+        .crate_provides("stem", [0x9bce094a026970c4_u64]) // stem.capnp types live in the stem crate
         .run()
-        .expect("failed to compile capnp schema");
+        .expect("failed to compile capnp schemas");
     println!("cargo:rerun-if-changed={}", capnp_file.display());
+    println!("cargo:rerun-if-changed={}", membrane_file.display());
 }

capnp/membrane.capnp

@@ -0,0 +1,23 @@
+# Wetware session extension for stem's generic Membrane.
+#
+# The RPC bootstrap capability for wetware guests is
+# Stem.Membrane(WetwareSession).  Guests call graft() and receive a
+# Session whose extension field carries epoch-scoped Host and Executor
+# capabilities.  When the on-chain epoch advances, all capabilities from
+# the previous session are revoked.
+
+@0xa59e04af26eca82f;
+
+using Stem = import "stem.capnp";
+# Vendored copy; no code generated (see crate_provides in build.rs).
+
+using Peer = import "peer.capnp";
+# Local peer interfaces: Host, Executor, Process, ByteStream.
+
+struct WetwareSession {
+  host     @0 :Peer.Host;     # Swarm-level operations (id, addrs, peers, connect).
+  executor @1 :Peer.Executor; # WASM execution (runBytes, echo).
+}
+
+# The bootstrap type for wetware guests is Stem.Membrane(WetwareSession).
+# No new interface needed — we use stem's generic Membrane directly.

capnp/peer.capnp

@@ -1,32 +1,64 @@
+# Wetware peer interfaces.
+#
+# These capabilities are surfaced to WASM guests through the Membrane's
+# epoch-scoped session (see membrane.capnp).  Each capability wrapper
+# holds an EpochGuard and fails with a stale-epoch error once the guard
+# detects the epoch has advanced.
+
 @0xbf5147b78c0e6a2f;
 
 struct PeerInfo {
-  peerId @0 :Data;
-  addrs @1 :List(Data);
+  peerId @0 :Data;       # libp2p peer ID, serialized.
+  addrs @1 :List(Data);  # Multiaddrs for this peer, each serialized.
 }
 
 interface Host {
   id @0 () -> (peerId :Data);
+  # Return this node's libp2p peer ID.
+
   addrs @1 () -> (addrs :List(Data));
+  # Return the multiaddrs this node is listening on.
+
   peers @2 () -> (peers :List(PeerInfo));
+  # List currently connected peers.
+
   connect @3 (peerId :Data, addrs :List(Data)) -> ();
+  # Dial a peer by ID and multiaddrs.
+
   executor @4 () -> (executor :Executor);
+  # Obtain an Executor scoped to the same epoch as this Host.
 }
 
 interface Executor {
   runBytes @0 (wasm :Data, args :List(Text), env :List(Text)) -> (process :Process);
+  # Instantiate a WASM component from raw bytes and return a handle to
+  # its running process.
+
   echo @1 (message :Text) -> (response :Text);
+  # Diagnostic echo — returns the message unmodified.
 }
 
 interface Process {
   stdin @0 () -> (stream :ByteStream);
+  # Writable stream connected to the guest's standard input.
+
   stdout @1 () -> (stream :ByteStream);
+  # Readable stream connected to the guest's standard output.
+
   stderr @2 () -> (stream :ByteStream);
+  # Readable stream connected to the guest's standard error.
+
   wait @3 () -> (exitCode :Int32);
+  # Block until the process exits and return its exit code.
 }
 
 interface ByteStream {
   read @0 (maxBytes :UInt32) -> (data :Data);
+  # Read up to maxBytes from the stream.  Returns empty data at EOF.
+
   write @1 (data :Data) -> ();
+  # Write data to the stream.
+
   close @2 () -> ();
+  # Close the stream.  Further reads return EOF; further writes fail.
 }

capnp/stem.capnp

@@ -1,8 +1,14 @@
+# Vendored from github.com/wetware/stem.  This file MUST stay in sync with
+# stem's canonical copy.  No Rust code is generated for it — build.rs uses
+# capnpc::CompilerCommand::crate_provides("stem", [...]) so that downstream
+# schemas can import it for type resolution while referencing the stem crate's
+# generated types at compile time.  See doc/capnp-cross-crate.md.
+
 @0x9bce094a026970c4;
 
 struct Epoch {
-  seq @0 :UInt64;        # Monotonic epoch sequence number (from Atom.seq).
-  head @1 :Data;         # Opaque head bytes from the Atom contract.
+  seq @0 :UInt64;        # Monotonic epoch sequence number (from Stem.seq).
+  head @1 :Data;         # Opaque head bytes from the Stem contract.
   adoptedBlock @2 :UInt64;# Block number at which this epoch was adopted.
 }
 

doc/capnp-cross-crate.md

@@ -0,0 +1,76 @@
+# Cross-Crate Cap'n Proto Schema Sharing
+
+## Problem
+
+capnpc generates Rust trait types (e.g. `Server`, `Client`) per crate. If two
+crates both compile the same `.capnp` file, their generated traits are
+**distinct types** — a struct implementing crate A's `Server` trait does not
+satisfy crate B's `Server` trait, even though the schema is identical.
+
+Concretely: `stem` compiles `stem.capnp` and exports `MembraneServer` which
+implements `stem::stem_capnp::membrane::Server`. If `rs` also compiles
+`stem.capnp`, it gets its own `rs::stem_capnp::membrane::Server` — and stem's
+`MembraneServer` doesn't implement it.
+
+## Solution: `crate_provides`
+
+capnpc (≥ 0.17.2) has `CompilerCommand::crate_provides(crate_name, file_ids)`.
+This tells the code generator: "don't generate code for these schema files;
+instead, emit `use` statements that reference the named crate's generated
+modules."
+
+```rust
+// rs/build.rs
+capnpc::CompilerCommand::new()
+    .file("capnp/peer.capnp")
+    .file("capnp/membrane.capnp")   // imports stem.capnp
+    .crate_provides("stem", [0x9bce094a026970c4]) // stem.capnp file ID
+    .run()
+    .expect("failed to compile capnp schemas");
+```
+
+The file ID is the `@0x...` annotation at the top of each `.capnp` file:
+
+```capnp
+# stem.capnp
+@0x9bce094a026970c4;
+```
+
+## Requirements
+
+1. **The `.capnp` file must still be on disk.** capnpc needs it for import
+   resolution when other schemas (e.g. `membrane.capnp`) reference it. We
+   vendor `stem.capnp` into `rs/capnp/` for this reason.
+
+2. **The providing crate must expose its generated module.** stem does this via:
+   ```rust
+   pub mod stem_capnp {
+       include!(concat!(env!("OUT_DIR"), "/capnp/stem_capnp.rs"));
+   }
+   ```
+
+3. **The consuming crate depends on the provider normally** (via `Cargo.toml`).
+   No special dependency features needed.
+
+## How it works in this project
+
+- **stem** compiles `stem.capnp` → generates `stem::stem_capnp` with
+  `Membrane`, `Session`, `StatusPoller`, `Epoch`, etc.
+- **rs** compiles `membrane.capnp` (which imports `stem.capnp`) but declares
+  `crate_provides("stem", ...)` for `stem.capnp`. capnpc generates code for
+  `membrane.capnp` only, with references like `stem::stem_capnp::membrane::*`.
+- rs uses `stem::membrane::MembraneServer` directly — no trait mismatch because
+  both sides share the same generated `stem::stem_capnp` types.
+
+## What NOT to do
+
+Don't compile the same `.capnp` in both crates without `crate_provides`. You'll
+get a confusing `E0277` error like:
+
+```
+the trait `stem_capnp::membrane::Server<…>` is not implemented for `MembraneServer<…>`
+note: `MembraneServer<…>` implements similarly named trait
+      `stem::stem_capnp::membrane::Server`, but not `stem_capnp::membrane::Server<…>`
+```
+
+The fix is always `crate_provides` — not reimplementing the server locally.

guests/pid0/build.rs

@@ -1,24 +1,24 @@
-// TODO: Extract a shared build-support crate so all guests use a common
-// capnp compilation helper instead of duplicating this logic.
-
 use std::env;
 use std::path::Path;
 
 fn main() {
     let manifest_dir = env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set");
     let capnp_dir = Path::new(&manifest_dir)
-        .join("../..")
+        .join("..")
+        .join("..")
         .join("capnp")
         .canonicalize()
         .expect("capnp dir not found");
 
     capnpc::CompilerCommand::new()
         .src_prefix(&capnp_dir)
         .file(capnp_dir.join("peer.capnp"))
+        .file(capnp_dir.join("membrane.capnp"))
+        .file(capnp_dir.join("stem.capnp"))
         .run()
-        .expect("failed to compile capnp schema");
-    println!(
-        "cargo:rerun-if-changed={}",
-        capnp_dir.join("peer.capnp").display()
-    );
+        .expect("failed to compile capnp schemas");
+
+    println!("cargo:rerun-if-changed={}", capnp_dir.join("peer.capnp").display());
+    println!("cargo:rerun-if-changed={}", capnp_dir.join("membrane.capnp").display());
+    println!("cargo:rerun-if-changed={}", capnp_dir.join("stem.capnp").display());
 }

guests/pid0/src/lib.rs

@@ -8,6 +8,19 @@ mod peer_capnp {
     include!(concat!(env!("OUT_DIR"), "/peer_capnp.rs"));
 }
 
+#[allow(dead_code)]
+mod stem_capnp {
+    include!(concat!(env!("OUT_DIR"), "/stem_capnp.rs"));
+}
+
+#[allow(dead_code)]
+mod membrane_capnp {
+    include!(concat!(env!("OUT_DIR"), "/membrane_capnp.rs"));
+}
+
+/// Bootstrap capability: a Membrane whose sessions carry our WetwareSession extension.
+type Membrane = stem_capnp::membrane::Client<membrane_capnp::wetware_session::Owned>;
+
 struct StderrLogger;
 
 impl log::Log for StderrLogger {
@@ -49,32 +62,44 @@ fn run_impl() {
     init_logging();
     log::trace!("pid0: start");
 
-    wetware_guest::run(|host: peer_capnp::host::Client| async move {
-        log::trace!("pid0: rpc bootstrapped");
-        const CHILD_WASM: &[u8] =
-            include_bytes!("../../child-echo/target/wasm32-wasip2/release/child_echo.wasm");
-
-        let executor = host.executor_request().send().pipeline.get_executor();
-
-        let mut request = executor.run_bytes_request();
-        {
-            let mut params = request.get();
-            params.set_wasm(CHILD_WASM);
-            params.reborrow().init_args(0);
-            params.reborrow().init_env(0);
-        }
-        log::trace!("pid0: runBytes sent");
-
-        let run_resp = request.send().promise.await?;
-        let process = run_resp.get()?.get_process()?;
-        log::trace!("pid0: got process");
-
-        let wait_resp = process.wait_request().send().promise.await?;
-        let exit_code = wait_resp.get()?.get_exit_code();
-        log::trace!("pid0: child exited with code {}", exit_code);
-
-        Ok(())
-    });
+    // Bootstrap a Membrane(WetwareSession) instead of a bare Host.
+    // The membrane provides epoch-scoped sessions with Host + Executor.
+    wetware_guest::run(
+        |membrane: Membrane| async move {
+            log::trace!("pid0: rpc bootstrapped, grafting onto membrane");
+
+            // Graft onto the membrane to get an epoch-scoped session.
+            // No signer needed — stem's graft() currently ignores it.
+            let graft_resp = membrane.graft_request().send().promise.await?;
+            let session = graft_resp.get()?.get_session()?;
+            let ext = session.get_extension()?;
+            log::trace!("pid0: grafted, got session");
+
+            let executor = ext.get_executor()?;
+
+            const CHILD_WASM: &[u8] =
+                include_bytes!("../../child-echo/target/wasm32-wasip2/release/child_echo.wasm");
+
+            let mut request = executor.run_bytes_request();
+            {
+                let mut params = request.get();
+                params.set_wasm(CHILD_WASM);
+                params.reborrow().init_args(0);
+                params.reborrow().init_env(0);
+            }
+            log::trace!("pid0: runBytes sent");
+
+            let run_resp = request.send().promise.await?;
+            let process = run_resp.get()?.get_process()?;
+            log::trace!("pid0: got process");
+
+            let wait_resp = process.wait_request().send().promise.await?;
+            let exit_code = wait_resp.get()?.get_exit_code();
+            log::trace!("pid0: child exited with code {}", exit_code);
+
+            Ok(())
+        },
+    );
 
     log::trace!("pid0: cleanup complete");
 }

src/cell/executor.rs

@@ -225,8 +225,21 @@ impl Cell {
         let (reader, writer) = handles
             .take_host_split()
             .ok_or_else(|| anyhow::anyhow!("host stream missing; RPC streams already consumed"))?;
-        let rpc_system =
-            crate::rpc::build_peer_rpc(reader, writer, network_state, swarm_cmd_tx, wasm_debug);
+        // Static epoch (never advances) — real epoch wiring is a future concern.
+        let initial_epoch = stem::membrane::Epoch {
+            seq: 0,
+            head: vec![],
+            adopted_block: 0,
+        };
+        let (_epoch_tx, epoch_rx) = tokio::sync::watch::channel(initial_epoch);
+        let rpc_system = crate::rpc::membrane::build_membrane_rpc(
+            reader,
+            writer,
+            network_state,
+            swarm_cmd_tx,
+            wasm_debug,
+            epoch_rx,
+        );
 
         info!("Starting streams RPC server for guest");
         let local = tokio::task::LocalSet::new();

src/lib.rs

@@ -15,9 +15,15 @@ pub mod loaders;
 #[cfg(not(target_arch = "wasm32"))]
 pub mod rpc;
 #[cfg(not(target_arch = "wasm32"))]
+#[allow(unused_parens)]
 pub mod peer_capnp {
     include!(concat!(env!("OUT_DIR"), "/capnp/peer_capnp.rs"));
 }
+#[cfg(not(target_arch = "wasm32"))]
+#[allow(unused_parens)]
+pub mod membrane_capnp {
+    include!(concat!(env!("OUT_DIR"), "/capnp/membrane_capnp.rs"));
+}
 
 // Modules available for both host and guest
 pub mod config;