feat(plugin-wasm): migrate plugin system to WASI 0.3 / component-model async

crates/plugin-wasm/Cargo.toml

@@ -15,8 +15,8 @@ publish = false
 [dependencies]
 streamkit-core = { workspace = true }
 
-wasmtime = { version = "44.0.1", features = ["component-model", "async"] }
-wasmtime-wasi = "44.0.1"
+wasmtime = { version = "45.0.1", features = ["component-model", "component-model-async", "async"] }
+wasmtime-wasi = { version = "45.0.1", features = ["p3"] }
 
 anyhow = "1.0"
 tokio = { workspace = true, features = ["macros", "sync"] }
@@ -29,7 +29,7 @@ serde-saphyr = { workspace = true }
 futures = { workspace = true }
 
 [build-dependencies]
-wasmtime = { version = "44.0.1", features = ["component-model"] }
+wasmtime = { version = "45.0.1", features = ["component-model"] }
 
 [dev-dependencies]
 tempfile = "3.27"

crates/plugin-wasm/src/lib.rs

@@ -23,11 +23,15 @@ mod bindings {
         path: "../../wit",
         world: "plugin",
         imports: { default: async },
-        exports: { default: async },
+        exports: { default: async | store },
+        with: {
+            "wasi": wasmtime_wasi::p3::bindings,
+        },
+        require_store_data_send: true,
     });
 }
 
-use bindings::streamkit::plugin::host::Host;
+use bindings::streamkit::plugin::host::{Host, HostWithStore};
 pub use bindings::streamkit::plugin::types as wit_types;
 use bindings::Plugin;
 
