From 74d94d33b0c23a04e2b48508fcaecf642d1bf2fd Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Tue, 3 Feb 2026 14:50:20 -0800 Subject: [PATCH] Implementing channels to be handled in wasm --- channels-src/slack/Cargo.toml | 29 + channels-src/slack/build.sh | 37 + channels-src/slack/slack.capabilities.json | 38 + channels-src/slack/src/lib.rs | 376 ++++++ src/channels/mod.rs | 25 + src/channels/wasm/capabilities.rs | 316 +++++ src/channels/wasm/error.rs | 83 ++ src/channels/wasm/host.rs | 500 ++++++++ src/channels/wasm/loader.rs | 360 ++++++ src/channels/wasm/mod.rs | 102 ++ src/channels/wasm/router.rs | 480 ++++++++ src/channels/wasm/runtime.rs | 285 +++++ src/channels/wasm/schema.rs | 417 +++++++ src/channels/wasm/wrapper.rs | 1274 ++++++++++++++++++++ src/config.rs | 23 + src/main.rs | 40 +- src/tools/wasm/mod.rs | 2 +- tests/wasm_channel_integration.rs | 446 +++++++ wit/channel.wit | 276 +++++ 19 files changed, 5107 insertions(+), 2 deletions(-) create mode 100644 channels-src/slack/Cargo.toml create mode 100755 channels-src/slack/build.sh create mode 100644 channels-src/slack/slack.capabilities.json create mode 100644 channels-src/slack/src/lib.rs create mode 100644 src/channels/wasm/capabilities.rs create mode 100644 src/channels/wasm/error.rs create mode 100644 src/channels/wasm/host.rs create mode 100644 src/channels/wasm/loader.rs create mode 100644 src/channels/wasm/mod.rs create mode 100644 src/channels/wasm/router.rs create mode 100644 src/channels/wasm/runtime.rs create mode 100644 src/channels/wasm/schema.rs create mode 100644 src/channels/wasm/wrapper.rs create mode 100644 tests/wasm_channel_integration.rs create mode 100644 wit/channel.wit diff --git a/channels-src/slack/Cargo.toml b/channels-src/slack/Cargo.toml new file mode 100644 index 00000000..c434519e --- /dev/null +++ b/channels-src/slack/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "slack-channel" +version = "0.1.0" +edition = "2021" +description = "Slack Events API channel for NEAR Agent" +license = "MIT OR Apache-2.0" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +# WIT bindgen for WASM component model +wit-bindgen = "0.36" + +# Serialization +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" + +# HMAC for signature validation +hmac = "0.12" +sha2 = "0.10" +hex = "0.4" + +[profile.release] +# Optimize for size +opt-level = "s" +lto = true +strip = true +codegen-units = 1 diff --git a/channels-src/slack/build.sh b/channels-src/slack/build.sh new file mode 100755 index 00000000..5b568c8d --- /dev/null +++ b/channels-src/slack/build.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# Build the Slack channel WASM component +# +# Prerequisites: +# - Rust with wasm32-wasip2 target: rustup target add wasm32-wasip2 +# - wasm-tools for component creation: cargo install wasm-tools +# +# Output: +# - slack.wasm - WASM component ready for deployment +# - slack.capabilities.json - Capabilities file (copy alongside .wasm) + +set -euo pipefail + +cd "$(dirname "$0")" + +echo "Building Slack channel WASM component..." + +# Build the WASM module +cargo build --release --target wasm32-wasip2 + +# Convert to component model (if not already a component) +# wasm-tools component new is idempotent on components +WASM_PATH="target/wasm32-wasip2/release/slack_channel.wasm" + +if [ -f "$WASM_PATH" ]; then + # Create component if needed + wasm-tools component new "$WASM_PATH" -o slack.wasm 2>/dev/null || cp "$WASM_PATH" slack.wasm + + # Optimize the component + wasm-tools strip slack.wasm -o slack.wasm + + echo "Built: slack.wasm ($(du -h slack.wasm | cut -f1))" + echo "Copy slack.wasm and slack.capabilities.json to ~/.near-agent/channels/" +else + echo "Error: WASM output not found at $WASM_PATH" + exit 1 +fi diff --git a/channels-src/slack/slack.capabilities.json b/channels-src/slack/slack.capabilities.json new file mode 100644 index 00000000..2b8070ff --- /dev/null +++ b/channels-src/slack/slack.capabilities.json @@ -0,0 +1,38 @@ +{ + "type": "channel", + "name": "slack", + "description": "Slack Events API channel for receiving and responding to Slack messages", + "capabilities": { + "http": { + "allowlist": [ + { "host": "slack.com", "path_prefix": "/api/" } + ], + "credentials": { + "slack_bot": { + "secret_name": "slack_bot_token", + "location": { "type": "bearer" }, + "host_patterns": ["slack.com"] + } + }, + "rate_limit": { + "requests_per_minute": 50, + "requests_per_hour": 1000 + } + }, + "secrets": { + "allowed_names": ["slack_*"] + }, + "channel": { + "allowed_paths": ["/webhook/slack"], + "allow_polling": false, + "workspace_prefix": "channels/slack/", + "emit_rate_limit": { + "messages_per_minute": 100, + "messages_per_hour": 5000 + } + } + }, + "config": { + "signing_secret_name": "slack_signing_secret" + } +} diff --git a/channels-src/slack/src/lib.rs b/channels-src/slack/src/lib.rs new file mode 100644 index 00000000..18748833 --- /dev/null +++ b/channels-src/slack/src/lib.rs @@ -0,0 +1,376 @@ +//! Slack Events API channel for NEAR Agent. +//! +//! This WASM component implements the channel interface for handling Slack +//! webhooks and sending messages back to Slack. +//! +//! # Features +//! +//! - URL verification for Slack Events API +//! - Message event parsing (@mentions, DMs) +//! - Thread support for conversations +//! - Response posting via Slack Web API +//! +//! # Security +//! +//! - Signature validation is handled by the host (webhook secrets) +//! - Bot token is injected by host during HTTP requests +//! - WASM never sees raw credentials + +// Generate bindings from the WIT file +wit_bindgen::generate!({ + world: "sandboxed-channel", + path: "../../wit/channel.wit", +}); + +use serde::{Deserialize, Serialize}; + +// Re-export generated types +use exports::near::agent::channel::{ + AgentResponse, ChannelConfig, Guest, HttpEndpointConfig, IncomingHttpRequest, + OutgoingHttpResponse, PollConfig, +}; +use near::agent::channel_host::{self, EmittedMessage}; + +/// Slack event wrapper. +#[derive(Debug, Deserialize)] +struct SlackEventWrapper { + /// Event type (url_verification, event_callback, etc.) + #[serde(rename = "type")] + event_type: String, + + /// Challenge token for URL verification. + challenge: Option, + + /// The actual event payload (for event_callback). + event: Option, + + /// Team ID that sent this event. + team_id: Option, + + /// Event ID for deduplication. + event_id: Option, +} + +/// Slack event payload. +#[derive(Debug, Deserialize)] +struct SlackEvent { + /// Event type (message, app_mention, etc.) + #[serde(rename = "type")] + event_type: String, + + /// User who triggered the event. + user: Option, + + /// Channel where the event occurred. + channel: Option, + + /// Message text. + text: Option, + + /// Thread timestamp (for threaded messages). + thread_ts: Option, + + /// Message timestamp. + ts: Option, + + /// Bot ID (if message is from a bot). + bot_id: Option, + + /// Subtype (bot_message, etc.) + subtype: Option, +} + +/// Metadata stored with emitted messages for response routing. +#[derive(Debug, Serialize, Deserialize)] +struct SlackMessageMetadata { + /// Slack channel ID. + channel: String, + + /// Thread timestamp for threaded replies. + thread_ts: Option, + + /// Original message timestamp. + message_ts: String, + + /// Team ID. + team_id: Option, +} + +/// Slack API response for chat.postMessage. +#[derive(Debug, Deserialize)] +struct SlackPostMessageResponse { + ok: bool, + error: Option, + ts: Option, +} + +/// Channel configuration from capabilities file. +#[derive(Debug, Deserialize)] +struct SlackConfig { + /// Name of secret containing signing secret (for verification by host). + #[serde(default = "default_signing_secret_name")] + signing_secret_name: String, +} + +fn default_signing_secret_name() -> String { + "slack_signing_secret".to_string() +} + +struct SlackChannel; + +impl Guest for SlackChannel { + fn on_start(config_json: String) -> Result { + // Parse configuration + let _config: SlackConfig = serde_json::from_str(&config_json) + .map_err(|e| format!("Failed to parse config: {}", e))?; + + channel_host::log(channel_host::LogLevel::Info, "Slack channel starting"); + + Ok(ChannelConfig { + display_name: "Slack".to_string(), + http_endpoints: vec![HttpEndpointConfig { + path: "/webhook/slack".to_string(), + methods: vec!["POST".to_string()], + require_secret: true, + }], + poll: None, // Slack uses push via webhooks, no polling needed + }) + } + + fn on_http_request(req: IncomingHttpRequest) -> OutgoingHttpResponse { + // Parse the request body + let body_str = match std::str::from_utf8(&req.body) { + Ok(s) => s, + Err(_) => { + return json_response(400, serde_json::json!({"error": "Invalid UTF-8 body"})); + } + }; + + // Parse as Slack event + let event_wrapper: SlackEventWrapper = match serde_json::from_str(body_str) { + Ok(e) => e, + Err(e) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to parse Slack event: {}", e), + ); + return json_response(400, serde_json::json!({"error": "Invalid event payload"})); + } + }; + + match event_wrapper.event_type.as_str() { + // URL verification challenge (Slack setup) + "url_verification" => { + if let Some(challenge) = event_wrapper.challenge { + channel_host::log( + channel_host::LogLevel::Info, + "Responding to Slack URL verification", + ); + json_response(200, serde_json::json!({"challenge": challenge})) + } else { + json_response(400, serde_json::json!({"error": "Missing challenge"})) + } + } + + // Actual event callback + "event_callback" => { + if let Some(event) = event_wrapper.event { + handle_slack_event( + event, + event_wrapper.team_id, + event_wrapper.event_id, + ); + } + // Always respond 200 quickly to Slack (they have a 3s timeout) + json_response(200, serde_json::json!({"ok": true})) + } + + // Unknown event type + _ => { + channel_host::log( + channel_host::LogLevel::Warn, + &format!("Unknown Slack event type: {}", event_wrapper.event_type), + ); + json_response(200, serde_json::json!({"ok": true})) + } + } + } + + fn on_poll() { + // Slack uses webhooks, no polling needed + } + + fn on_respond(response: AgentResponse) -> Result<(), String> { + // Parse metadata to get channel info + let metadata: SlackMessageMetadata = serde_json::from_str(&response.metadata_json) + .map_err(|e| format!("Failed to parse metadata: {}", e))?; + + // Build Slack API request + let mut payload = serde_json::json!({ + "channel": metadata.channel, + "text": response.content, + }); + + // Add thread_ts for threaded replies + if let Some(thread_ts) = response.thread_id.or(metadata.thread_ts) { + payload["thread_ts"] = serde_json::Value::String(thread_ts); + } + + let payload_bytes = serde_json::to_vec(&payload) + .map_err(|e| format!("Failed to serialize payload: {}", e))?; + + // Make HTTP request to Slack API + // The bot token is injected by the host based on credential configuration + let headers = serde_json::json!({ + "Content-Type": "application/json" + }); + + let result = channel_host::http_request( + "POST", + "https://slack.com/api/chat.postMessage", + &headers.to_string(), + Some(&payload_bytes), + ); + + match result { + Ok(http_response) => { + if http_response.status != 200 { + return Err(format!( + "Slack API returned status {}", + http_response.status + )); + } + + // Parse Slack response + let slack_response: SlackPostMessageResponse = + serde_json::from_slice(&http_response.body).map_err(|e| { + format!("Failed to parse Slack response: {}", e) + })?; + + if !slack_response.ok { + return Err(format!( + "Slack API error: {}", + slack_response.error.unwrap_or_else(|| "unknown".to_string()) + )); + } + + channel_host::log( + channel_host::LogLevel::Debug, + &format!( + "Posted message to Slack channel {}: ts={}", + metadata.channel, + slack_response.ts.unwrap_or_default() + ), + ); + + Ok(()) + } + Err(e) => Err(format!("HTTP request failed: {}", e)), + } + } + + fn on_shutdown() { + channel_host::log(channel_host::LogLevel::Info, "Slack channel shutting down"); + } +} + +/// Handle a Slack event and emit message if applicable. +fn handle_slack_event( + event: SlackEvent, + team_id: Option, + _event_id: Option, +) { + match event.event_type.as_str() { + // Direct mention of the bot + "app_mention" => { + if let (Some(user), Some(channel), Some(text), Some(ts)) = + (event.user, event.channel.clone(), event.text, event.ts.clone()) + { + emit_message(user, text, channel, event.thread_ts.or(Some(ts)), team_id); + } + } + + // Direct message to the bot + "message" => { + // Skip messages from bots (including ourselves) + if event.bot_id.is_some() || event.subtype.is_some() { + return; + } + + if let (Some(user), Some(channel), Some(text), Some(ts)) = + (event.user, event.channel.clone(), event.text, event.ts.clone()) + { + // Only process DMs (channel IDs starting with D) + if channel.starts_with('D') { + emit_message(user, text, channel, event.thread_ts.or(Some(ts)), team_id); + } + } + } + + _ => { + channel_host::log( + channel_host::LogLevel::Debug, + &format!("Ignoring Slack event type: {}", event.event_type), + ); + } + } +} + +/// Emit a message to the agent. +fn emit_message( + user_id: String, + text: String, + channel: String, + thread_ts: Option, + team_id: Option, +) { + let message_ts = thread_ts.clone().unwrap_or_default(); + + let metadata = SlackMessageMetadata { + channel: channel.clone(), + thread_ts: thread_ts.clone(), + message_ts: message_ts.clone(), + team_id, + }; + + let metadata_json = + serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); + + // Strip @ mentions of the bot from the text for cleaner messages + let cleaned_text = strip_bot_mention(&text); + + channel_host::emit_message(EmittedMessage { + user_id, + user_name: None, // Could fetch from Slack API if needed + content: cleaned_text, + thread_id: thread_ts, + metadata_json, + }); +} + +/// Strip leading bot mention from text. +fn strip_bot_mention(text: &str) -> String { + // Slack mentions look like <@U12345678> + let trimmed = text.trim(); + if trimmed.starts_with("<@") { + if let Some(end) = trimmed.find('>') { + return trimmed[end + 1..].trim_start().to_string(); + } + } + trimmed.to_string() +} + +/// Create a JSON HTTP response. +fn json_response(status: u16, value: serde_json::Value) -> OutgoingHttpResponse { + let body = serde_json::to_vec(&value).unwrap_or_default(); + let headers = serde_json::json!({"Content-Type": "application/json"}); + + OutgoingHttpResponse { + status, + headers_json: headers.to_string(), + body, + } +} + +// Export the component +export!(SlackChannel); diff --git a/src/channels/mod.rs b/src/channels/mod.rs index d47be5f9..550707d1 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -2,12 +2,37 @@ //! //! Channels receive messages from external sources (CLI, HTTP, etc.) //! and convert them to a unified message format for the agent to process. +//! +//! # Architecture +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────────────┐ +//! │ ChannelManager │ +//! │ │ +//! │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ +//! │ │ TuiChannel │ │ HttpChannel │ │ WasmChannel │ ... │ +//! │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ +//! │ │ │ │ │ +//! │ └─────────────────┴─────────────────┘ │ +//! │ │ │ +//! │ select_all (futures) │ +//! │ │ │ +//! │ ▼ │ +//! │ MessageStream │ +//! └─────────────────────────────────────────────────────────────────────┘ +//! ``` +//! +//! # WASM Channels +//! +//! WASM channels allow dynamic loading of channel implementations at runtime. +//! See the [`wasm`] module for details. mod channel; pub mod cli; mod http; mod manager; mod repl; +pub mod wasm; pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; pub use cli::{AppEvent, TuiChannel}; diff --git a/src/channels/wasm/capabilities.rs b/src/channels/wasm/capabilities.rs new file mode 100644 index 00000000..2e3f47b3 --- /dev/null +++ b/src/channels/wasm/capabilities.rs @@ -0,0 +1,316 @@ +//! Channel-specific capabilities for WASM channels. +//! +//! Defines the capability system that controls what a WASM channel can do. +//! Channels have additional capabilities beyond tools: HTTP endpoint registration, +//! message emission, and workspace write access within their namespace. + +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +use crate::tools::wasm::{Capabilities as ToolCapabilities, RateLimitConfig}; + +/// Minimum allowed polling interval (30 seconds). +pub const MIN_POLL_INTERVAL_MS: u32 = 30_000; + +/// Default emit rate limit. +pub const DEFAULT_EMIT_RATE_PER_MINUTE: u32 = 100; +pub const DEFAULT_EMIT_RATE_PER_HOUR: u32 = 5000; + +/// Capabilities specific to WASM channels. +/// +/// Extends tool capabilities with channel-specific permissions. +#[derive(Debug, Clone)] +pub struct ChannelCapabilities { + /// Base tool capabilities (HTTP, secrets, workspace_read, etc.). + pub tool_capabilities: ToolCapabilities, + + /// HTTP paths this channel can register for webhooks. + /// Paths must start with "/webhook/" by convention. + pub allowed_paths: Vec, + + /// Whether polling is allowed for this channel. + pub allow_polling: bool, + + /// Minimum poll interval in milliseconds. + /// Enforced to be at least MIN_POLL_INTERVAL_MS. + pub min_poll_interval_ms: u32, + + /// Workspace prefix for this channel's storage. + /// All workspace writes are automatically prefixed. + /// Example: "channels/slack/" means writes to "state.json" become "channels/slack/state.json". + pub workspace_prefix: String, + + /// Rate limiting for emit_message calls. + pub emit_rate_limit: EmitRateLimitConfig, + + /// Maximum message content size in bytes. + pub max_message_size: usize, + + /// Callback timeout duration. + pub callback_timeout: Duration, +} + +impl Default for ChannelCapabilities { + fn default() -> Self { + Self { + tool_capabilities: ToolCapabilities::default(), + allowed_paths: Vec::new(), + allow_polling: false, + min_poll_interval_ms: MIN_POLL_INTERVAL_MS, + workspace_prefix: String::new(), + emit_rate_limit: EmitRateLimitConfig::default(), + max_message_size: 64 * 1024, // 64 KB + callback_timeout: Duration::from_secs(30), + } + } +} + +impl ChannelCapabilities { + /// Create capabilities for a channel with the given name. + pub fn for_channel(name: &str) -> Self { + Self { + workspace_prefix: format!("channels/{}/", name), + ..Default::default() + } + } + + /// Add an allowed HTTP path. + pub fn with_path(mut self, path: impl Into) -> Self { + self.allowed_paths.push(path.into()); + self + } + + /// Enable polling with the given minimum interval. + pub fn with_polling(mut self, min_interval_ms: u32) -> Self { + self.allow_polling = true; + self.min_poll_interval_ms = min_interval_ms.max(MIN_POLL_INTERVAL_MS); + self + } + + /// Set the emit rate limit. + pub fn with_emit_rate_limit(mut self, rate_limit: EmitRateLimitConfig) -> Self { + self.emit_rate_limit = rate_limit; + self + } + + /// Set the callback timeout. + pub fn with_callback_timeout(mut self, timeout: Duration) -> Self { + self.callback_timeout = timeout; + self + } + + /// Set the base tool capabilities. + pub fn with_tool_capabilities(mut self, capabilities: ToolCapabilities) -> Self { + self.tool_capabilities = capabilities; + self + } + + /// Check if a path is allowed for this channel. + pub fn is_path_allowed(&self, path: &str) -> bool { + self.allowed_paths.iter().any(|p| p == path) + } + + /// Validate and normalize a poll interval. + /// + /// Returns the interval clamped to minimum, or an error if polling is disabled. + pub fn validate_poll_interval(&self, interval_ms: u32) -> Result { + if !self.allow_polling { + return Err("Polling not allowed for this channel".to_string()); + } + + Ok(interval_ms.max(self.min_poll_interval_ms)) + } + + /// Prefix a workspace path for this channel. + /// + /// Ensures all workspace writes are scoped to the channel's namespace. + pub fn prefix_workspace_path(&self, path: &str) -> String { + if self.workspace_prefix.is_empty() { + path.to_string() + } else { + format!("{}{}", self.workspace_prefix, path) + } + } + + /// Check if a workspace path is valid for this channel. + /// + /// Paths cannot escape the channel's namespace. + pub fn validate_workspace_path(&self, path: &str) -> Result { + // Block absolute paths + if path.starts_with('/') { + return Err("Absolute paths not allowed".to_string()); + } + + // Block path traversal + if path.contains("..") { + return Err("Parent directory references not allowed".to_string()); + } + + // Block null bytes + if path.contains('\0') { + return Err("Null bytes not allowed".to_string()); + } + + // Prefix with channel namespace + Ok(self.prefix_workspace_path(path)) + } +} + +/// Configuration for an HTTP endpoint the channel wants to register. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HttpEndpointConfig { + /// Path to register (e.g., "/webhook/slack"). + pub path: String, + + /// HTTP methods to accept (e.g., ["POST"]). + pub methods: Vec, + + /// Whether secret validation is required. + pub require_secret: bool, +} + +impl HttpEndpointConfig { + /// Create a POST webhook endpoint. + pub fn post_webhook(path: impl Into) -> Self { + Self { + path: path.into(), + methods: vec!["POST".to_string()], + require_secret: true, + } + } +} + +/// Polling configuration returned by the channel. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PollConfig { + /// Polling interval in milliseconds. + pub interval_ms: u32, + + /// Whether polling is enabled. + pub enabled: bool, +} + +impl Default for PollConfig { + fn default() -> Self { + Self { + interval_ms: MIN_POLL_INTERVAL_MS, + enabled: false, + } + } +} + +/// Rate limiting configuration for message emission. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EmitRateLimitConfig { + /// Maximum messages per minute. + pub messages_per_minute: u32, + + /// Maximum messages per hour. + pub messages_per_hour: u32, +} + +impl Default for EmitRateLimitConfig { + fn default() -> Self { + Self { + messages_per_minute: DEFAULT_EMIT_RATE_PER_MINUTE, + messages_per_hour: DEFAULT_EMIT_RATE_PER_HOUR, + } + } +} + +impl From for EmitRateLimitConfig { + fn from(config: RateLimitConfig) -> Self { + Self { + messages_per_minute: config.requests_per_minute, + messages_per_hour: config.requests_per_hour, + } + } +} + +#[cfg(test)] +mod tests { + use crate::channels::wasm::capabilities::{ + ChannelCapabilities, EmitRateLimitConfig, HttpEndpointConfig, MIN_POLL_INTERVAL_MS, + }; + + #[test] + fn test_default_capabilities() { + let caps = ChannelCapabilities::default(); + assert!(caps.allowed_paths.is_empty()); + assert!(!caps.allow_polling); + assert_eq!(caps.min_poll_interval_ms, MIN_POLL_INTERVAL_MS); + } + + #[test] + fn test_for_channel() { + let caps = ChannelCapabilities::for_channel("slack"); + assert_eq!(caps.workspace_prefix, "channels/slack/"); + } + + #[test] + fn test_path_allowed() { + let caps = ChannelCapabilities::default() + .with_path("/webhook/slack") + .with_path("/webhook/slack/events"); + + assert!(caps.is_path_allowed("/webhook/slack")); + assert!(caps.is_path_allowed("/webhook/slack/events")); + assert!(!caps.is_path_allowed("/webhook/telegram")); + } + + #[test] + fn test_poll_interval_validation() { + let caps = ChannelCapabilities::default().with_polling(60_000); + + // Valid interval + assert_eq!(caps.validate_poll_interval(90_000).unwrap(), 90_000); + + // Too short, clamped to minimum + assert_eq!(caps.validate_poll_interval(1000).unwrap(), 60_000); + + // Polling disabled + let no_poll_caps = ChannelCapabilities::default(); + assert!(no_poll_caps.validate_poll_interval(60_000).is_err()); + } + + #[test] + fn test_workspace_path_validation() { + let caps = ChannelCapabilities::for_channel("slack"); + + // Valid path + let result = caps.validate_workspace_path("state.json"); + assert_eq!(result.unwrap(), "channels/slack/state.json"); + + // Nested path + let result = caps.validate_workspace_path("data/users.json"); + assert_eq!(result.unwrap(), "channels/slack/data/users.json"); + + // Block absolute paths + let result = caps.validate_workspace_path("/etc/passwd"); + assert!(result.is_err()); + + // Block path traversal + let result = caps.validate_workspace_path("../secrets/key.txt"); + assert!(result.is_err()); + + // Block null bytes + let result = caps.validate_workspace_path("file\0.txt"); + assert!(result.is_err()); + } + + #[test] + fn test_http_endpoint_config() { + let endpoint = HttpEndpointConfig::post_webhook("/webhook/slack"); + assert_eq!(endpoint.path, "/webhook/slack"); + assert_eq!(endpoint.methods, vec!["POST"]); + assert!(endpoint.require_secret); + } + + #[test] + fn test_emit_rate_limit_default() { + let limit = EmitRateLimitConfig::default(); + assert_eq!(limit.messages_per_minute, 100); + assert_eq!(limit.messages_per_hour, 5000); + } +} diff --git a/src/channels/wasm/error.rs b/src/channels/wasm/error.rs new file mode 100644 index 00000000..76ed4d34 --- /dev/null +++ b/src/channels/wasm/error.rs @@ -0,0 +1,83 @@ +//! Error types for WASM channels. + +use std::path::PathBuf; + +/// Error during WASM channel operations. +#[derive(Debug, thiserror::Error)] +pub enum WasmChannelError { + #[error("Channel {name} failed to start: {reason}")] + StartupFailed { name: String, reason: String }, + + #[error("Channel {name} callback failed: {reason}")] + CallbackFailed { name: String, reason: String }, + + #[error("Channel {name} WASM execution trapped: {reason}")] + Trapped { name: String, reason: String }, + + #[error("Channel {name} callback '{callback}' timed out")] + Timeout { name: String, callback: String }, + + #[error("Channel {name} execution panicked: {reason}")] + ExecutionPanicked { name: String, reason: String }, + + #[error("Channel {name} emit rate limited")] + EmitRateLimited { name: String }, + + #[error("Channel {name} HTTP path not allowed: {path}")] + PathNotAllowed { name: String, path: String }, + + #[error("Channel {name} polling interval too short: {interval_ms}ms (minimum: {min_ms}ms)")] + PollIntervalTooShort { + name: String, + interval_ms: u32, + min_ms: u32, + }, + + #[error("Channel {name} workspace path escape attempt: {path}")] + WorkspaceEscape { name: String, path: String }, + + #[error("Channel {name} exhausted fuel limit ({limit})")] + FuelExhausted { name: String, limit: u64 }, + + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + + #[error("WASM file not found: {0}")] + WasmNotFound(PathBuf), + + #[error("Capabilities file not found: {0}")] + CapabilitiesNotFound(PathBuf), + + #[error("Invalid capabilities JSON: {0}")] + InvalidCapabilities(String), + + #[error("WASM compilation error: {0}")] + Compilation(String), + + #[error("WASM instantiation error: {0}")] + Instantiation(String), + + #[error("Invalid channel name: {0}")] + InvalidName(String), + + #[error("Channel {name} not found")] + NotFound { name: String }, + + #[error("Channel module missing export: {0}")] + MissingExport(String), + + #[error("Invalid response from WASM: {0}")] + InvalidResponse(String), + + #[error("Runtime not initialized")] + RuntimeNotInitialized, + + #[error("Configuration error: {0}")] + Config(String), +} + +impl From for WasmChannelError { + fn from(err: crate::tools::wasm::WasmError) -> Self { + WasmChannelError::Compilation(err.to_string()) + } +} diff --git a/src/channels/wasm/host.rs b/src/channels/wasm/host.rs new file mode 100644 index 00000000..89b2a313 --- /dev/null +++ b/src/channels/wasm/host.rs @@ -0,0 +1,500 @@ +//! Host state for WASM channel execution. +//! +//! Extends the base tool host state with channel-specific functionality: +//! - Message emission (queueing messages to send to the agent) +//! - Workspace write access (scoped to channel namespace) +//! - Rate limiting for message emission + +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::channels::wasm::capabilities::{ChannelCapabilities, EmitRateLimitConfig}; +use crate::channels::wasm::error::WasmChannelError; +use crate::tools::wasm::{HostState, LogLevel}; + +/// Maximum emitted messages per callback execution. +const MAX_EMITS_PER_EXECUTION: usize = 100; + +/// Maximum message content size (64 KB). +const MAX_MESSAGE_CONTENT_SIZE: usize = 64 * 1024; + +/// A message emitted by a WASM channel to be sent to the agent. +#[derive(Debug, Clone)] +pub struct EmittedMessage { + /// User identifier within the channel. + pub user_id: String, + + /// Optional user display name. + pub user_name: Option, + + /// Message content. + pub content: String, + + /// Optional thread ID for threaded conversations. + pub thread_id: Option, + + /// Channel-specific metadata as JSON string. + pub metadata_json: String, + + /// Timestamp when the message was emitted. + pub emitted_at_millis: u64, +} + +impl EmittedMessage { + /// Create a new emitted message. + pub fn new(user_id: impl Into, content: impl Into) -> Self { + Self { + user_id: user_id.into(), + user_name: None, + content: content.into(), + thread_id: None, + metadata_json: "{}".to_string(), + emitted_at_millis: SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0), + } + } + + /// Set the user name. + pub fn with_user_name(mut self, name: impl Into) -> Self { + self.user_name = Some(name.into()); + self + } + + /// Set the thread ID. + pub fn with_thread_id(mut self, thread_id: impl Into) -> Self { + self.thread_id = Some(thread_id.into()); + self + } + + /// Set metadata JSON. + pub fn with_metadata(mut self, metadata_json: impl Into) -> Self { + self.metadata_json = metadata_json.into(); + self + } +} + +/// A pending workspace write operation. +#[derive(Debug, Clone)] +pub struct PendingWorkspaceWrite { + /// Full path (already prefixed with channel namespace). + pub path: String, + + /// Content to write. + pub content: String, +} + +/// Host state for WASM channel callbacks. +/// +/// Maintains all side effects during callback execution and enforces limits. +/// This is the channel-specific equivalent of HostState for tools. +pub struct ChannelHostState { + /// Base tool host state (logging, time, HTTP, etc.). + base: HostState, + + /// Channel name (for error messages). + channel_name: String, + + /// Channel capabilities. + capabilities: ChannelCapabilities, + + /// Emitted messages (queued for delivery). + emitted_messages: Vec, + + /// Pending workspace writes. + pending_writes: Vec, + + /// Emit count for rate limiting within this execution. + emit_count: u32, + + /// Whether emit is still allowed (false after rate limit hit). + emit_enabled: bool, + + /// Count of emits dropped due to rate limiting. + emits_dropped: usize, +} + +impl std::fmt::Debug for ChannelHostState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ChannelHostState") + .field("channel_name", &self.channel_name) + .field("emitted_messages_count", &self.emitted_messages.len()) + .field("pending_writes_count", &self.pending_writes.len()) + .field("emit_count", &self.emit_count) + .field("emit_enabled", &self.emit_enabled) + .field("emits_dropped", &self.emits_dropped) + .finish() + } +} + +impl ChannelHostState { + /// Create a new channel host state. + pub fn new(channel_name: impl Into, capabilities: ChannelCapabilities) -> Self { + let base = HostState::new(capabilities.tool_capabilities.clone()); + + Self { + base, + channel_name: channel_name.into(), + capabilities, + emitted_messages: Vec::new(), + pending_writes: Vec::new(), + emit_count: 0, + emit_enabled: true, + emits_dropped: 0, + } + } + + /// Get the channel name. + pub fn channel_name(&self) -> &str { + &self.channel_name + } + + /// Get the capabilities. + pub fn capabilities(&self) -> &ChannelCapabilities { + &self.capabilities + } + + /// Get the base host state for tool capabilities. + pub fn base(&self) -> &HostState { + &self.base + } + + /// Get mutable access to the base host state. + pub fn base_mut(&mut self) -> &mut HostState { + &mut self.base + } + + /// Emit a message from the channel. + /// + /// Messages are queued and delivered after callback execution completes. + /// Rate limiting is enforced per-execution and globally. + pub fn emit_message(&mut self, msg: EmittedMessage) -> Result<(), WasmChannelError> { + // Check per-execution limit + if !self.emit_enabled { + self.emits_dropped += 1; + return Ok(()); // Silently drop, don't fail execution + } + + if self.emitted_messages.len() >= MAX_EMITS_PER_EXECUTION { + self.emit_enabled = false; + self.emits_dropped += 1; + tracing::warn!( + channel = %self.channel_name, + limit = MAX_EMITS_PER_EXECUTION, + "Channel emit limit reached, further messages dropped" + ); + return Ok(()); + } + + // Validate message content size + if msg.content.len() > MAX_MESSAGE_CONTENT_SIZE { + tracing::warn!( + channel = %self.channel_name, + size = msg.content.len(), + max = MAX_MESSAGE_CONTENT_SIZE, + "Message content too large, truncating" + ); + let mut truncated = msg.content[..MAX_MESSAGE_CONTENT_SIZE].to_string(); + truncated.push_str("... (truncated)"); + let msg = EmittedMessage { + content: truncated, + ..msg + }; + self.emitted_messages.push(msg); + } else { + self.emitted_messages.push(msg); + } + + self.emit_count += 1; + Ok(()) + } + + /// Take all emitted messages (clears the queue). + pub fn take_emitted_messages(&mut self) -> Vec { + std::mem::take(&mut self.emitted_messages) + } + + /// Get the number of emitted messages. + pub fn emitted_count(&self) -> usize { + self.emitted_messages.len() + } + + /// Get the number of emits dropped due to rate limiting. + pub fn emits_dropped(&self) -> usize { + self.emits_dropped + } + + /// Write to workspace (scoped to channel namespace). + /// + /// Writes are queued and committed after callback execution completes. + pub fn workspace_write(&mut self, path: &str, content: String) -> Result<(), WasmChannelError> { + // Validate and prefix path + let full_path = self + .capabilities + .validate_workspace_path(path) + .map_err(|reason| WasmChannelError::WorkspaceEscape { + name: self.channel_name.clone(), + path: reason, + })?; + + self.pending_writes.push(PendingWorkspaceWrite { + path: full_path, + content, + }); + + Ok(()) + } + + /// Take all pending workspace writes (clears the queue). + pub fn take_pending_writes(&mut self) -> Vec { + std::mem::take(&mut self.pending_writes) + } + + /// Get the number of pending workspace writes. + pub fn pending_writes_count(&self) -> usize { + self.pending_writes.len() + } + + /// Log a message (delegates to base). + pub fn log( + &mut self, + level: LogLevel, + message: String, + ) -> Result<(), crate::tools::wasm::WasmError> { + self.base.log(level, message) + } + + /// Get current timestamp in milliseconds (delegates to base). + pub fn now_millis(&self) -> u64 { + self.base.now_millis() + } + + /// Read from workspace (delegates to base). + pub fn workspace_read( + &self, + path: &str, + ) -> Result, crate::tools::wasm::WasmError> { + // Prefix the path with channel namespace before reading + let full_path = self.capabilities.prefix_workspace_path(path); + self.base.workspace_read(&full_path) + } + + /// Check if a secret exists (delegates to base). + pub fn secret_exists(&self, name: &str) -> bool { + self.base.secret_exists(name) + } + + /// Check if HTTP is allowed (delegates to base). + pub fn check_http_allowed(&self, url: &str, method: &str) -> Result<(), String> { + self.base.check_http_allowed(url, method) + } + + /// Record an HTTP request (delegates to base). + pub fn record_http_request(&mut self) -> Result<(), String> { + self.base.record_http_request() + } + + /// Take logs (delegates to base). + pub fn take_logs(&mut self) -> Vec { + self.base.take_logs() + } +} + +/// Rate limiter for channel message emission. +/// +/// Tracks emission rates across multiple executions. +pub struct ChannelEmitRateLimiter { + config: EmitRateLimitConfig, + minute_window: RateWindow, + hour_window: RateWindow, +} + +struct RateWindow { + count: u32, + window_start: u64, + window_duration_ms: u64, +} + +impl RateWindow { + fn new(duration_ms: u64) -> Self { + Self { + count: 0, + window_start: 0, + window_duration_ms: duration_ms, + } + } + + fn check_and_record(&mut self, now_ms: u64, limit: u32) -> bool { + // Reset window if expired + if now_ms.saturating_sub(self.window_start) > self.window_duration_ms { + self.count = 0; + self.window_start = now_ms; + } + + if self.count >= limit { + return false; + } + + self.count += 1; + true + } +} + +#[allow(dead_code)] +impl ChannelEmitRateLimiter { + /// Create a new rate limiter with the given config. + pub fn new(config: EmitRateLimitConfig) -> Self { + Self { + config, + minute_window: RateWindow::new(60_000), // 1 minute + hour_window: RateWindow::new(3_600_000), // 1 hour + } + } + + /// Check if an emit is allowed and record it if so. + /// + /// Returns true if the emit is allowed, false if rate limited. + pub fn check_and_record(&mut self) -> bool { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + // Check both windows + let minute_ok = self + .minute_window + .check_and_record(now, self.config.messages_per_minute); + let hour_ok = self + .hour_window + .check_and_record(now, self.config.messages_per_hour); + + minute_ok && hour_ok + } + + /// Get the current emission count for the minute window. + pub fn minute_count(&self) -> u32 { + self.minute_window.count + } + + /// Get the current emission count for the hour window. + pub fn hour_count(&self) -> u32 { + self.hour_window.count + } +} + +#[cfg(test)] +mod tests { + use crate::channels::wasm::capabilities::{ChannelCapabilities, EmitRateLimitConfig}; + use crate::channels::wasm::host::{ + ChannelEmitRateLimiter, ChannelHostState, EmittedMessage, MAX_EMITS_PER_EXECUTION, + }; + + #[test] + fn test_emit_message_basic() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + let msg = EmittedMessage::new("user123", "Hello, world!"); + state.emit_message(msg).unwrap(); + + assert_eq!(state.emitted_count(), 1); + + let messages = state.take_emitted_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].user_id, "user123"); + assert_eq!(messages[0].content, "Hello, world!"); + + // Queue should be cleared + assert_eq!(state.emitted_count(), 0); + } + + #[test] + fn test_emit_message_with_metadata() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + let msg = EmittedMessage::new("user123", "Hello") + .with_user_name("John Doe") + .with_thread_id("thread-1") + .with_metadata(r#"{"key": "value"}"#); + + state.emit_message(msg).unwrap(); + + let messages = state.take_emitted_messages(); + assert_eq!(messages[0].user_name, Some("John Doe".to_string())); + assert_eq!(messages[0].thread_id, Some("thread-1".to_string())); + assert_eq!(messages[0].metadata_json, r#"{"key": "value"}"#); + } + + #[test] + fn test_emit_per_execution_limit() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + // Fill up to limit + for i in 0..MAX_EMITS_PER_EXECUTION { + let msg = EmittedMessage::new("user", format!("Message {}", i)); + state.emit_message(msg).unwrap(); + } + + // This should be dropped silently + let msg = EmittedMessage::new("user", "Should be dropped"); + state.emit_message(msg).unwrap(); + + assert_eq!(state.emitted_count(), MAX_EMITS_PER_EXECUTION); + assert_eq!(state.emits_dropped(), 1); + } + + #[test] + fn test_workspace_write_prefixing() { + let caps = ChannelCapabilities::for_channel("slack"); + let mut state = ChannelHostState::new("slack", caps); + + state + .workspace_write("state.json", "{}".to_string()) + .unwrap(); + + let writes = state.take_pending_writes(); + assert_eq!(writes.len(), 1); + assert_eq!(writes[0].path, "channels/slack/state.json"); + } + + #[test] + fn test_workspace_write_path_traversal_blocked() { + let caps = ChannelCapabilities::for_channel("slack"); + let mut state = ChannelHostState::new("slack", caps); + + // Try to escape namespace + let result = state.workspace_write("../secrets.json", "{}".to_string()); + assert!(result.is_err()); + + // Absolute path + let result = state.workspace_write("/etc/passwd", "{}".to_string()); + assert!(result.is_err()); + } + + #[test] + fn test_rate_limiter_basic() { + let config = EmitRateLimitConfig { + messages_per_minute: 10, + messages_per_hour: 100, + }; + let mut limiter = ChannelEmitRateLimiter::new(config); + + // Should allow 10 messages + for _ in 0..10 { + assert!(limiter.check_and_record()); + } + + // 11th should be blocked + assert!(!limiter.check_and_record()); + } + + #[test] + fn test_channel_name() { + let caps = ChannelCapabilities::for_channel("telegram"); + let state = ChannelHostState::new("telegram", caps); + + assert_eq!(state.channel_name(), "telegram"); + } +} diff --git a/src/channels/wasm/loader.rs b/src/channels/wasm/loader.rs new file mode 100644 index 00000000..2d91e288 --- /dev/null +++ b/src/channels/wasm/loader.rs @@ -0,0 +1,360 @@ +//! WASM channel loader for loading channels from files or directories. +//! +//! Loads WASM channel modules from the filesystem (default: ~/.near-agent/channels/). +//! Each channel consists of: +//! - `.wasm` - The compiled WASM component +//! - `.capabilities.json` - Channel capabilities and configuration + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use tokio::fs; + +use crate::channels::wasm::capabilities::ChannelCapabilities; +use crate::channels::wasm::error::WasmChannelError; +use crate::channels::wasm::runtime::WasmChannelRuntime; +use crate::channels::wasm::schema::ChannelCapabilitiesFile; +use crate::channels::wasm::wrapper::WasmChannel; + +/// Loads WASM channels from the filesystem. +pub struct WasmChannelLoader { + runtime: Arc, +} + +impl WasmChannelLoader { + /// Create a new loader with the given runtime. + pub fn new(runtime: Arc) -> Self { + Self { runtime } + } + + /// Load a single WASM channel from a file pair. + /// + /// Expects: + /// - `wasm_path`: Path to the `.wasm` file + /// - `capabilities_path`: Path to the `.capabilities.json` file (optional) + /// + /// If no capabilities file is provided, the channel gets minimal capabilities. + pub async fn load_from_files( + &self, + name: &str, + wasm_path: &Path, + capabilities_path: Option<&Path>, + ) -> Result { + // Validate name + if name.is_empty() || name.contains('/') || name.contains('\\') || name.contains("..") { + return Err(WasmChannelError::InvalidName(name.to_string())); + } + + // Read WASM bytes + if !wasm_path.exists() { + return Err(WasmChannelError::WasmNotFound(wasm_path.to_path_buf())); + } + let wasm_bytes = fs::read(wasm_path).await?; + + // Read capabilities file + let (capabilities, config_json, description) = if let Some(cap_path) = capabilities_path { + if cap_path.exists() { + let cap_bytes = fs::read(cap_path).await?; + let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes) + .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?; + + let caps = cap_file.to_capabilities(); + let config = cap_file.config_json(); + let desc = cap_file.description.clone(); + + (caps, config, desc) + } else { + tracing::warn!( + path = %cap_path.display(), + "Capabilities file not found, using defaults" + ); + ( + ChannelCapabilities::for_channel(name), + "{}".to_string(), + None, + ) + } + } else { + ( + ChannelCapabilities::for_channel(name), + "{}".to_string(), + None, + ) + }; + + // Prepare the module + let prepared = self + .runtime + .prepare(name, &wasm_bytes, None, description) + .await?; + + // Create the channel + let channel = WasmChannel::new(self.runtime.clone(), prepared, capabilities, config_json); + + tracing::info!( + name = name, + wasm_path = %wasm_path.display(), + "Loaded WASM channel from file" + ); + + Ok(channel) + } + + /// Load all WASM channels from a directory. + /// + /// Scans the directory for `*.wasm` files and loads each one, looking for + /// a matching `*.capabilities.json` sidecar file. + /// + /// # Directory Layout + /// + /// ```text + /// channels/ + /// ├── slack.wasm <- Channel WASM component + /// ├── slack.capabilities.json <- Capabilities (optional) + /// ├── telegram.wasm + /// └── telegram.capabilities.json + /// ``` + pub async fn load_from_dir(&self, dir: &Path) -> Result { + if !dir.is_dir() { + return Err(WasmChannelError::Io(std::io::Error::new( + std::io::ErrorKind::NotADirectory, + format!("{} is not a directory", dir.display()), + ))); + } + + let mut results = LoadResults::default(); + let mut entries = fs::read_dir(dir).await?; + + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + + // Only process .wasm files + if path.extension().and_then(|e| e.to_str()) != Some("wasm") { + continue; + } + + // Extract channel name from filename + let name = match path.file_stem().and_then(|s| s.to_str()) { + Some(n) => n.to_string(), + None => { + results.errors.push(( + path.clone(), + WasmChannelError::InvalidName("invalid filename".to_string()), + )); + continue; + } + }; + + // Look for sidecar capabilities file + let cap_path = path.with_extension("capabilities.json"); + let cap_path_option = if cap_path.exists() { + Some(cap_path.as_path()) + } else { + None + }; + + match self.load_from_files(&name, &path, cap_path_option).await { + Ok(channel) => { + results.loaded.push(channel); + } + Err(e) => { + tracing::error!( + name = name, + path = %path.display(), + error = %e, + "Failed to load WASM channel" + ); + results.errors.push((path, e)); + } + } + } + + if !results.loaded.is_empty() { + tracing::info!( + count = results.loaded.len(), + channels = ?results.loaded.iter().map(|c| c.channel_name()).collect::>(), + "Loaded WASM channels from directory" + ); + } + + Ok(results) + } +} + +/// Results from loading multiple channels. +#[derive(Default)] +pub struct LoadResults { + /// Successfully loaded channels. + pub loaded: Vec, + + /// Errors encountered (path, error). + pub errors: Vec<(PathBuf, WasmChannelError)>, +} + +impl LoadResults { + /// Check if all channels loaded successfully. + pub fn all_succeeded(&self) -> bool { + self.errors.is_empty() + } + + /// Get the count of successfully loaded channels. + pub fn success_count(&self) -> usize { + self.loaded.len() + } + + /// Get the count of failed channels. + pub fn error_count(&self) -> usize { + self.errors.len() + } + + /// Take ownership of loaded channels. + pub fn take_channels(self) -> Vec { + self.loaded + } +} + +/// Discover WASM channel files in a directory without loading them. +/// +/// Returns a map of channel name -> (wasm_path, capabilities_path). +#[allow(dead_code)] +pub async fn discover_channels( + dir: &Path, +) -> Result, std::io::Error> { + let mut channels = HashMap::new(); + + if !dir.is_dir() { + return Ok(channels); + } + + let mut entries = fs::read_dir(dir).await?; + + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + + if path.extension().and_then(|e| e.to_str()) != Some("wasm") { + continue; + } + + let name = match path.file_stem().and_then(|s| s.to_str()) { + Some(n) => n.to_string(), + None => continue, + }; + + let cap_path = path.with_extension("capabilities.json"); + + channels.insert( + name, + DiscoveredChannel { + wasm_path: path, + capabilities_path: if cap_path.exists() { + Some(cap_path) + } else { + None + }, + }, + ); + } + + Ok(channels) +} + +/// A discovered WASM channel (not yet loaded). +#[derive(Debug)] +pub struct DiscoveredChannel { + /// Path to the WASM file. + pub wasm_path: PathBuf, + + /// Path to the capabilities file (if present). + pub capabilities_path: Option, +} + +/// Get the default channels directory path. +/// +/// Returns ~/.near-agent/channels/ +#[allow(dead_code)] +pub fn default_channels_dir() -> PathBuf { + dirs::home_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join(".near-agent") + .join("channels") +} + +#[cfg(test)] +mod tests { + use std::io::Write; + + use tempfile::TempDir; + + use crate::channels::wasm::loader::{WasmChannelLoader, discover_channels}; + use crate::channels::wasm::runtime::{WasmChannelRuntime, WasmChannelRuntimeConfig}; + use std::sync::Arc; + + #[tokio::test] + async fn test_discover_channels_empty_dir() { + let dir = TempDir::new().unwrap(); + let channels = discover_channels(dir.path()).await.unwrap(); + assert!(channels.is_empty()); + } + + #[tokio::test] + async fn test_discover_channels_with_wasm() { + let dir = TempDir::new().unwrap(); + + // Create a fake .wasm file + let wasm_path = dir.path().join("slack.wasm"); + std::fs::File::create(&wasm_path).unwrap(); + + let channels = discover_channels(dir.path()).await.unwrap(); + assert_eq!(channels.len(), 1); + assert!(channels.contains_key("slack")); + assert!(channels["slack"].capabilities_path.is_none()); + } + + #[tokio::test] + async fn test_discover_channels_with_capabilities() { + let dir = TempDir::new().unwrap(); + + // Create wasm and capabilities files + std::fs::File::create(dir.path().join("telegram.wasm")).unwrap(); + let mut cap_file = + std::fs::File::create(dir.path().join("telegram.capabilities.json")).unwrap(); + cap_file.write_all(b"{}").unwrap(); + + let channels = discover_channels(dir.path()).await.unwrap(); + assert_eq!(channels.len(), 1); + assert!(channels["telegram"].capabilities_path.is_some()); + } + + #[tokio::test] + async fn test_discover_channels_ignores_non_wasm() { + let dir = TempDir::new().unwrap(); + + // Create non-wasm files + std::fs::File::create(dir.path().join("readme.md")).unwrap(); + std::fs::File::create(dir.path().join("config.json")).unwrap(); + std::fs::File::create(dir.path().join("channel.wasm")).unwrap(); + + let channels = discover_channels(dir.path()).await.unwrap(); + assert_eq!(channels.len(), 1); + assert!(channels.contains_key("channel")); + } + + #[tokio::test] + async fn test_loader_invalid_name() { + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); + let loader = WasmChannelLoader::new(runtime); + + let dir = TempDir::new().unwrap(); + let wasm_path = dir.path().join("test.wasm"); + + // Invalid name with path separator + let result = loader.load_from_files("../escape", &wasm_path, None).await; + assert!(result.is_err()); + + // Empty name + let result = loader.load_from_files("", &wasm_path, None).await; + assert!(result.is_err()); + } +} diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs new file mode 100644 index 00000000..757969f9 --- /dev/null +++ b/src/channels/wasm/mod.rs @@ -0,0 +1,102 @@ +//! WASM-extensible channel system. +//! +//! This module provides a runtime for executing WASM-based channels using a +//! Host-Managed Event Loop pattern. The host (Rust) manages infrastructure +//! (HTTP server, polling), while WASM modules define channel behavior through +//! callbacks. +//! +//! # Architecture +//! +//! ```text +//! ┌─────────────────────────────────────────────────────────────────────────────────┐ +//! │ Host-Managed Event Loop │ +//! │ │ +//! │ ┌─────────────┐ ┌──────────────┐ ┌──────────────┐ │ +//! │ │ HTTP │ │ Polling │ │ Timer │ │ +//! │ │ Router │ │ Scheduler │ │ Scheduler │ │ +//! │ └──────┬──────┘ └──────┬───────┘ └──────┬───────┘ │ +//! │ │ │ │ │ +//! │ └───────────────────┴────────────────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌─────────────────┐ │ +//! │ │ Event Router │ │ +//! │ └────────┬────────┘ │ +//! │ │ │ +//! │ ┌──────────────────┼──────────────────┐ │ +//! │ ▼ ▼ ▼ │ +//! │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ +//! │ │ on_http_req │ │ on_poll │ │ on_respond │ WASM Exports │ +//! │ └─────────────┘ └─────────────┘ └─────────────┘ │ +//! │ │ │ │ │ +//! │ └──────────────────┴──────────────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌─────────────────┐ │ +//! │ │ Host Imports │ │ +//! │ │ emit_message │──────────▶ MessageStream │ +//! │ │ http_request │ │ +//! │ │ log, etc. │ │ +//! │ └─────────────────┘ │ +//! └─────────────────────────────────────────────────────────────────────────────────┘ +//! ``` +//! +//! # Key Design Decisions +//! +//! 1. **Fresh Instance Per Callback** (NEAR Pattern) - Full isolation, no shared mutable state +//! 2. **Host Manages Infrastructure** - HTTP server, polling, timing in Rust +//! 3. **WASM Defines Behavior** - Callbacks for events, message parsing, response handling +//! 4. **Reuse Tool Runtime** - Share Wasmtime engine, extend capabilities +//! +//! # Security Model +//! +//! | Threat | Mitigation | +//! |--------|------------| +//! | Path hijacking | `allowed_paths` restricts registrable endpoints | +//! | Token exposure | Injected at host boundary, WASM never sees | +//! | State pollution | Fresh instance per callback | +//! | Workspace escape | Paths prefixed with `channels//` | +//! | Message spam | Rate limiting on `emit_message` | +//! | Resource exhaustion | Fuel metering, memory limits, callback timeout | +//! | Polling abuse | Minimum 30s interval enforced | +//! +//! # Example Usage +//! +//! ```ignore +//! use near_agent::channels::wasm::{WasmChannelLoader, WasmChannelRuntime}; +//! +//! // Create runtime (can share engine with tool runtime) +//! let runtime = WasmChannelRuntime::new(config)?; +//! +//! // Load channels from directory +//! let loader = WasmChannelLoader::new(runtime); +//! let channels = loader.load_from_dir(Path::new("~/.near-agent/channels/")).await?; +//! +//! // Add to channel manager +//! for channel in channels { +//! manager.add(Box::new(channel)); +//! } +//! ``` + +mod capabilities; +mod error; +mod host; +mod loader; +mod router; +mod runtime; +mod schema; +mod wrapper; + +// Core types +pub use capabilities::{ChannelCapabilities, EmitRateLimitConfig, HttpEndpointConfig, PollConfig}; +pub use error::WasmChannelError; +pub use host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage}; +pub use loader::{ + DiscoveredChannel, LoadResults, WasmChannelLoader, default_channels_dir, discover_channels, +}; +pub use router::{ + RegisteredEndpoint, WasmChannelRouter, WasmChannelServer, create_wasm_channel_router, +}; +pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig}; +pub use schema::{ChannelCapabilitiesFile, ChannelConfig}; +pub use wrapper::{HttpResponse, WasmChannel}; diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs new file mode 100644 index 00000000..e7d66f6a --- /dev/null +++ b/src/channels/wasm/router.rs @@ -0,0 +1,480 @@ +//! HTTP router for WASM channel webhooks. +//! +//! Routes incoming HTTP requests to the appropriate WASM channel based on +//! registered paths. Handles secret validation at the host level. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; + +use axum::{ + Json, Router, + body::Bytes, + extract::{Path, Query, State}, + http::{HeaderMap, Method, StatusCode}, + response::IntoResponse, + routing::{get, post}, +}; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; + +use crate::channels::wasm::wrapper::WasmChannel; + +/// A registered HTTP endpoint for a WASM channel. +#[derive(Debug, Clone)] +pub struct RegisteredEndpoint { + /// Channel name that owns this endpoint. + pub channel_name: String, + /// HTTP path (e.g., "/webhook/slack"). + pub path: String, + /// Allowed HTTP methods. + pub methods: Vec, + /// Whether secret validation is required. + pub require_secret: bool, +} + +/// Router for WASM channel HTTP endpoints. +pub struct WasmChannelRouter { + /// Registered channels by name. + channels: RwLock>>, + /// Path to channel mapping for fast lookup. + path_to_channel: RwLock>, + /// Expected webhook secrets by channel name. + secrets: RwLock>, +} + +impl WasmChannelRouter { + /// Create a new router. + pub fn new() -> Self { + Self { + channels: RwLock::new(HashMap::new()), + path_to_channel: RwLock::new(HashMap::new()), + secrets: RwLock::new(HashMap::new()), + } + } + + /// Register a channel with its endpoints. + pub async fn register( + &self, + channel: Arc, + endpoints: Vec, + secret: Option, + ) { + let name = channel.channel_name().to_string(); + + // Store the channel + self.channels.write().await.insert(name.clone(), channel); + + // Register path mappings + let mut path_map = self.path_to_channel.write().await; + for endpoint in endpoints { + path_map.insert(endpoint.path.clone(), name.clone()); + tracing::info!( + channel = %name, + path = %endpoint.path, + methods = ?endpoint.methods, + "Registered WASM channel HTTP endpoint" + ); + } + + // Store secret if provided + if let Some(s) = secret { + self.secrets.write().await.insert(name, s); + } + } + + /// Unregister a channel and its endpoints. + pub async fn unregister(&self, channel_name: &str) { + self.channels.write().await.remove(channel_name); + self.secrets.write().await.remove(channel_name); + + // Remove all paths for this channel + self.path_to_channel + .write() + .await + .retain(|_, name| name != channel_name); + + tracing::info!( + channel = %channel_name, + "Unregistered WASM channel" + ); + } + + /// Get the channel for a given path. + pub async fn get_channel_for_path(&self, path: &str) -> Option> { + let path_map = self.path_to_channel.read().await; + let channel_name = path_map.get(path)?; + + self.channels.read().await.get(channel_name).cloned() + } + + /// Validate a secret for a channel. + pub async fn validate_secret(&self, channel_name: &str, provided: &str) -> bool { + let secrets = self.secrets.read().await; + match secrets.get(channel_name) { + Some(expected) => expected == provided, + None => true, // No secret required + } + } + + /// Check if a channel requires a secret. + pub async fn requires_secret(&self, channel_name: &str) -> bool { + self.secrets.read().await.contains_key(channel_name) + } + + /// List all registered channels. + pub async fn list_channels(&self) -> Vec { + self.channels.read().await.keys().cloned().collect() + } + + /// List all registered paths. + pub async fn list_paths(&self) -> Vec { + self.path_to_channel.read().await.keys().cloned().collect() + } +} + +impl Default for WasmChannelRouter { + fn default() -> Self { + Self::new() + } +} + +/// Shared state for the HTTP server. +#[allow(dead_code)] +#[derive(Clone)] +pub struct RouterState { + router: Arc, +} + +impl RouterState { + pub fn new(router: Arc) -> Self { + Self { router } + } +} + +/// Webhook request body for WASM channels. +#[allow(dead_code)] +#[derive(Debug, Deserialize)] +pub struct WasmWebhookRequest { + /// Optional secret for authentication. + #[serde(default)] + pub secret: Option, +} + +/// Health response. +#[allow(dead_code)] +#[derive(Debug, Serialize)] +struct HealthResponse { + status: String, + channels: Vec, +} + +/// Handler for health check endpoint. +#[allow(dead_code)] +async fn health_handler(State(state): State) -> impl IntoResponse { + let channels = state.router.list_channels().await; + Json(HealthResponse { + status: "healthy".to_string(), + channels, + }) +} + +/// Generic webhook handler that routes to the appropriate WASM channel. +#[allow(dead_code)] +async fn webhook_handler( + State(state): State, + method: Method, + Path(path): Path, + Query(query): Query>, + headers: HeaderMap, + body: Bytes, +) -> impl IntoResponse { + let full_path = format!("/webhook/{}", path); + + // Find the channel for this path + let channel = match state.router.get_channel_for_path(&full_path).await { + Some(c) => c, + None => { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ + "error": "Channel not found for path", + "path": full_path + })), + ); + } + }; + + let channel_name = channel.channel_name(); + + // Check if secret is required + if state.router.requires_secret(channel_name).await { + // Try to get secret from query param or header + let provided_secret = query.get("secret").cloned().or_else(|| { + headers + .get("X-Webhook-Secret") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) + }); + + match provided_secret { + Some(secret) => { + if !state.router.validate_secret(channel_name, &secret).await { + return ( + StatusCode::UNAUTHORIZED, + Json(serde_json::json!({ + "error": "Invalid webhook secret" + })), + ); + } + } + None => { + return ( + StatusCode::UNAUTHORIZED, + Json(serde_json::json!({ + "error": "Webhook secret required" + })), + ); + } + } + } + + // Convert headers to HashMap + let headers_map: HashMap = headers + .iter() + .filter_map(|(k, v)| { + v.to_str() + .ok() + .map(|v| (k.as_str().to_string(), v.to_string())) + }) + .collect(); + + // Call the WASM channel + let secret_validated = state.router.requires_secret(channel_name).await; + match channel + .call_on_http_request( + method.as_str(), + &full_path, + &headers_map, + &query, + &body, + secret_validated, + ) + .await + { + Ok(response) => { + let status = + StatusCode::from_u16(response.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + + // Build response with headers + let body_json: serde_json::Value = serde_json::from_slice(&response.body) + .unwrap_or_else(|_| { + serde_json::json!({ + "raw": String::from_utf8_lossy(&response.body).to_string() + }) + }); + + (status, Json(body_json)) + } + Err(e) => { + tracing::error!( + channel = %channel_name, + error = %e, + "WASM channel callback failed" + ); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ + "error": "Channel callback failed", + "details": e.to_string() + })), + ) + } + } +} + +/// Create an Axum router for WASM channel webhooks. +/// +/// This router can be merged with the existing HTTP channel router. +#[allow(dead_code)] +pub fn create_wasm_channel_router(router: Arc) -> Router { + let state = RouterState::new(router); + + Router::new() + .route("/wasm-channels/health", get(health_handler)) + // Catch-all for webhook paths + .route("/webhook/*path", get(webhook_handler)) + .route("/webhook/*path", post(webhook_handler)) + .with_state(state) +} + +/// HTTP server for WASM channel webhooks. +#[allow(dead_code)] +pub struct WasmChannelServer { + router: Arc, +} + +#[allow(dead_code)] +impl WasmChannelServer { + /// Create a new server. + pub fn new(router: Arc) -> Self { + Self { router } + } + + /// Start the HTTP server. + /// + /// Returns a handle that can be used to shut down the server. + pub async fn start( + &self, + addr: SocketAddr, + ) -> Result, std::io::Error> { + let app = create_wasm_channel_router(self.router.clone()); + + let listener = tokio::net::TcpListener::bind(addr).await?; + + tracing::info!( + addr = %addr, + "WASM channel HTTP server started" + ); + + let handle = tokio::spawn(async move { + if let Err(e) = axum::serve(listener, app).await { + tracing::error!("WASM channel HTTP server error: {}", e); + } + }); + + Ok(handle) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use crate::channels::wasm::capabilities::ChannelCapabilities; + use crate::channels::wasm::router::{RegisteredEndpoint, WasmChannelRouter}; + use crate::channels::wasm::runtime::{ + PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig, + }; + use crate::channels::wasm::wrapper::WasmChannel; + use crate::tools::wasm::ResourceLimits; + + fn create_test_channel(name: &str) -> Arc { + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); + + let prepared = Arc::new(PreparedChannelModule { + name: name.to_string(), + description: format!("Test channel: {}", name), + component_bytes: Vec::new(), + limits: ResourceLimits::default(), + }); + + let capabilities = + ChannelCapabilities::for_channel(name).with_path(format!("/webhook/{}", name)); + + Arc::new(WasmChannel::new( + runtime, + prepared, + capabilities, + "{}".to_string(), + )) + } + + #[tokio::test] + async fn test_router_register_and_lookup() { + let router = WasmChannelRouter::new(); + let channel = create_test_channel("slack"); + + let endpoints = vec![RegisteredEndpoint { + channel_name: "slack".to_string(), + path: "/webhook/slack".to_string(), + methods: vec!["POST".to_string()], + require_secret: true, + }]; + + router + .register(channel, endpoints, Some("secret123".to_string())) + .await; + + // Should find channel by path + let found = router.get_channel_for_path("/webhook/slack").await; + assert!(found.is_some()); + assert_eq!(found.unwrap().channel_name(), "slack"); + + // Should not find non-existent path + let not_found = router.get_channel_for_path("/webhook/telegram").await; + assert!(not_found.is_none()); + } + + #[tokio::test] + async fn test_router_secret_validation() { + let router = WasmChannelRouter::new(); + let channel = create_test_channel("slack"); + + router + .register(channel, vec![], Some("secret123".to_string())) + .await; + + // Correct secret + assert!(router.validate_secret("slack", "secret123").await); + + // Wrong secret + assert!(!router.validate_secret("slack", "wrong").await); + + // Channel without secret always validates + let channel2 = create_test_channel("telegram"); + router.register(channel2, vec![], None).await; + assert!(router.validate_secret("telegram", "anything").await); + } + + #[tokio::test] + async fn test_router_unregister() { + let router = WasmChannelRouter::new(); + let channel = create_test_channel("slack"); + + let endpoints = vec![RegisteredEndpoint { + channel_name: "slack".to_string(), + path: "/webhook/slack".to_string(), + methods: vec!["POST".to_string()], + require_secret: false, + }]; + + router.register(channel, endpoints, None).await; + + // Should exist + assert!( + router + .get_channel_for_path("/webhook/slack") + .await + .is_some() + ); + + // Unregister + router.unregister("slack").await; + + // Should no longer exist + assert!( + router + .get_channel_for_path("/webhook/slack") + .await + .is_none() + ); + } + + #[tokio::test] + async fn test_router_list_channels() { + let router = WasmChannelRouter::new(); + + let channel1 = create_test_channel("slack"); + let channel2 = create_test_channel("telegram"); + + router.register(channel1, vec![], None).await; + router.register(channel2, vec![], None).await; + + let channels = router.list_channels().await; + assert_eq!(channels.len(), 2); + assert!(channels.contains(&"slack".to_string())); + assert!(channels.contains(&"telegram".to_string())); + } +} diff --git a/src/channels/wasm/runtime.rs b/src/channels/wasm/runtime.rs new file mode 100644 index 00000000..62f9d19c --- /dev/null +++ b/src/channels/wasm/runtime.rs @@ -0,0 +1,285 @@ +//! WASM channel runtime for managing compiled channel components. +//! +//! Similar to tool runtime, follows the principle: compile once at registration, +//! instantiate fresh per callback execution. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::RwLock; +use wasmtime::{Config, Engine, OptLevel}; + +use crate::channels::wasm::error::WasmChannelError; +use crate::tools::wasm::{FuelConfig, ResourceLimits}; + +/// Configuration for the WASM channel runtime. +#[derive(Debug, Clone)] +pub struct WasmChannelRuntimeConfig { + /// Default resource limits for channels. + pub default_limits: ResourceLimits, + /// Fuel configuration. + pub fuel_config: FuelConfig, + /// Whether to cache compiled modules. + pub cache_compiled: bool, + /// Directory for compiled module cache. + pub cache_dir: Option, + /// Cranelift optimization level. + pub optimization_level: OptLevel, + /// Default callback timeout. + pub callback_timeout: Duration, +} + +impl Default for WasmChannelRuntimeConfig { + fn default() -> Self { + Self { + default_limits: ResourceLimits { + // Channels may need more memory for message buffering + memory_bytes: 50 * 1024 * 1024, // 50 MB + fuel: 10_000_000, + timeout: Duration::from_secs(60), + }, + fuel_config: FuelConfig::default(), + cache_compiled: true, + cache_dir: None, + optimization_level: OptLevel::Speed, + callback_timeout: Duration::from_secs(30), + } + } +} + +impl WasmChannelRuntimeConfig { + /// Create a minimal config for testing. + pub fn for_testing() -> Self { + Self { + default_limits: ResourceLimits { + memory_bytes: 5 * 1024 * 1024, // 5 MB + fuel: 1_000_000, + timeout: Duration::from_secs(5), + }, + fuel_config: FuelConfig::with_limit(1_000_000), + cache_compiled: false, + cache_dir: None, + optimization_level: OptLevel::None, // Faster compilation for tests + callback_timeout: Duration::from_secs(5), + } + } +} + +/// A compiled WASM channel component ready for instantiation. +#[derive(Debug)] +pub struct PreparedChannelModule { + /// Channel name. + pub name: String, + /// Channel description. + pub description: String, + /// Compiled component bytes (public for testing, otherwise use component_bytes()). + pub(crate) component_bytes: Vec, + /// Resource limits for this channel. + pub limits: ResourceLimits, +} + +impl PreparedChannelModule { + /// Get the compiled component bytes. + pub fn component_bytes(&self) -> &[u8] { + &self.component_bytes + } + + /// Create a PreparedChannelModule for testing purposes. + /// + /// Creates a module with no actual WASM bytes, suitable for testing + /// channel infrastructure without requiring a real WASM component. + pub fn for_testing(name: impl Into, description: impl Into) -> Self { + Self { + name: name.into(), + description: description.into(), + component_bytes: Vec::new(), + limits: ResourceLimits::default(), + } + } +} + +/// WASM channel runtime. +/// +/// Manages the Wasmtime engine and a cache of prepared channel modules. +pub struct WasmChannelRuntime { + /// Wasmtime engine with configured settings. + engine: Engine, + /// Runtime configuration. + config: WasmChannelRuntimeConfig, + /// Cache of prepared modules by name. + modules: RwLock>>, +} + +impl WasmChannelRuntime { + /// Create a new runtime with the given configuration. + pub fn new(config: WasmChannelRuntimeConfig) -> Result { + let mut wasmtime_config = Config::new(); + + // Enable fuel consumption for CPU limiting + if config.fuel_config.enabled { + wasmtime_config.consume_fuel(true); + } + + // Enable epoch interruption as a backup timeout mechanism + wasmtime_config.epoch_interruption(true); + + // Enable component model (WASI Preview 2) + wasmtime_config.wasm_component_model(true); + + // Disable threads (simplifies security model) + wasmtime_config.wasm_threads(false); + + // Set optimization level + wasmtime_config.cranelift_opt_level(config.optimization_level); + + // Disable debug info in production + wasmtime_config.debug_info(false); + + let engine = Engine::new(&wasmtime_config).map_err(|e| { + WasmChannelError::Config(format!("Failed to create Wasmtime engine: {}", e)) + })?; + + Ok(Self { + engine, + config, + modules: RwLock::new(HashMap::new()), + }) + } + + /// Get the Wasmtime engine. + pub fn engine(&self) -> &Engine { + &self.engine + } + + /// Get the runtime configuration. + pub fn config(&self) -> &WasmChannelRuntimeConfig { + &self.config + } + + /// Prepare a WASM channel component for execution. + /// + /// This validates and compiles the component. + /// The compiled component is cached for fast instantiation. + pub async fn prepare( + &self, + name: &str, + wasm_bytes: &[u8], + limits: Option, + description: Option, + ) -> Result, WasmChannelError> { + // Check if already prepared + if let Some(module) = self.modules.read().await.get(name) { + return Ok(Arc::clone(module)); + } + + let name = name.to_string(); + let wasm_bytes = wasm_bytes.to_vec(); + let engine = self.engine.clone(); + let default_limits = self.config.default_limits.clone(); + let desc = description.unwrap_or_else(|| format!("WASM channel: {}", name)); + + // Compile in blocking task (Wasmtime compilation is synchronous) + let prepared = tokio::task::spawn_blocking(move || { + // Validate and compile the component + let _component = wasmtime::component::Component::new(&engine, &wasm_bytes) + .map_err(|e| WasmChannelError::Compilation(e.to_string()))?; + + Ok::<_, WasmChannelError>(PreparedChannelModule { + name: name.clone(), + description: desc, + component_bytes: wasm_bytes, + limits: limits.unwrap_or(default_limits), + }) + }) + .await + .map_err(|e| { + WasmChannelError::Compilation(format!("Preparation task panicked: {}", e)) + })??; + + let prepared = Arc::new(prepared); + + // Cache the prepared module + if self.config.cache_compiled { + self.modules + .write() + .await + .insert(prepared.name.clone(), Arc::clone(&prepared)); + } + + tracing::info!( + name = %prepared.name, + "Prepared WASM channel for execution" + ); + + Ok(prepared) + } + + /// Get a prepared module by name. + pub async fn get(&self, name: &str) -> Option> { + self.modules.read().await.get(name).cloned() + } + + /// Remove a prepared module from the cache. + pub async fn remove(&self, name: &str) -> Option> { + self.modules.write().await.remove(name) + } + + /// List all prepared module names. + pub async fn list(&self) -> Vec { + self.modules.read().await.keys().cloned().collect() + } + + /// Clear all cached modules. + pub async fn clear(&self) { + self.modules.write().await.clear(); + } +} + +impl std::fmt::Debug for WasmChannelRuntime { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WasmChannelRuntime") + .field("config", &self.config) + .field("modules", &">") + .finish() + } +} + +#[cfg(test)] +mod tests { + use crate::channels::wasm::runtime::{WasmChannelRuntime, WasmChannelRuntimeConfig}; + + #[test] + fn test_runtime_config_default() { + let config = WasmChannelRuntimeConfig::default(); + assert!(config.cache_compiled); + assert!(config.fuel_config.enabled); + // Channels get more memory than tools + assert_eq!(config.default_limits.memory_bytes, 50 * 1024 * 1024); + } + + #[test] + fn test_runtime_config_for_testing() { + let config = WasmChannelRuntimeConfig::for_testing(); + assert!(!config.cache_compiled); + assert_eq!(config.default_limits.memory_bytes, 5 * 1024 * 1024); + } + + #[test] + fn test_runtime_creation() { + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = WasmChannelRuntime::new(config).unwrap(); + assert!(runtime.config().fuel_config.enabled); + } + + #[tokio::test] + async fn test_module_cache_operations() { + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = WasmChannelRuntime::new(config).unwrap(); + + // Initially empty + assert!(runtime.list().await.is_empty()); + assert!(runtime.get("test").await.is_none()); + } +} diff --git a/src/channels/wasm/schema.rs b/src/channels/wasm/schema.rs new file mode 100644 index 00000000..228c0214 --- /dev/null +++ b/src/channels/wasm/schema.rs @@ -0,0 +1,417 @@ +//! JSON schema for WASM channel capabilities files. +//! +//! External WASM channels declare their required capabilities via a sidecar JSON file +//! (e.g., `slack.capabilities.json`). This module defines the schema for those files +//! and provides conversion to runtime [`ChannelCapabilities`]. +//! +//! # Example Capabilities File +//! +//! ```json +//! { +//! "type": "channel", +//! "name": "slack", +//! "description": "Slack Events API channel", +//! "capabilities": { +//! "http": { +//! "allowlist": [ +//! { "host": "slack.com", "path_prefix": "/api/" } +//! ], +//! "credentials": { +//! "slack_bot": { +//! "secret_name": "slack_bot_token", +//! "location": { "type": "bearer" }, +//! "host_patterns": ["slack.com"] +//! } +//! } +//! }, +//! "secrets": { "allowed_names": ["slack_*"] }, +//! "channel": { +//! "allowed_paths": ["/webhook/slack"], +//! "allow_polling": false, +//! "workspace_prefix": "channels/slack/", +//! "emit_rate_limit": { "messages_per_minute": 100 } +//! } +//! }, +//! "config": { +//! "signing_secret_name": "slack_signing_secret" +//! } +//! } +//! ``` + +use std::collections::HashMap; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +use crate::channels::wasm::capabilities::{ + ChannelCapabilities, EmitRateLimitConfig, MIN_POLL_INTERVAL_MS, +}; +use crate::tools::wasm::{CapabilitiesFile as ToolCapabilitiesFile, RateLimitSchema}; + +/// Root schema for a channel capabilities JSON file. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ChannelCapabilitiesFile { + /// File type, must be "channel". + #[serde(default = "default_type")] + pub r#type: String, + + /// Channel name. + pub name: String, + + /// Channel description. + #[serde(default)] + pub description: Option, + + /// Capabilities (tool + channel specific). + #[serde(default)] + pub capabilities: ChannelCapabilitiesSchema, + + /// Channel-specific configuration passed to on_start. + #[serde(default)] + pub config: HashMap, +} + +fn default_type() -> String { + "channel".to_string() +} + +impl ChannelCapabilitiesFile { + /// Parse from JSON string. + pub fn from_json(json: &str) -> Result { + serde_json::from_str(json) + } + + /// Parse from JSON bytes. + pub fn from_bytes(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes) + } + + /// Convert to runtime ChannelCapabilities. + pub fn to_capabilities(&self) -> ChannelCapabilities { + self.capabilities.to_channel_capabilities(&self.name) + } + + /// Get the channel config as JSON string. + pub fn config_json(&self) -> String { + serde_json::to_string(&self.config).unwrap_or_else(|_| "{}".to_string()) + } +} + +/// Schema for channel capabilities. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ChannelCapabilitiesSchema { + /// Tool capabilities (HTTP, secrets, workspace_read). + #[serde(flatten)] + pub tool: Option, + + /// Channel-specific capabilities. + #[serde(default)] + pub channel: Option, +} + +impl ChannelCapabilitiesSchema { + /// Convert to runtime ChannelCapabilities. + pub fn to_channel_capabilities(&self, channel_name: &str) -> ChannelCapabilities { + let tool_caps = self + .tool + .as_ref() + .map(|t| t.to_capabilities()) + .unwrap_or_default(); + + let mut caps = + ChannelCapabilities::for_channel(channel_name).with_tool_capabilities(tool_caps); + + if let Some(channel) = &self.channel { + caps.allowed_paths = channel.allowed_paths.clone(); + caps.allow_polling = channel.allow_polling; + caps.min_poll_interval_ms = channel + .min_poll_interval_ms + .unwrap_or(MIN_POLL_INTERVAL_MS) + .max(MIN_POLL_INTERVAL_MS); + + if let Some(prefix) = &channel.workspace_prefix { + caps.workspace_prefix = prefix.clone(); + } + + if let Some(rate) = &channel.emit_rate_limit { + caps.emit_rate_limit = rate.to_emit_rate_limit(); + } + + if let Some(max_size) = channel.max_message_size { + caps.max_message_size = max_size; + } + + if let Some(timeout_secs) = channel.callback_timeout_secs { + caps.callback_timeout = Duration::from_secs(timeout_secs); + } + } + + caps + } +} + +/// Channel-specific capabilities schema. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ChannelSpecificCapabilitiesSchema { + /// HTTP paths the channel can register for webhooks. + #[serde(default)] + pub allowed_paths: Vec, + + /// Whether polling is allowed. + #[serde(default)] + pub allow_polling: bool, + + /// Minimum poll interval in milliseconds. + #[serde(default)] + pub min_poll_interval_ms: Option, + + /// Workspace prefix for storage (overrides default). + #[serde(default)] + pub workspace_prefix: Option, + + /// Rate limiting for emit_message. + #[serde(default)] + pub emit_rate_limit: Option, + + /// Maximum message content size in bytes. + #[serde(default)] + pub max_message_size: Option, + + /// Callback timeout in seconds. + #[serde(default)] + pub callback_timeout_secs: Option, +} + +/// Schema for emit rate limiting. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EmitRateLimitSchema { + /// Maximum messages per minute. + #[serde(default = "default_messages_per_minute")] + pub messages_per_minute: u32, + + /// Maximum messages per hour. + #[serde(default = "default_messages_per_hour")] + pub messages_per_hour: u32, +} + +fn default_messages_per_minute() -> u32 { + 100 +} + +fn default_messages_per_hour() -> u32 { + 5000 +} + +impl EmitRateLimitSchema { + fn to_emit_rate_limit(&self) -> EmitRateLimitConfig { + EmitRateLimitConfig { + messages_per_minute: self.messages_per_minute, + messages_per_hour: self.messages_per_hour, + } + } +} + +impl From for EmitRateLimitSchema { + fn from(schema: RateLimitSchema) -> Self { + Self { + messages_per_minute: schema.requests_per_minute, + messages_per_hour: schema.requests_per_hour, + } + } +} + +/// Channel configuration returned by on_start. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ChannelConfig { + /// Display name for the channel. + pub display_name: String, + + /// HTTP endpoints to register. + #[serde(default)] + pub http_endpoints: Vec, + + /// Polling configuration. + #[serde(default)] + pub poll: Option, +} + +impl Default for ChannelConfig { + fn default() -> Self { + Self { + display_name: "WASM Channel".to_string(), + http_endpoints: Vec::new(), + poll: None, + } + } +} + +/// HTTP endpoint configuration schema. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HttpEndpointConfigSchema { + /// Path to register. + pub path: String, + + /// HTTP methods to accept. + #[serde(default)] + pub methods: Vec, + + /// Whether secret validation is required. + #[serde(default)] + pub require_secret: bool, +} + +/// Polling configuration schema. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PollConfigSchema { + /// Polling interval in milliseconds. + pub interval_ms: u32, + + /// Whether polling is enabled. + #[serde(default)] + pub enabled: bool, +} + +#[cfg(test)] +mod tests { + use crate::channels::wasm::schema::ChannelCapabilitiesFile; + + #[test] + fn test_parse_minimal() { + let json = r#"{ + "name": "test" + }"#; + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert_eq!(file.name, "test"); + assert_eq!(file.r#type, "channel"); + } + + #[test] + fn test_parse_full_slack_example() { + let json = r#"{ + "type": "channel", + "name": "slack", + "description": "Slack Events API channel", + "capabilities": { + "http": { + "allowlist": [ + { "host": "slack.com", "path_prefix": "/api/" } + ], + "credentials": { + "slack_bot": { + "secret_name": "slack_bot_token", + "location": { "type": "bearer" }, + "host_patterns": ["slack.com"] + } + }, + "rate_limit": { "requests_per_minute": 50, "requests_per_hour": 1000 } + }, + "secrets": { "allowed_names": ["slack_*"] }, + "channel": { + "allowed_paths": ["/webhook/slack"], + "allow_polling": false, + "emit_rate_limit": { "messages_per_minute": 100, "messages_per_hour": 5000 } + } + }, + "config": { + "signing_secret_name": "slack_signing_secret" + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert_eq!(file.name, "slack"); + assert_eq!( + file.description, + Some("Slack Events API channel".to_string()) + ); + + let caps = file.to_capabilities(); + assert!(caps.is_path_allowed("/webhook/slack")); + assert!(!caps.allow_polling); + assert_eq!(caps.workspace_prefix, "channels/slack/"); + + // Check tool capabilities were parsed + assert!(caps.tool_capabilities.http.is_some()); + assert!(caps.tool_capabilities.secrets.is_some()); + + // Check config + let config_json = file.config_json(); + assert!(config_json.contains("signing_secret_name")); + } + + #[test] + fn test_parse_with_polling() { + let json = r#"{ + "name": "telegram", + "capabilities": { + "channel": { + "allowed_paths": [], + "allow_polling": true, + "min_poll_interval_ms": 60000 + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + let caps = file.to_capabilities(); + + assert!(caps.allow_polling); + assert_eq!(caps.min_poll_interval_ms, 60000); + } + + #[test] + fn test_min_poll_interval_enforced() { + let json = r#"{ + "name": "test", + "capabilities": { + "channel": { + "allow_polling": true, + "min_poll_interval_ms": 1000 + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + let caps = file.to_capabilities(); + + // Should be clamped to minimum + assert_eq!(caps.min_poll_interval_ms, 30000); + } + + #[test] + fn test_workspace_prefix_override() { + let json = r#"{ + "name": "custom", + "capabilities": { + "channel": { + "workspace_prefix": "integrations/custom/" + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + let caps = file.to_capabilities(); + + assert_eq!(caps.workspace_prefix, "integrations/custom/"); + } + + #[test] + fn test_emit_rate_limit() { + let json = r#"{ + "name": "test", + "capabilities": { + "channel": { + "emit_rate_limit": { + "messages_per_minute": 50, + "messages_per_hour": 1000 + } + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + let caps = file.to_capabilities(); + + assert_eq!(caps.emit_rate_limit.messages_per_minute, 50); + assert_eq!(caps.emit_rate_limit.messages_per_hour, 1000); + } +} diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs new file mode 100644 index 00000000..1996b7af --- /dev/null +++ b/src/channels/wasm/wrapper.rs @@ -0,0 +1,1274 @@ +//! WASM channel wrapper implementing the Channel trait. +//! +//! Wraps a prepared WASM channel module and provides the Channel interface. +//! Each callback (on_start, on_http_request, on_poll, on_respond) creates +//! a fresh WASM instance for isolation. +//! +//! # Architecture +//! +//! ```text +//! ┌──────────────────────────────────────────────────────────────┐ +//! │ WasmChannel │ +//! │ │ +//! │ ┌─────────────┐ call_on_* ┌──────────────────────┐ │ +//! │ │ Channel │ ────────────> │ execute_callback │ │ +//! │ │ Trait │ │ (fresh instance) │ │ +//! │ └─────────────┘ └──────────┬───────────┘ │ +//! │ │ │ +//! │ ▼ │ +//! │ ┌──────────────────────────────────────────────────────┐ │ +//! │ │ ChannelStoreData │ │ +//! │ │ ┌─────────────┐ ┌──────────────────────────────┐ │ │ +//! │ │ │ limiter │ │ ChannelHostState │ │ │ +//! │ │ └─────────────┘ │ - emitted_messages │ │ │ +//! │ │ │ - pending_writes │ │ │ +//! │ │ │ - base HostState (logging) │ │ │ +//! │ │ └──────────────────────────────┘ │ │ +//! │ └──────────────────────────────────────────────────────┘ │ +//! └──────────────────────────────────────────────────────────────┘ +//! ``` + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use tokio::sync::{RwLock, mpsc, oneshot}; +use tokio_stream::wrappers::ReceiverStream; +use uuid::Uuid; +use wasmtime::Store; +use wasmtime::component::{Component, Linker, Val}; + +use crate::channels::wasm::capabilities::ChannelCapabilities; +use crate::channels::wasm::error::WasmChannelError; +use crate::channels::wasm::host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage}; +use crate::channels::wasm::router::RegisteredEndpoint; +use crate::channels::wasm::runtime::{PreparedChannelModule, WasmChannelRuntime}; +use crate::channels::wasm::schema::ChannelConfig; +use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; +use crate::error::ChannelError; +use crate::tools::wasm::LogLevel; +use crate::tools::wasm::WasmResourceLimiter; + +/// Store data for WASM channel execution. +/// +/// Contains the resource limiter and channel-specific host state. +struct ChannelStoreData { + limiter: WasmResourceLimiter, + host_state: ChannelHostState, +} + +impl ChannelStoreData { + fn new(memory_limit: u64, channel_name: &str, capabilities: ChannelCapabilities) -> Self { + Self { + limiter: WasmResourceLimiter::new(memory_limit), + host_state: ChannelHostState::new(channel_name, capabilities), + } + } +} + +/// A WASM-based channel implementing the Channel trait. +#[allow(dead_code)] +pub struct WasmChannel { + /// Channel name. + name: String, + + /// Runtime for WASM execution. + runtime: Arc, + + /// Prepared module (compiled WASM). + prepared: Arc, + + /// Channel capabilities. + capabilities: ChannelCapabilities, + + /// Channel configuration JSON (passed to on_start). + config_json: String, + + /// Channel configuration returned by on_start. + channel_config: RwLock>, + + /// Message sender (for emitting messages to the stream). + message_tx: RwLock>>, + + /// Pending responses (for synchronous response handling). + pending_responses: RwLock>>, + + /// Rate limiter for message emission. + rate_limiter: RwLock, + + /// Shutdown signal sender. + shutdown_tx: RwLock>>, + + /// Registered HTTP endpoints. + endpoints: RwLock>, +} + +impl WasmChannel { + /// Create a new WASM channel. + pub fn new( + runtime: Arc, + prepared: Arc, + capabilities: ChannelCapabilities, + config_json: String, + ) -> Self { + let name = prepared.name.clone(); + let rate_limiter = ChannelEmitRateLimiter::new(capabilities.emit_rate_limit.clone()); + + Self { + name, + runtime, + prepared, + capabilities, + config_json, + channel_config: RwLock::new(None), + message_tx: RwLock::new(None), + pending_responses: RwLock::new(HashMap::new()), + rate_limiter: RwLock::new(rate_limiter), + shutdown_tx: RwLock::new(None), + endpoints: RwLock::new(Vec::new()), + } + } + + /// Get the channel name. + pub fn channel_name(&self) -> &str { + &self.name + } + + /// Get the channel capabilities. + pub fn capabilities(&self) -> &ChannelCapabilities { + &self.capabilities + } + + /// Get the registered endpoints. + pub async fn endpoints(&self) -> Vec { + self.endpoints.read().await.clone() + } + + /// Add channel host functions to the linker. + /// + /// These functions are imported by the WASM channel module. + fn add_host_functions(linker: &mut Linker) -> Result<(), WasmChannelError> { + // host.log(level: log-level, message: string) + linker + .root() + .func_wrap( + "log", + |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (level, message): (i32, String)| { + let log_level = match level { + 0 => LogLevel::Trace, + 1 => LogLevel::Debug, + 2 => LogLevel::Info, + 3 => LogLevel::Warn, + 4 => LogLevel::Error, + _ => LogLevel::Info, + }; + // Ignore errors from logging (rate limiting) + let _ = ctx.data_mut().host_state.log(log_level, message); + Ok(()) + }, + ) + .map_err(|e| WasmChannelError::Config(format!("Failed to add log function: {}", e)))?; + + // host.now-millis() -> u64 + linker + .root() + .func_wrap( + "now-millis", + |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (): ()| + -> anyhow::Result<(u64,)> { + Ok((ctx.data().host_state.now_millis(),)) + }, + ) + .map_err(|e| { + WasmChannelError::Config(format!("Failed to add now-millis function: {}", e)) + })?; + + // host.workspace-read(path: string) -> option + linker + .root() + .func_wrap( + "workspace-read", + |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (path,): (String,)| + -> anyhow::Result<(Option,)> { + let result = ctx.data().host_state.workspace_read(&path).ok().flatten(); + Ok((result,)) + }, + ) + .map_err(|e| { + WasmChannelError::Config(format!("Failed to add workspace-read function: {}", e)) + })?; + + // host.workspace-write(path: string, content: string) -> result<_, string> + linker + .root() + .func_wrap( + "workspace-write", + |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (path, content): (String, String)| + -> anyhow::Result<(Result<(), String>,)> { + let result = ctx + .data_mut() + .host_state + .workspace_write(&path, content) + .map_err(|e| e.to_string()); + Ok((result,)) + }, + ) + .map_err(|e| { + WasmChannelError::Config(format!("Failed to add workspace-write function: {}", e)) + })?; + + // host.emit-message(msg: emitted-message) + // The message is passed as a record with fields: user-id, user-name, content, thread-id, metadata-json + linker + .root() + .func_wrap( + "emit-message", + |mut ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (user_id, user_name, content, thread_id, metadata_json): ( + String, + Option, + String, + Option, + String, + )| { + let mut msg = EmittedMessage::new(user_id, content); + if let Some(name) = user_name { + msg = msg.with_user_name(name); + } + if let Some(tid) = thread_id { + msg = msg.with_thread_id(tid); + } + msg = msg.with_metadata(metadata_json); + + // Ignore errors (rate limiting just drops messages) + let _ = ctx.data_mut().host_state.emit_message(msg); + Ok(()) + }, + ) + .map_err(|e| { + WasmChannelError::Config(format!("Failed to add emit-message function: {}", e)) + })?; + + // host.secret-exists(name: string) -> bool + linker + .root() + .func_wrap( + "secret-exists", + |ctx: wasmtime::StoreContextMut<'_, ChannelStoreData>, + (name,): (String,)| + -> anyhow::Result<(bool,)> { + Ok((ctx.data().host_state.secret_exists(&name),)) + }, + ) + .map_err(|e| { + WasmChannelError::Config(format!("Failed to add secret-exists function: {}", e)) + })?; + + Ok(()) + } + + /// Execute a WASM callback synchronously (called from spawn_blocking). + /// + /// This is the core execution logic shared by all callbacks. + fn execute_callback_sync( + runtime: &WasmChannelRuntime, + prepared: &PreparedChannelModule, + capabilities: &ChannelCapabilities, + export_name: &str, + build_args: F, + ) -> Result<(R, ChannelHostState), WasmChannelError> + where + F: FnOnce() -> ( + Vec, + Box Result>, + ), + { + let engine = runtime.engine(); + let limits = &prepared.limits; + + // Create fresh store with channel state (NEAR pattern: fresh instance per call) + let store_data = + ChannelStoreData::new(limits.memory_bytes, &prepared.name, capabilities.clone()); + let mut store = Store::new(engine, store_data); + + // Configure fuel if enabled + if runtime.config().fuel_config.enabled { + store + .set_fuel(limits.fuel) + .map_err(|e| WasmChannelError::Config(format!("Failed to set fuel: {}", e)))?; + } + + // Configure epoch deadline for timeout backup + store.epoch_deadline_trap(); + store.set_epoch_deadline(1); + + // Set up resource limiter + store.limiter(|data| &mut data.limiter); + + // Compile the component (uses cached bytes) + let component = Component::new(engine, prepared.component_bytes()) + .map_err(|e| WasmChannelError::Compilation(e.to_string()))?; + + // Create linker and add host functions + let mut linker = Linker::new(engine); + Self::add_host_functions(&mut linker)?; + + // Instantiate the component + let instance = linker + .instantiate(&mut store, &component) + .map_err(|e| WasmChannelError::Instantiation(e.to_string()))?; + + // Get the export function + let func = instance + .get_func(&mut store, export_name) + .ok_or_else(|| WasmChannelError::MissingExport(export_name.to_string()))?; + + // Build arguments and result extractor + let (args, extract_result) = build_args(); + + // Call the function + let mut results = vec![Val::Bool(false)]; // Placeholder + func.call(&mut store, &args, &mut results).map_err(|e| { + let error_str = e.to_string(); + if error_str.contains("out of fuel") { + WasmChannelError::FuelExhausted { + name: prepared.name.clone(), + limit: limits.fuel, + } + } else if error_str.contains("unreachable") { + WasmChannelError::Trapped { + name: prepared.name.clone(), + reason: "unreachable code executed".to_string(), + } + } else { + WasmChannelError::Trapped { + name: prepared.name.clone(), + reason: error_str, + } + } + })?; + + // Post-call completion (cleanup) + func.post_return(&mut store) + .map_err(|e| WasmChannelError::Trapped { + name: prepared.name.clone(), + reason: format!("post_return failed: {}", e), + })?; + + // Extract result + let result = extract_result(&results[0])?; + + // Get host state with emitted messages and pending writes + let host_state = std::mem::replace( + &mut store.data_mut().host_state, + ChannelHostState::new(&prepared.name, capabilities.clone()), + ); + + Ok((result, host_state)) + } + + /// Execute the on_start callback. + /// + /// Returns the channel configuration for HTTP endpoint registration. + async fn call_on_start(&self) -> Result { + // If no WASM bytes, return default config (for testing) + if self.prepared.component_bytes.is_empty() { + tracing::info!( + channel = %self.name, + "WASM channel on_start called (no WASM module, returning defaults)" + ); + return Ok(ChannelConfig { + display_name: self.prepared.description.clone(), + http_endpoints: Vec::new(), + poll: None, + }); + } + + let runtime = Arc::clone(&self.runtime); + let prepared = Arc::clone(&self.prepared); + let capabilities = self.capabilities.clone(); + let config_json = self.config_json.clone(); + let timeout = self.runtime.config().callback_timeout; + let channel_name_for_error = self.name.clone(); + + // Execute in blocking task with timeout + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + Self::execute_callback_sync(&runtime, &prepared, &capabilities, "on-start", || { + let args = vec![Val::String(config_json)]; + let extract = Box::new(|result: &Val| extract_channel_config(result)); + (args, extract) + }) + }) + .await + .map_err(|e| WasmChannelError::ExecutionPanicked { + name: channel_name_for_error.clone(), + reason: e.to_string(), + })? + }) + .await; + + match result { + Ok(Ok((config, _host_state))) => { + tracing::info!( + channel = %self.name, + display_name = %config.display_name, + endpoints = config.http_endpoints.len(), + "WASM channel on_start completed" + ); + Ok(config) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(WasmChannelError::Timeout { + name: self.name.clone(), + callback: "on_start".to_string(), + }), + } + } + + /// Execute the on_http_request callback. + /// + /// Called when an HTTP request arrives at a registered endpoint. + pub async fn call_on_http_request( + &self, + method: &str, + path: &str, + headers: &HashMap, + query: &HashMap, + body: &[u8], + secret_validated: bool, + ) -> Result { + // If no WASM bytes, return 200 OK (for testing) + if self.prepared.component_bytes.is_empty() { + tracing::debug!( + channel = %self.name, + method = method, + path = path, + "WASM channel on_http_request called (no WASM module)" + ); + return Ok(HttpResponse::ok()); + } + + let runtime = Arc::clone(&self.runtime); + let prepared = Arc::clone(&self.prepared); + let capabilities = self.capabilities.clone(); + let timeout = self.runtime.config().callback_timeout; + + // Prepare request data + let method = method.to_string(); + let path = path.to_string(); + let headers_json = serde_json::to_string(&headers).unwrap_or_default(); + let query_json = serde_json::to_string(&query).unwrap_or_default(); + let body = body.to_vec(); + + // Clone name for error handling before moving into closure + let channel_name_for_error = self.name.clone(); + + // Execute in blocking task with timeout + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + Self::execute_callback_sync( + &runtime, + &prepared, + &capabilities, + "on-http-request", + || { + // Build incoming-http-request record + let request = Val::Record(vec![ + ("method".to_string(), Val::String(method)), + ("path".to_string(), Val::String(path)), + ("headers-json".to_string(), Val::String(headers_json)), + ("query-json".to_string(), Val::String(query_json)), + ( + "body".to_string(), + Val::List(body.into_iter().map(Val::U8).collect()), + ), + ("secret-validated".to_string(), Val::Bool(secret_validated)), + ]); + let args = vec![request]; + let extract = Box::new(|result: &Val| extract_http_response(result)); + (args, extract) + }, + ) + }) + .await + .map_err(|e| WasmChannelError::ExecutionPanicked { + name: channel_name_for_error.clone(), + reason: e.to_string(), + })? + }) + .await; + + let channel_name = self.name.clone(); + match result { + Ok(Ok((response, mut host_state))) => { + // Process emitted messages + let emitted = host_state.take_emitted_messages(); + self.process_emitted_messages(emitted).await?; + + tracing::debug!( + channel = %channel_name, + status = response.status, + "WASM channel on_http_request completed" + ); + Ok(response) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(WasmChannelError::Timeout { + name: channel_name, + callback: "on_http_request".to_string(), + }), + } + } + + /// Execute the on_poll callback. + /// + /// Called periodically if polling is configured. + pub async fn call_on_poll(&self) -> Result<(), WasmChannelError> { + // If no WASM bytes, do nothing (for testing) + if self.prepared.component_bytes.is_empty() { + tracing::debug!( + channel = %self.name, + "WASM channel on_poll called (no WASM module)" + ); + return Ok(()); + } + + let runtime = Arc::clone(&self.runtime); + let prepared = Arc::clone(&self.prepared); + let capabilities = self.capabilities.clone(); + let timeout = self.runtime.config().callback_timeout; + let channel_name = self.name.clone(); + + // Execute in blocking task with timeout + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + Self::execute_callback_sync(&runtime, &prepared, &capabilities, "on-poll", || { + let args = vec![]; + let extract = Box::new(|_result: &Val| Ok(())); + (args, extract) + }) + }) + .await + .map_err(|e| WasmChannelError::ExecutionPanicked { + name: channel_name.clone(), + reason: e.to_string(), + })? + }) + .await; + + let channel_name = self.name.clone(); + match result { + Ok(Ok(((), mut host_state))) => { + // Process emitted messages + let emitted = host_state.take_emitted_messages(); + self.process_emitted_messages(emitted).await?; + + tracing::debug!( + channel = %channel_name, + "WASM channel on_poll completed" + ); + Ok(()) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(WasmChannelError::Timeout { + name: channel_name, + callback: "on_poll".to_string(), + }), + } + } + + /// Execute the on_respond callback. + /// + /// Called when the agent has a response to send back. + pub async fn call_on_respond( + &self, + message_id: Uuid, + content: &str, + thread_id: Option<&str>, + metadata_json: &str, + ) -> Result<(), WasmChannelError> { + // If no WASM bytes, do nothing (for testing) + if self.prepared.component_bytes.is_empty() { + tracing::debug!( + channel = %self.name, + message_id = %message_id, + "WASM channel on_respond called (no WASM module)" + ); + return Ok(()); + } + + let runtime = Arc::clone(&self.runtime); + let prepared = Arc::clone(&self.prepared); + let capabilities = self.capabilities.clone(); + let timeout = self.runtime.config().callback_timeout; + let channel_name = self.name.clone(); + + // Prepare response data + let message_id_str = message_id.to_string(); + let content = content.to_string(); + let thread_id = thread_id.map(|s| s.to_string()); + let metadata_json = metadata_json.to_string(); + + // Execute in blocking task with timeout + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + Self::execute_callback_sync( + &runtime, + &prepared, + &capabilities, + "on-respond", + || { + // Build agent-response record + let response = Val::Record(vec![ + ("message-id".to_string(), Val::String(message_id_str)), + ("content".to_string(), Val::String(content)), + ( + "thread-id".to_string(), + match thread_id { + Some(tid) => Val::Option(Some(Box::new(Val::String(tid)))), + None => Val::Option(None), + }, + ), + ("metadata-json".to_string(), Val::String(metadata_json)), + ]); + let args = vec![response]; + let extract = Box::new(|result: &Val| extract_result_unit(result)); + (args, extract) + }, + ) + }) + .await + .map_err(|e| WasmChannelError::ExecutionPanicked { + name: channel_name.clone(), + reason: e.to_string(), + })? + }) + .await; + + let channel_name = self.name.clone(); + match result { + Ok(Ok(((), _host_state))) => { + tracing::debug!( + channel = %channel_name, + message_id = %message_id, + "WASM channel on_respond completed" + ); + Ok(()) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(WasmChannelError::Timeout { + name: channel_name, + callback: "on_respond".to_string(), + }), + } + } + + /// Process emitted messages from a callback. + async fn process_emitted_messages( + &self, + messages: Vec, + ) -> Result<(), WasmChannelError> { + if messages.is_empty() { + return Ok(()); + } + + let tx_guard = self.message_tx.read().await; + let Some(tx) = tx_guard.as_ref() else { + tracing::warn!( + channel = %self.name, + count = messages.len(), + "Messages emitted but no sender available" + ); + return Ok(()); + }; + + let mut rate_limiter = self.rate_limiter.write().await; + + for emitted in messages { + // Check rate limit + if !rate_limiter.check_and_record() { + tracing::warn!( + channel = %self.name, + "Message emission rate limited" + ); + return Err(WasmChannelError::EmitRateLimited { + name: self.name.clone(), + }); + } + + // Convert to IncomingMessage + let mut msg = IncomingMessage::new(&self.name, &emitted.user_id, &emitted.content); + + if let Some(name) = emitted.user_name { + msg = msg.with_user_name(name); + } + + if let Some(thread_id) = emitted.thread_id { + msg = msg.with_thread(thread_id); + } + + // Parse metadata JSON + if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) { + msg = msg.with_metadata(metadata); + } + + // Send to stream + if tx.send(msg).await.is_err() { + tracing::warn!( + channel = %self.name, + "Failed to send emitted message, channel closed" + ); + break; + } + } + + Ok(()) + } + + /// Start the polling loop if configured. + fn start_polling(&self, interval: Duration, shutdown_rx: oneshot::Receiver<()>) { + let channel_name = self.name.clone(); + + // Clone self reference for the async block + // In a real implementation, we'd hold an Arc + tokio::spawn(async move { + let mut interval_timer = tokio::time::interval(interval); + let mut shutdown = std::pin::pin!(shutdown_rx); + + loop { + tokio::select! { + _ = interval_timer.tick() => { + tracing::debug!( + channel = %channel_name, + "Polling tick (stub - would call on_poll)" + ); + // In real implementation: self.call_on_poll().await + } + _ = &mut shutdown => { + tracing::info!( + channel = %channel_name, + "Polling stopped" + ); + break; + } + } + } + }); + } +} + +#[async_trait] +impl Channel for WasmChannel { + fn name(&self) -> &str { + &self.name + } + + async fn start(&self) -> Result { + // Create message channel + let (tx, rx) = mpsc::channel(256); + *self.message_tx.write().await = Some(tx); + + // Create shutdown channel + let (shutdown_tx, _shutdown_rx) = oneshot::channel(); + *self.shutdown_tx.write().await = Some(shutdown_tx); + + // Call on_start to get configuration + let config = self + .call_on_start() + .await + .map_err(|e| ChannelError::StartupFailed { + name: self.name.clone(), + reason: e.to_string(), + })?; + + // Store the config + *self.channel_config.write().await = Some(config.clone()); + + // Register HTTP endpoints + let mut endpoints = Vec::new(); + for endpoint in &config.http_endpoints { + // Validate path is allowed + if !self.capabilities.is_path_allowed(&endpoint.path) { + tracing::warn!( + channel = %self.name, + path = %endpoint.path, + "HTTP endpoint path not allowed by capabilities" + ); + continue; + } + + endpoints.push(RegisteredEndpoint { + channel_name: self.name.clone(), + path: endpoint.path.clone(), + methods: endpoint.methods.clone(), + require_secret: endpoint.require_secret, + }); + } + *self.endpoints.write().await = endpoints; + + // Start polling if configured + if let Some(poll_config) = &config.poll { + if poll_config.enabled { + let interval = self + .capabilities + .validate_poll_interval(poll_config.interval_ms) + .map_err(|e| ChannelError::StartupFailed { + name: self.name.clone(), + reason: e, + })?; + + // Create a new shutdown receiver for polling + let (_poll_shutdown_tx, poll_shutdown_rx) = oneshot::channel(); + + self.start_polling(Duration::from_millis(interval as u64), poll_shutdown_rx); + } + } + + tracing::info!( + channel = %self.name, + display_name = %config.display_name, + endpoints = config.http_endpoints.len(), + "WASM channel started" + ); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + + async fn respond( + &self, + msg: &IncomingMessage, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + // Check if there's a pending synchronous response waiter + if let Some(tx) = self.pending_responses.write().await.remove(&msg.id) { + let _ = tx.send(response.content.clone()); + } + + // Call WASM on_respond + let metadata_json = serde_json::to_string(&response.metadata).unwrap_or_default(); + self.call_on_respond( + msg.id, + &response.content, + response.thread_id.as_deref(), + &metadata_json, + ) + .await + .map_err(|e| ChannelError::SendFailed { + name: self.name.clone(), + reason: e.to_string(), + })?; + + Ok(()) + } + + async fn send_status(&self, status: StatusUpdate) -> Result<(), ChannelError> { + // WASM channels don't support status updates by default + // Could be extended with an optional on_status callback + let _ = status; + Ok(()) + } + + async fn health_check(&self) -> Result<(), ChannelError> { + // Check if we have an active message sender + if self.message_tx.read().await.is_some() { + Ok(()) + } else { + Err(ChannelError::HealthCheckFailed { + name: self.name.clone(), + }) + } + } + + async fn shutdown(&self) -> Result<(), ChannelError> { + // Send shutdown signal + if let Some(tx) = self.shutdown_tx.write().await.take() { + let _ = tx.send(()); + } + + // Clear the message sender + *self.message_tx.write().await = None; + + // TODO: Call WASM on_shutdown if we add that callback + + tracing::info!( + channel = %self.name, + "WASM channel shut down" + ); + + Ok(()) + } +} + +impl std::fmt::Debug for WasmChannel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WasmChannel") + .field("name", &self.name) + .field("prepared", &self.prepared.name) + .finish() + } +} + +// ============================================================================ +// Value Extraction Helpers +// ============================================================================ + +/// Extract ChannelConfig from a WIT result. +fn extract_channel_config(val: &Val) -> Result { + // Result is (ok: option, err: option) + match val { + Val::Result(result) => match result.as_ref() { + Ok(Some(config_val)) => extract_channel_config_inner(config_val), + Ok(None) => Err(WasmChannelError::InvalidResponse( + "on-start returned empty Ok".to_string(), + )), + Err(Some(err_val)) => { + if let Val::String(err) = err_val.as_ref() { + Err(WasmChannelError::CallbackFailed { + name: "channel".to_string(), + reason: err.clone(), + }) + } else { + Err(WasmChannelError::InvalidResponse( + "on-start error is not a string".to_string(), + )) + } + } + Err(None) => Err(WasmChannelError::InvalidResponse( + "on-start returned empty Err".to_string(), + )), + }, + // Fallback: try to parse as record directly (for simpler implementations) + Val::Record(_) => extract_channel_config_inner(val), + _ => Err(WasmChannelError::InvalidResponse(format!( + "Expected result or record, got {:?}", + std::mem::discriminant(val) + ))), + } +} + +/// Extract ChannelConfig from a channel-config record. +fn extract_channel_config_inner(val: &Val) -> Result { + match val { + Val::Record(fields) => { + let mut display_name = String::new(); + let mut http_endpoints = Vec::new(); + let mut poll = None; + + for (name, field_val) in fields { + match name.as_str() { + "display-name" => { + if let Val::String(s) = field_val { + display_name = s.clone(); + } + } + "http-endpoints" => { + if let Val::List(endpoints) = field_val { + for ep in endpoints { + if let Ok(endpoint) = extract_http_endpoint_config(ep) { + http_endpoints.push(endpoint); + } + } + } + } + "poll" => { + if let Val::Option(Some(poll_val)) = field_val { + poll = extract_poll_config(poll_val).ok(); + } + } + _ => {} + } + } + + Ok(ChannelConfig { + display_name, + http_endpoints, + poll, + }) + } + _ => Err(WasmChannelError::InvalidResponse( + "Expected record for channel-config".to_string(), + )), + } +} + +/// Extract HttpEndpointConfigSchema from a record. +fn extract_http_endpoint_config( + val: &Val, +) -> Result { + match val { + Val::Record(fields) => { + let mut path = String::new(); + let mut methods = Vec::new(); + let mut require_secret = false; + + for (name, field_val) in fields { + match name.as_str() { + "path" => { + if let Val::String(s) = field_val { + path = s.clone(); + } + } + "methods" => { + if let Val::List(list) = field_val { + for item in list { + if let Val::String(s) = item { + methods.push(s.clone()); + } + } + } + } + "require-secret" => { + if let Val::Bool(b) = field_val { + require_secret = *b; + } + } + _ => {} + } + } + + Ok(crate::channels::wasm::schema::HttpEndpointConfigSchema { + path, + methods, + require_secret, + }) + } + _ => Err(WasmChannelError::InvalidResponse( + "Expected record for http-endpoint-config".to_string(), + )), + } +} + +/// Extract PollConfigSchema from a record. +fn extract_poll_config( + val: &Val, +) -> Result { + match val { + Val::Record(fields) => { + let mut interval_ms = 30_000; + let mut enabled = false; + + for (name, field_val) in fields { + match name.as_str() { + "interval-ms" => { + if let Val::U32(n) = field_val { + interval_ms = *n; + } + } + "enabled" => { + if let Val::Bool(b) = field_val { + enabled = *b; + } + } + _ => {} + } + } + + Ok(crate::channels::wasm::schema::PollConfigSchema { + interval_ms, + enabled, + }) + } + _ => Err(WasmChannelError::InvalidResponse( + "Expected record for poll-config".to_string(), + )), + } +} + +/// Extract HttpResponse from a WIT outgoing-http-response record. +fn extract_http_response(val: &Val) -> Result { + match val { + Val::Record(fields) => { + let mut status = 200u16; + let mut headers = HashMap::new(); + let mut body = Vec::new(); + + for (name, field_val) in fields { + match name.as_str() { + "status" => { + if let Val::U16(s) = field_val { + status = *s; + } + } + "headers-json" => { + if let Val::String(s) = field_val { + if let Ok(h) = serde_json::from_str::>(s) { + headers = h; + } + } + } + "body" => { + if let Val::List(bytes) = field_val { + body = bytes + .iter() + .filter_map(|v| if let Val::U8(b) = v { Some(*b) } else { None }) + .collect(); + } + } + _ => {} + } + } + + Ok(HttpResponse { + status, + headers, + body, + }) + } + _ => Err(WasmChannelError::InvalidResponse( + "Expected record for http-response".to_string(), + )), + } +} + +/// Extract unit result from a WIT result<_, string>. +fn extract_result_unit(val: &Val) -> Result<(), WasmChannelError> { + match val { + Val::Result(result) => match result.as_ref() { + Ok(_) => Ok(()), + Err(Some(err_val)) => { + if let Val::String(err) = err_val.as_ref() { + Err(WasmChannelError::CallbackFailed { + name: "channel".to_string(), + reason: err.clone(), + }) + } else { + Err(WasmChannelError::InvalidResponse( + "Error is not a string".to_string(), + )) + } + } + Err(None) => Err(WasmChannelError::InvalidResponse( + "Returned empty Err".to_string(), + )), + }, + // Unit return (for on-poll which returns nothing) + Val::Tuple(items) if items.is_empty() => Ok(()), + _ => Ok(()), // Treat anything else as success for unit-returning callbacks + } +} + +/// HTTP response from a WASM channel callback. +#[derive(Debug, Clone)] +pub struct HttpResponse { + /// HTTP status code. + pub status: u16, + /// Response headers. + pub headers: HashMap, + /// Response body. + pub body: Vec, +} + +impl HttpResponse { + /// Create an OK response. + pub fn ok() -> Self { + Self { + status: 200, + headers: HashMap::new(), + body: Vec::new(), + } + } + + /// Create a JSON response. + pub fn json(value: serde_json::Value) -> Self { + let body = serde_json::to_vec(&value).unwrap_or_default(); + let mut headers = HashMap::new(); + headers.insert("Content-Type".to_string(), "application/json".to_string()); + Self { + status: 200, + headers, + body, + } + } + + /// Create an error response. + pub fn error(status: u16, message: &str) -> Self { + Self { + status, + headers: HashMap::new(), + body: message.as_bytes().to_vec(), + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use crate::channels::Channel; + use crate::channels::wasm::capabilities::ChannelCapabilities; + use crate::channels::wasm::runtime::{ + PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig, + }; + use crate::channels::wasm::wrapper::{HttpResponse, WasmChannel}; + use crate::tools::wasm::ResourceLimits; + + fn create_test_channel() -> WasmChannel { + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); + + let prepared = Arc::new(PreparedChannelModule { + name: "test".to_string(), + description: "Test channel".to_string(), + component_bytes: Vec::new(), + limits: ResourceLimits::default(), + }); + + let capabilities = ChannelCapabilities::for_channel("test").with_path("/webhook/test"); + + WasmChannel::new(runtime, prepared, capabilities, "{}".to_string()) + } + + #[test] + fn test_channel_name() { + let channel = create_test_channel(); + assert_eq!(channel.name(), "test"); + } + + #[test] + fn test_http_response_ok() { + let response = HttpResponse::ok(); + assert_eq!(response.status, 200); + assert!(response.body.is_empty()); + } + + #[test] + fn test_http_response_json() { + let response = HttpResponse::json(serde_json::json!({"key": "value"})); + assert_eq!(response.status, 200); + assert_eq!( + response.headers.get("Content-Type"), + Some(&"application/json".to_string()) + ); + } + + #[test] + fn test_http_response_error() { + let response = HttpResponse::error(400, "Bad request"); + assert_eq!(response.status, 400); + assert_eq!(response.body, b"Bad request"); + } + + #[tokio::test] + async fn test_channel_start_and_shutdown() { + let channel = create_test_channel(); + + // Start should succeed + let stream = channel.start().await; + assert!(stream.is_ok()); + + // Health check should pass + assert!(channel.health_check().await.is_ok()); + + // Shutdown should succeed + assert!(channel.shutdown().await.is_ok()); + + // Health check should fail after shutdown + assert!(channel.health_check().await.is_err()); + } +} diff --git a/src/config.rs b/src/config.rs index 1f431e21..bfd082a8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -180,6 +180,10 @@ fn default_session_path() -> PathBuf { pub struct ChannelsConfig { pub cli: CliConfig, pub http: Option, + /// Directory containing WASM channel modules (default: ~/.near-agent/channels/). + pub wasm_channels_dir: std::path::PathBuf, + /// Whether WASM channels are enabled. + pub wasm_channels_enabled: bool, } #[derive(Debug, Clone)] @@ -222,10 +226,29 @@ impl ChannelsConfig { enabled: cli_enabled, }, http, + wasm_channels_dir: optional_env("WASM_CHANNELS_DIR")? + .map(PathBuf::from) + .unwrap_or_else(default_channels_dir), + wasm_channels_enabled: optional_env("WASM_CHANNELS_ENABLED")? + .map(|s| s.parse()) + .transpose() + .map_err(|e| ConfigError::InvalidValue { + key: "WASM_CHANNELS_ENABLED".to_string(), + message: format!("must be 'true' or 'false': {e}"), + })? + .unwrap_or(true), }) } } +/// Get the default channels directory (~/.near-agent/channels/). +fn default_channels_dir() -> PathBuf { + dirs::home_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join(".near-agent") + .join("channels") +} + /// Agent behavior configuration. #[derive(Debug, Clone)] pub struct AgentConfig { diff --git a/src/main.rs b/src/main.rs index 1741d993..075d16a3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,10 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitEx use near_agent::{ agent::{Agent, AgentDeps}, - channels::{AppEvent, ChannelManager, HttpChannel, ReplChannel, TuiChannel}, + channels::{ + AppEvent, ChannelManager, HttpChannel, ReplChannel, TuiChannel, + wasm::{WasmChannelLoader, WasmChannelRuntime, WasmChannelRuntimeConfig}, + }, cli::{Cli, Command, run_tool_command}, config::Config, history::Store, @@ -286,6 +289,41 @@ async fn main() -> anyhow::Result<()> { } } + // Load WASM channels if enabled + if config.channels.wasm_channels_enabled && config.channels.wasm_channels_dir.exists() { + match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { + Ok(runtime) => { + let runtime = Arc::new(runtime); + let loader = WasmChannelLoader::new(Arc::clone(&runtime)); + + match loader + .load_from_dir(&config.channels.wasm_channels_dir) + .await + { + Ok(results) => { + for channel in results.loaded { + tracing::info!("Loaded WASM channel: {}", channel.channel_name()); + channels.add(Box::new(channel)); + } + for (path, err) in &results.errors { + tracing::warn!( + "Failed to load WASM channel {}: {}", + path.display(), + err + ); + } + } + Err(e) => { + tracing::warn!("Failed to scan WASM channels directory: {}", e); + } + } + } + Err(e) => { + tracing::warn!("Failed to initialize WASM channel runtime: {}", e); + } + } + } + // Create workspace for agent (shared with memory tools) let workspace = store.as_ref().map(|s| { let mut ws = Workspace::new("default", s.pool()); diff --git a/src/tools/wasm/mod.rs b/src/tools/wasm/mod.rs index 3dcb6fcd..c496acbe 100644 --- a/src/tools/wasm/mod.rs +++ b/src/tools/wasm/mod.rs @@ -118,4 +118,4 @@ pub use storage::{ pub use loader::{DiscoveredTool, LoadResults, WasmLoadError, WasmToolLoader, discover_tools}; // Capabilities schema (for parsing *.capabilities.json files) -pub use capabilities_schema::CapabilitiesFile; +pub use capabilities_schema::{CapabilitiesFile, RateLimitSchema}; diff --git a/tests/wasm_channel_integration.rs b/tests/wasm_channel_integration.rs new file mode 100644 index 00000000..3b2241bf --- /dev/null +++ b/tests/wasm_channel_integration.rs @@ -0,0 +1,446 @@ +//! Integration tests for the WASM channel system. +//! +//! These tests verify the full flow of WASM channel operations: +//! - Channel loading from filesystem +//! - HTTP webhook routing +//! - Message emission and delivery +//! - Response handling + +use std::collections::HashMap; +use std::sync::Arc; + +use near_agent::channels::Channel; +use near_agent::channels::wasm::{ + ChannelCapabilities, EmitRateLimitConfig, PreparedChannelModule, RegisteredEndpoint, + WasmChannel, WasmChannelRouter, WasmChannelRuntime, WasmChannelRuntimeConfig, +}; +use tempfile::TempDir; + +/// Create a test runtime for WASM channel operations. +fn create_test_runtime() -> Arc { + let config = WasmChannelRuntimeConfig::for_testing(); + Arc::new(WasmChannelRuntime::new(config).expect("Failed to create runtime")) +} + +/// Create a test channel with minimal configuration. +fn create_test_channel( + runtime: Arc, + name: &str, + paths: Vec<&str>, +) -> WasmChannel { + let prepared = Arc::new(PreparedChannelModule::for_testing( + name, + format!("Test channel: {}", name), + )); + + let mut capabilities = ChannelCapabilities::for_channel(name); + for path in paths { + capabilities = capabilities.with_path(path.to_string()); + } + + WasmChannel::new(runtime, prepared, capabilities, "{}".to_string()) +} + +mod router_tests { + use super::*; + + #[tokio::test] + async fn test_register_and_route_channel() { + let router = WasmChannelRouter::new(); + let runtime = create_test_runtime(); + + let channel = Arc::new(create_test_channel( + runtime, + "test-channel", + vec!["/webhook/test"], + )); + + let endpoints = vec![RegisteredEndpoint { + channel_name: "test-channel".to_string(), + path: "/webhook/test".to_string(), + methods: vec!["POST".to_string()], + require_secret: false, + }]; + + router.register(channel.clone(), endpoints, None).await; + + // Verify channel is found by path + let found = router.get_channel_for_path("/webhook/test").await; + assert!(found.is_some()); + assert_eq!(found.unwrap().channel_name(), "test-channel"); + + // Verify non-existent path returns None + let not_found = router.get_channel_for_path("/webhook/nonexistent").await; + assert!(not_found.is_none()); + } + + #[tokio::test] + async fn test_secret_validation() { + let router = WasmChannelRouter::new(); + let runtime = create_test_runtime(); + + let channel = Arc::new(create_test_channel( + runtime, + "secure-channel", + vec!["/webhook/secure"], + )); + + router + .register(channel, vec![], Some("my-secret-123".to_string())) + .await; + + // Correct secret validates + assert!( + router + .validate_secret("secure-channel", "my-secret-123") + .await + ); + + // Wrong secret fails + assert!( + !router + .validate_secret("secure-channel", "wrong-secret") + .await + ); + + // Non-existent channel without secret always validates + assert!(router.validate_secret("nonexistent", "anything").await); + } + + #[tokio::test] + async fn test_unregister_channel() { + let router = WasmChannelRouter::new(); + let runtime = create_test_runtime(); + + let channel = Arc::new(create_test_channel( + runtime, + "temp-channel", + vec!["/webhook/temp"], + )); + + let endpoints = vec![RegisteredEndpoint { + channel_name: "temp-channel".to_string(), + path: "/webhook/temp".to_string(), + methods: vec!["POST".to_string()], + require_secret: false, + }]; + + router.register(channel, endpoints, None).await; + + // Channel exists + assert!(router.get_channel_for_path("/webhook/temp").await.is_some()); + + // Unregister + router.unregister("temp-channel").await; + + // Channel no longer exists + assert!(router.get_channel_for_path("/webhook/temp").await.is_none()); + } + + #[tokio::test] + async fn test_multiple_channels() { + let router = WasmChannelRouter::new(); + let runtime = create_test_runtime(); + + // Register multiple channels + for name in &["slack", "telegram", "discord"] { + let channel = Arc::new(create_test_channel( + Arc::clone(&runtime), + name, + vec![&format!("/webhook/{}", name)], + )); + + let endpoints = vec![RegisteredEndpoint { + channel_name: name.to_string(), + path: format!("/webhook/{}", name), + methods: vec!["POST".to_string()], + require_secret: false, + }]; + + router.register(channel, endpoints, None).await; + } + + // Verify all channels are registered + let channels = router.list_channels().await; + assert_eq!(channels.len(), 3); + assert!(channels.contains(&"slack".to_string())); + assert!(channels.contains(&"telegram".to_string())); + assert!(channels.contains(&"discord".to_string())); + + // Verify all paths work + for name in &["slack", "telegram", "discord"] { + let found = router + .get_channel_for_path(&format!("/webhook/{}", name)) + .await; + assert!(found.is_some()); + assert_eq!(found.unwrap().channel_name(), *name); + } + } +} + +mod channel_lifecycle_tests { + use super::*; + + #[tokio::test] + async fn test_channel_start_and_shutdown() { + let runtime = create_test_runtime(); + let channel = create_test_channel(runtime, "lifecycle-test", vec!["/webhook/lifecycle"]); + + // Start channel + let stream = channel.start().await; + assert!(stream.is_ok()); + + // Health check should pass + assert!(channel.health_check().await.is_ok()); + + // Shutdown + assert!(channel.shutdown().await.is_ok()); + + // Health check should fail after shutdown + assert!(channel.health_check().await.is_err()); + } + + #[tokio::test] + async fn test_channel_http_callback() { + let runtime = create_test_runtime(); + let channel = create_test_channel(runtime, "http-test", vec!["/webhook/http"]); + + // Start channel + let _stream = channel.start().await.expect("Failed to start channel"); + + // Call HTTP callback (stub implementation returns 200 OK) + let response = channel + .call_on_http_request( + "POST", + "/webhook/http", + &HashMap::new(), + &HashMap::new(), + b"{}", + true, + ) + .await + .expect("HTTP callback failed"); + + assert_eq!(response.status, 200); + + // Cleanup + channel.shutdown().await.expect("Shutdown failed"); + } +} + +mod loader_tests { + use super::*; + use std::io::Write; + + #[tokio::test] + async fn test_discover_channels_empty_dir() { + let dir = TempDir::new().expect("Failed to create temp dir"); + + let channels = near_agent::channels::wasm::discover_channels(dir.path()) + .await + .expect("Discovery failed"); + + assert!(channels.is_empty()); + } + + #[tokio::test] + async fn test_discover_channels_with_wasm_files() { + let dir = TempDir::new().expect("Failed to create temp dir"); + + // Create fake WASM files + std::fs::File::create(dir.path().join("slack.wasm")).expect("Failed to create file"); + std::fs::File::create(dir.path().join("telegram.wasm")).expect("Failed to create file"); + + let channels = near_agent::channels::wasm::discover_channels(dir.path()) + .await + .expect("Discovery failed"); + + assert_eq!(channels.len(), 2); + assert!(channels.contains_key("slack")); + assert!(channels.contains_key("telegram")); + } + + #[tokio::test] + async fn test_discover_channels_with_capabilities() { + let dir = TempDir::new().expect("Failed to create temp dir"); + + // Create WASM and capabilities file + std::fs::File::create(dir.path().join("custom.wasm")).expect("Failed to create wasm"); + + let mut cap_file = std::fs::File::create(dir.path().join("custom.capabilities.json")) + .expect("Failed to create capabilities"); + cap_file + .write_all( + br#"{ + "name": "custom", + "capabilities": { + "channel": { + "allowed_paths": ["/webhook/custom"] + } + } + }"#, + ) + .expect("Failed to write capabilities"); + + let channels = near_agent::channels::wasm::discover_channels(dir.path()) + .await + .expect("Discovery failed"); + + assert_eq!(channels.len(), 1); + assert!(channels["custom"].capabilities_path.is_some()); + } + + #[tokio::test] + async fn test_discover_channels_ignores_non_wasm() { + let dir = TempDir::new().expect("Failed to create temp dir"); + + // Create various non-WASM files + std::fs::File::create(dir.path().join("readme.md")).expect("Failed to create file"); + std::fs::File::create(dir.path().join("config.json")).expect("Failed to create file"); + std::fs::File::create(dir.path().join("channel.wasm")).expect("Failed to create file"); + + let channels = near_agent::channels::wasm::discover_channels(dir.path()) + .await + .expect("Discovery failed"); + + // Only the .wasm file should be discovered + assert_eq!(channels.len(), 1); + assert!(channels.contains_key("channel")); + } +} + +mod capabilities_tests { + use super::*; + + #[test] + fn test_capabilities_path_validation() { + let caps = ChannelCapabilities::for_channel("test") + .with_path("/webhook/test") + .with_path("/api/events"); + + assert!(caps.is_path_allowed("/webhook/test")); + assert!(caps.is_path_allowed("/api/events")); + assert!(!caps.is_path_allowed("/other/path")); + } + + #[test] + fn test_capabilities_workspace_prefix() { + let caps = ChannelCapabilities::for_channel("slack"); + + assert_eq!(caps.workspace_prefix, "channels/slack/"); + + // Validate path prefixing + let prefixed = caps.prefix_workspace_path("state.json"); + assert_eq!(prefixed, "channels/slack/state.json"); + } + + #[test] + fn test_capabilities_workspace_path_validation() { + let caps = ChannelCapabilities::for_channel("test"); + + // Valid paths + assert!(caps.validate_workspace_path("state.json").is_ok()); + assert!(caps.validate_workspace_path("data/file.txt").is_ok()); + + // Invalid paths (traversal attempts) + assert!(caps.validate_workspace_path("../escape.txt").is_err()); + assert!(caps.validate_workspace_path("/absolute/path").is_err()); + assert!(caps.validate_workspace_path("data/../escape").is_err()); + } + + #[test] + fn test_capabilities_poll_interval_validation() { + let caps = ChannelCapabilities::for_channel("test").with_polling(30_000); + + // Valid interval (returns as-is) + let result = caps.validate_poll_interval(60_000); + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 60_000); + + // Too short interval is clamped to minimum (not rejected) + let result = caps.validate_poll_interval(1_000); + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 30_000); + + // Minimum interval passes as-is + let result = caps.validate_poll_interval(30_000); + assert!(result.is_ok()); + assert_eq!(result.unwrap(), 30_000); + + // Polling disabled returns error + let no_poll_caps = ChannelCapabilities::for_channel("no-poll"); + assert!(no_poll_caps.validate_poll_interval(60_000).is_err()); + } + + #[test] + fn test_emit_rate_limit_config() { + let config = EmitRateLimitConfig { + messages_per_minute: 100, + messages_per_hour: 5000, + }; + + assert_eq!(config.messages_per_minute, 100); + assert_eq!(config.messages_per_hour, 5000); + } +} + +mod message_emission_tests { + use super::*; + use near_agent::channels::wasm::{ChannelHostState, EmittedMessage}; + + #[test] + fn test_emit_message_basic() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + let msg = EmittedMessage::new("user123", "Hello, world!"); + state.emit_message(msg).expect("Emit should succeed"); + + assert_eq!(state.emitted_count(), 1); + + let messages = state.take_emitted_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].user_id, "user123"); + assert_eq!(messages[0].content, "Hello, world!"); + + // Queue should be cleared + assert_eq!(state.emitted_count(), 0); + } + + #[test] + fn test_emit_message_with_metadata() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + let msg = EmittedMessage::new("user123", "Hello") + .with_user_name("John Doe") + .with_thread_id("thread-1") + .with_metadata(r#"{"channel": "C123"}"#); + + state.emit_message(msg).expect("Emit should succeed"); + + let messages = state.take_emitted_messages(); + assert_eq!(messages[0].user_name, Some("John Doe".to_string())); + assert_eq!(messages[0].thread_id, Some("thread-1".to_string())); + assert!(messages[0].metadata_json.contains("channel")); + } + + #[test] + fn test_emit_rate_limiting() { + let caps = ChannelCapabilities::for_channel("test"); + let mut state = ChannelHostState::new("test", caps); + + // Emit up to the per-execution limit + for i in 0..100 { + let msg = EmittedMessage::new("user", format!("Message {}", i)); + state.emit_message(msg).expect("Emit should succeed"); + } + + // Messages beyond the limit are silently dropped + let msg = EmittedMessage::new("user", "Should be dropped"); + state.emit_message(msg).expect("Emit should not fail"); + + assert_eq!(state.emitted_count(), 100); + assert_eq!(state.emits_dropped(), 1); + } +} diff --git a/wit/channel.wit b/wit/channel.wit new file mode 100644 index 00000000..f2514c5a --- /dev/null +++ b/wit/channel.wit @@ -0,0 +1,276 @@ +// WASM Channel Sandbox Interface +// +// Defines the contract between sandboxed channels and the host runtime. +// Channels export the `channel` interface; the host provides the `channel-host` interface. +// +// Architecture: Host-Managed Event Loop +// ┌─────────────────────────────────────────────────────────────────────────────────┐ +// │ Host-Managed Event Loop │ +// │ │ +// │ ┌─────────────┐ ┌──────────────┐ ┌──────────────┐ │ +// │ │ HTTP │ │ Polling │ │ Timer │ │ +// │ │ Router │ │ Scheduler │ │ Scheduler │ │ +// │ └──────┬──────┘ └──────┬───────┘ └──────┬───────┘ │ +// │ └───────────────────┴────────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌──────────────────┬──────────────────┐ │ +// │ ▼ ▼ ▼ │ +// │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ +// │ │ on-http-req │ │ on-poll │ │ on-respond │ WASM Exports │ +// │ └─────────────┘ └─────────────┘ └─────────────┘ │ +// │ │ │ │ │ +// │ └──────────────────┴──────────────────┘ │ +// │ │ │ +// │ ▼ │ +// │ ┌─────────────────┐ │ +// │ │ Host Imports │ │ +// │ │ emit-message │──────────▶ MessageStream │ +// │ │ http-request │ │ +// │ └─────────────────┘ │ +// └─────────────────────────────────────────────────────────────────────────────────┘ +// +// Security Model: +// - WASM channels are untrusted and run in a sandbox +// - Fresh instance per callback (no shared mutable state) +// - All capabilities are opt-in (default: no access) +// - Secrets are NEVER exposed to WASM; credentials are injected at host boundary +// - Workspace writes are prefixed with channels// to prevent escape +// - Message emission is rate-limited + +package near:agent; + +/// Host-provided capabilities for sandboxed channels. +/// +/// Extends base tool capabilities with channel-specific functions: +/// - emit-message: Queue messages for delivery to the agent +/// - workspace-write: Write to channel-namespaced workspace +interface channel-host { + // ==================== Base Capabilities (from tool host) ==================== + + /// Log levels for structured logging. + enum log-level { + trace, + debug, + info, + warn, + error, + } + + /// Emit a log message. + /// + /// Messages are collected and emitted after execution completes. + /// Rate-limited to 1000 entries per execution, 4KB per message. + log: func(level: log-level, message: string); + + /// Get the current timestamp in milliseconds since Unix epoch. + now-millis: func() -> u64; + + /// Read a file from the workspace. + /// + /// Path is automatically prefixed with channels//. + /// Path must be relative (no leading /) and cannot contain "..". + /// Returns None if the file doesn't exist. + workspace-read: func(path: string) -> option; + + /// Response from an HTTP request. + record http-response { + /// HTTP status code. + status: u16, + /// Response headers as JSON object string. + headers-json: string, + /// Response body bytes. + body: list, + } + + /// Make an HTTP request (if capability granted). + /// + /// Security: + /// - Only allowed endpoints (host/path patterns) can be accessed + /// - Credentials are injected by the host; WASM never sees them + /// - Response is scanned for leaked secrets before returning + /// - Rate-limited per channel + http-request: func( + method: string, + url: string, + headers-json: string, + body: option> + ) -> result; + + /// Check if a secret exists (if capability granted). + /// + /// Security: + /// - WASM can only check existence, NEVER read values + /// - Only allowed secret names can be checked + /// - Actual credentials are injected by host during HTTP requests + secret-exists: func(name: string) -> bool; + + // ==================== Channel-Specific Capabilities ==================== + + /// A message to emit to the agent. + record emitted-message { + /// User identifier within the channel (e.g., Slack user ID). + user-id: string, + /// Optional human-readable user name. + user-name: option, + /// Message content. + content: string, + /// Optional thread ID for threaded conversations. + thread-id: option, + /// Channel-specific metadata as JSON string. + metadata-json: string, + } + + /// Emit a message to the agent. + /// + /// Messages are queued during callback execution and delivered after + /// the callback completes successfully. + /// + /// Security: + /// - Rate-limited per execution (max 100 messages) + /// - Rate-limited globally per channel (configurable) + /// - Content size limited to 64KB + emit-message: func(msg: emitted-message); + + /// Write a file to the workspace. + /// + /// Path is automatically prefixed with channels//. + /// Path must be relative (no leading /) and cannot contain "..". + /// + /// Returns Err if: + /// - Path validation fails (traversal attempt, absolute path) + /// - Write operation fails + workspace-write: func(path: string, content: string) -> result<_, string>; +} + +/// Channel interface that sandboxed channels must implement. +interface channel { + // ==================== Configuration Types ==================== + + /// Configuration for an HTTP endpoint. + record http-endpoint-config { + /// Path to register (e.g., "/webhook/slack"). + path: string, + /// Allowed HTTP methods (e.g., ["POST"]). + methods: list, + /// Whether the endpoint requires secret validation. + require-secret: bool, + } + + /// Configuration for polling behavior. + record poll-config { + /// Polling interval in milliseconds (minimum 30000). + interval-ms: u32, + /// Whether polling is enabled. + enabled: bool, + } + + /// Channel configuration returned by on-start. + record channel-config { + /// Human-readable display name. + display-name: string, + /// HTTP endpoints to register. + http-endpoints: list, + /// Optional polling configuration. + poll: option, + } + + // ==================== Request/Response Types ==================== + + /// Incoming HTTP request from a webhook. + record incoming-http-request { + /// HTTP method (GET, POST, etc.). + method: string, + /// Request path. + path: string, + /// Request headers as JSON object string. + headers-json: string, + /// Query parameters as JSON object string. + query-json: string, + /// Request body bytes. + body: list, + /// Whether the webhook secret was validated by the host. + secret-validated: bool, + } + + /// HTTP response to return to the webhook caller. + record outgoing-http-response { + /// HTTP status code. + status: u16, + /// Response headers as JSON object string. + headers-json: string, + /// Response body bytes. + body: list, + } + + /// Agent response to be sent back to the channel. + record agent-response { + /// Unique message ID for correlation. + message-id: string, + /// Response content from the agent. + content: string, + /// Optional thread ID for threaded replies. + thread-id: option, + /// Channel-specific metadata as JSON string. + metadata-json: string, + } + + // ==================== Lifecycle Callbacks ==================== + + /// Initialize the channel. + /// + /// Called once when the channel is loaded. Returns configuration + /// describing HTTP endpoints and polling behavior. + /// + /// Arguments: + /// - config-json: Channel configuration from the capabilities file. + /// + /// Returns: + /// - Ok(channel-config): Configuration for the host to set up routing + /// - Err(string): Initialization failure message + on-start: func(config-json: string) -> result; + + /// Handle an incoming HTTP request. + /// + /// Called for each HTTP request to a registered endpoint. + /// Use emit-message to queue messages for the agent. + /// + /// Arguments: + /// - req: The incoming HTTP request + /// + /// Returns: + /// - HTTP response to send back to the caller + on-http-request: func(req: incoming-http-request) -> outgoing-http-response; + + /// Handle a polling tick. + /// + /// Called periodically if polling is configured. + /// Use emit-message to queue messages discovered during polling. + on-poll: func(); + + /// Deliver an agent response to the channel. + /// + /// Called when the agent has generated a response to a message + /// that was emitted by this channel. + /// + /// Arguments: + /// - response: The agent's response + /// + /// Returns: + /// - Ok: Response delivered successfully + /// - Err(string): Delivery failure message + on-respond: func(response: agent-response) -> result<_, string>; + + /// Clean up channel resources. + /// + /// Called when the channel is being unloaded. + on-shutdown: func(); +} + +/// World definition for sandboxed channels. +/// +/// Channels import host capabilities and export the channel interface. +world sandboxed-channel { + import channel-host; + export channel; +}