@@ -73,6 +77,7 @@ impl PluginRuntime {
     pub fn new(config: PluginRuntimeConfig) -> Result<Self> {
         let mut engine_config = Config::new();
         engine_config.wasm_component_model(true);
+        engine_config.wasm_component_model_async(true);
         engine_config.wasm_simd(config.enable_simd);
         if !config.enable_simd {
             engine_config.wasm_relaxed_simd(false);
@@ -82,7 +87,10 @@ impl PluginRuntime {
         let engine = Engine::new(&engine_config)?;
         let mut linker = Linker::new(&engine);
 
+        // Guests built with current toolchains (e.g. Rust's wasm32-wasip2) still
+        // import wasi@0.2 interfaces from std, so link both p2 and p3.
         wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
+        wasmtime_wasi::p3::add_to_linker(&mut linker)?;
         bindings::streamkit::plugin::host::add_to_linker::<HostState, HasSelf<_>>(
             &mut linker,
             |s| s,
@@ -130,12 +138,13 @@ impl PluginRuntime {
         let mut store = Store::new(&self.engine, host_state);
         store.limiter(|s| &mut s.limits);
 
-        let instance =
-            futures::executor::block_on(self.linker.instantiate_async(&mut store, component))?;
-        let plugin = Plugin::new(&mut store, &instance)?;
+        let metadata = futures::executor::block_on(async {
+            let instance = self.linker.instantiate_async(&mut store, component).await?;
+            let plugin = Plugin::new(&mut store, &instance)?;
 
-        let node = plugin.streamkit_plugin_node();
-        let metadata = futures::executor::block_on(node.call_metadata(&mut store))?;
+            let node = plugin.streamkit_plugin_node();
+            store.run_concurrent(async move |accessor| node.call_metadata(accessor).await).await?
+        })?;
 
         Ok(metadata)
     }
@@ -251,22 +260,31 @@ impl WasiView for HostState {
     }
 }
 
-impl Host for HostState {
-    async fn send_output(
-        &mut self,
+impl HostWithStore for HasSelf<HostState> {
+    async fn send_output<T: 'static>(
+        accessor: &wasmtime::component::Accessor<T, Self>,
         pin_name: String,
         packet: wit_types::Packet,
     ) -> Result<(), String> {
-        if let Some(sender) = &self.output_sender {
-            let core_packet = streamkit_core::types::Packet::try_from(packet)?;
-            // Tighten lock scope: acquire lock only for the send operation
-            sender.lock().await.send(&pin_name, core_packet).await.map_err(|e| e.to_string())?;
-            Ok(())
-        } else {
-            Err("Output sender not initialized".to_string())
-        }
+        let sender = accessor.with(|mut access| access.get().output_sender.clone());
+        send_packet_to_output(sender, &pin_name, packet).await
     }
+}
 
+async fn send_packet_to_output(
+    sender: Option<Arc<Mutex<streamkit_core::OutputSender>>>,
+    pin_name: &str,
+    packet: wit_types::Packet,
+) -> Result<(), String> {
+    let Some(sender) = sender else {
+        return Err("Output sender not initialized".to_string());
+    };
+    let core_packet = streamkit_core::types::Packet::try_from(packet)?;
+    let mut guard = sender.lock().await;
+    guard.send(pin_name, core_packet).await.map_err(|e| e.to_string())
+}
+
+impl Host for HostState {
     async fn log(&mut self, level: LogLevel, message: String) {
         match level {
             LogLevel::Debug => tracing::debug!("[Plugin] {}", message),
@@ -640,17 +658,11 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn host_state_send_output_returns_error_when_no_sender_is_attached() {
-        // Direct exercise of the Host trait impl: without an output_sender set,
-        // any send_output call must surface a clear error rather than panicking.
-        let mut state = HostState {
-            wasi: wasmtime_wasi::WasiCtx::builder().build(),
-            resource_table: wasmtime::component::ResourceTable::new(),
-            output_sender: None,
-            limits: wasmtime::StoreLimitsBuilder::new().memory_size(1 << 20).build(),
-        };
+    async fn send_packet_to_output_returns_error_when_no_sender_is_attached() {
+        // Without an output_sender set, any send_output call must surface a
+        // clear error rather than panicking.
         let packet = wit_types::Packet::Text("hi".to_string());
-        let err = <HostState as Host>::send_output(&mut state, "out".to_string(), packet)
+        let err = send_packet_to_output(None, "out", packet)
             .await
             .expect_err("send_output without sender must error");
         assert!(

crates/plugin-wasm/src/wrapper.rs

@@ -144,11 +144,15 @@ impl ProcessorNode for WasmNodeWrapper {
         // Access the resource interface for `node-instance`
         let instance_iface = node.node_instance();
 
+        // Drive all guest calls inside `run_concurrent` so the component-model
+        // async ABI can multiplex guest tasks while the host loop awaits them.
+        let loop_result = store
+            .run_concurrent(async move |accessor| {
         tracing::debug!(node = %node_id, "Calling plugin constructor");
 
         // Construct a new stateful instance in the plugin with parameters
         let instance_handle =
-            match instance_iface.call_constructor(&mut store, initial_params_json.as_deref()).await
+            match instance_iface.call_constructor(accessor, initial_params_json.clone()).await
             {
                 Ok(handle) => {
                     tracing::debug!(node = %node_id, "Plugin constructor succeeded");
@@ -198,7 +202,7 @@ impl ProcessorNode for WasmNodeWrapper {
                             };
 
                             match instance_iface
-                                .call_update_params(&mut store, instance_handle, params_json.as_deref())
+                                .call_update_params(accessor, instance_handle, params_json)
                                 .await
                             {
                                 Ok(Ok(())) => {
@@ -255,7 +259,7 @@ impl ProcessorNode for WasmNodeWrapper {
                             let wit_packet: wit_types::Packet = packet.into();
 
                             match instance_iface
-                                .call_process(&mut store, instance_handle, &input_pin, &wit_packet)
+                                .call_process(accessor, instance_handle, input_pin, wit_packet)
                                 .await
                             {
                                 Ok(Ok(())) => {}
@@ -310,7 +314,7 @@ impl ProcessorNode for WasmNodeWrapper {
         }
 
         // Clean up
-        if let Err(e) = instance_iface.call_cleanup(&mut store, instance_handle).await {
+        if let Err(e) = instance_iface.call_cleanup(accessor, instance_handle).await {
             tracing::warn!("Plugin cleanup error: {}", e);
         }
 
@@ -321,6 +325,13 @@ impl ProcessorNode for WasmNodeWrapper {
         );
 
         Ok(())
+            })
+            .await;
+
+        match loop_result {
+            Ok(result) => result,
+            Err(e) => Err(StreamKitError::Runtime(format!("Plugin execution error: {e}"))),
+        }
     }
 }
 

crates/plugin-wasm/tests/gain_plugin_smoke.rs

@@ -0,0 +1,110 @@
+// SPDX-FileCopyrightText: © 2025 StreamKit Contributors
+//
+// SPDX-License-Identifier: MPL-2.0
+
+//! End-to-end smoke test driving the prebuilt gain example plugin through the
+//! WASI 0.3 async host. Ignored by default because it requires the example
+//! component to be built first:
+//!
+//! ```sh
+//! cargo build --release --target wasm32-wasip2 \
+//!     --manifest-path examples/plugins/gain-wasm-rust/Cargo.toml
+//! ```
+
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use streamkit_core::node::{OutputRouting, OutputSender};
+use streamkit_core::types::{AudioFrame, Packet};
+use streamkit_core::{NodeContext, PipelineMode};
+use streamkit_plugin_wasm::{PluginRuntime, PluginRuntimeConfig};
+use tokio::sync::mpsc;
+
+fn example_plugin_path(relative: &str) -> PathBuf {
+    let path =
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../examples/plugins").join(relative);
+    assert!(path.exists(), "{} not found; build the example first", path.display());
+    path
+}
+
+async fn run_gain_plugin(path: PathBuf, expected_kind: &str) {
+    let runtime = PluginRuntime::new(PluginRuntimeConfig::default()).expect("runtime");
+    let plugin = runtime.load_plugin(&path).expect("load plugin");
+    assert_eq!(plugin.metadata().kind, expected_kind);
+
+    let node =
+        plugin.create_node(Some(&serde_json::json!({"gain_db": 6.0206}))).expect("create node");
+
+    let (input_tx, input_rx) = mpsc::channel(8);
+    let (state_tx, _state_rx) = mpsc::channel(32);
+    let (_control_tx, control_rx) = mpsc::channel(8);
+    let (routed_tx, mut routed_rx) = mpsc::channel(64);
+
+    let mut inputs = HashMap::new();
+    inputs.insert("in".to_string(), input_rx);
+
+    let context = NodeContext {
+        inputs,
+        input_types: HashMap::new(),
+        control_rx,
+        output_sender: OutputSender::new("gain".to_string(), OutputRouting::Routed(routed_tx)),
+        batch_size: 1,
+        state_tx,
+        stats_tx: None,
+        telemetry_tx: None,
+        session_id: None,
+        cancellation_token: None,
+        pin_management_rx: None,
+        audio_pool: None,
+        video_pool: None,
+        pipeline_mode: PipelineMode::Oneshot,
+        view_data_tx: None,
+        engine_control_tx: None,
+        asset_root: std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
+    };
+
+    let handle = tokio::spawn(async move { node.run(context).await });
+
+    input_tx
+        .send(Packet::Audio(AudioFrame::new(48000, 1, vec![0.5f32; 4])))
+        .await
+        .expect("send input");
+    drop(input_tx);
+
+    let (_node, pin, packet) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), routed_rx.recv())
+            .await
+            .expect("timed out waiting for plugin output")
+            .expect("output channel closed without packet");
+    assert_eq!(&*pin, "out");
+
+    match packet {
+        Packet::Audio(frame) => {
+            for sample in frame.samples.iter() {
+                assert!((sample - 1.0).abs() < 1e-3, "expected ~2x gain, got {sample}");
+            }
+        },
+        other => panic!("unexpected packet: {other:?}"),
+    }
+
+    handle.await.expect("join").expect("node run");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "requires the gain example component to be prebuilt"]
+async fn rust_gain_plugin_processes_audio_through_async_host() {
+    run_gain_plugin(
+        example_plugin_path("gain-wasm-rust/target/wasm32-wasip2/release/gain_plugin.wasm"),
+        "gain_filter_rust",
+    )
+    .await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "requires the gain example component to be prebuilt"]
+async fn c_gain_plugin_processes_audio_through_async_host() {
+    run_gain_plugin(example_plugin_path("gain-wasm-c/build/gain_plugin_c.wasm"), "gain_filter_c")
+        .await;
+}

docs/src/content/docs/guides/writing-plugins.md

@@ -172,7 +172,7 @@ WASM plugins run in a sandboxed WebAssembly Component Model runtime.
 ```bash
 cargo new --lib my-wasm-plugin
 cd my-wasm-plugin
-cargo install cargo-component
+rustup target add wasm32-wasip2
 ```
 
 ```toml
@@ -187,7 +187,7 @@ crate-type = ["cdylib"]
 
 [dependencies]
 streamkit-plugin-sdk-wasm = "0.1.0"
-wit-bindgen = "0.44"
+wit-bindgen = "0.58"
 serde_json = "1"
 ```
 
@@ -198,11 +198,11 @@ serde_json = "1"
 
 ```bash
 # Build
-cargo component build --release
+cargo build --release --target wasm32-wasip2
 
 # Upload
 curl -X POST \
-  -F plugin=@target/wasm32-wasip1/release/my_wasm_plugin.wasm \
+  -F plugin=@target/wasm32-wasip2/release/my_wasm_plugin.wasm \
   http://localhost:4545/api/v1/plugins
 ```
 
@@ -212,7 +212,6 @@ For complete, working examples:
 
 - `examples/plugins/gain-native`
 - `examples/plugins/gain-wasm-rust`
-- `examples/plugins/gain-wasm-go`
 - `examples/plugins/gain-wasm-c`
 
 ## Next Steps

examples/plugins/gain-wasm-c/README.md

@@ -34,7 +34,7 @@ The C plugin is expected to be comparable to or slightly smaller than Rust, with
 - ❌ Manual memory management (risk of memory leaks/bugs)
 - ❌ More verbose code (no high-level abstractions)
 - ❌ Requires manual JSON parsing (no serde equivalent)
-- ❌ Less integrated tooling compared to Rust's `cargo component`
+- ❌ Less integrated tooling compared to Rust's built-in `wasm32-wasip2` target
 
 ## Prerequisites
 
@@ -189,7 +189,7 @@ wasm-tools component new --help
 - General-purpose plugins (better ergonomics, safety)
 - Complex state management (borrow checker prevents bugs)
 - Rich JSON parsing (serde)
-- Integrated tooling (`cargo component`)
+- Integrated tooling (built-in `wasm32-wasip2` target)
 
 **When to use Go instead:**
 - Familiar to Go developers