diff --git a/src/app.rs b/src/app.rs index f553b726..6394625b 100644 --- a/src/app.rs +++ b/src/app.rs @@ -572,7 +572,7 @@ impl AppBuilder { let (dev_loaded_tool_names, _) = tokio::join!(wasm_tools_future, mcp_servers_future); // Load registry catalog entries for extension discovery - let catalog_entries = match crate::registry::RegistryCatalog::load_or_embedded() { + let mut catalog_entries = match crate::registry::RegistryCatalog::load_or_embedded() { Ok(catalog) => { let entries: Vec<_> = catalog .all() @@ -591,6 +591,15 @@ impl AppBuilder { } }; + // Append builtin entries (e.g. channel-relay integrations) so they appear + // in the web UI's available extensions list. + let builtin = crate::extensions::registry::builtin_entries(); + for entry in builtin { + if !catalog_entries.iter().any(|e| e.name == entry.name) { + catalog_entries.push(entry); + } + } + // Create extension manager. Use ephemeral in-memory secrets if no // persistent store is configured (listing/install/activate still work). let ext_secrets: Arc = if let Some(ref s) = diff --git a/src/channels/manager.rs b/src/channels/manager.rs index a0fdc087..b026ff85 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -56,6 +56,17 @@ impl ChannelManager { /// the agent loop. pub async fn hot_add(&self, channel: Box) -> Result<(), ChannelError> { let name = channel.name().to_string(); + + // Shut down any existing channel with the same name to avoid parallel consumers. + // The old forwarding task will stop when the channel's stream ends after shutdown. + { + let channels = self.channels.read().await; + if let Some(existing) = channels.get(&name) { + tracing::debug!(channel = %name, "Shutting down existing channel before hot-add replacement"); + let _ = existing.shutdown().await; + } + } + let stream = channel.start().await?; // Register for respond/broadcast/send_status @@ -337,4 +348,30 @@ mod tests { let msg = stream.next().await.expect("stream ended"); assert_eq!(msg.content, "background alert"); } + + #[tokio::test] + async fn test_hot_add_replaces_existing_channel() { + // Regression: hot_add must shut down the existing channel before replacing it, + // to prevent duplicate SSE consumers from running in parallel. + let manager = ChannelManager::new(); + let (stub1, _tx1) = StubChannel::new("relay"); + manager.add(Box::new(stub1)).await; + let mut stream = manager.start_all().await.expect("start_all"); + + // Hot-add a replacement channel with the same name + let (stub2, tx2) = StubChannel::new("relay"); + manager.hot_add(Box::new(stub2)).await.expect("hot_add"); + + // Send through the new channel — should arrive in the merged stream + tx2.send(IncomingMessage::new("relay", "u1", "from new")) + .await + .expect("send"); + let msg = stream.next().await.expect("stream"); + assert_eq!(msg.content, "from new"); + + // Verify only one channel entry exists + let channels = manager.channels.read().await; + assert_eq!(channels.len(), 1); + assert!(channels.contains_key("relay")); + } } diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 038b432f..289b64c7 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -30,6 +30,7 @@ mod channel; mod http; mod manager; +pub mod relay; mod repl; mod signal; pub mod wasm; diff --git a/src/channels/relay/channel.rs b/src/channels/relay/channel.rs new file mode 100644 index 00000000..cb64e882 --- /dev/null +++ b/src/channels/relay/channel.rs @@ -0,0 +1,642 @@ +//! Channel trait implementation for channel-relay SSE streams. +//! +//! `RelayChannel` connects to a channel-relay service via SSE, converts +//! incoming events to `IncomingMessage`s, and sends responses via the +//! relay's provider-specific proxy API (Slack). + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use tokio::sync::{RwLock, mpsc}; + +use crate::channels::relay::client::{RelayClient, RelayError}; +use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; +use crate::error::ChannelError; + +/// Default channel name for the Slack relay integration. +pub const DEFAULT_RELAY_NAME: &str = "slack-relay"; + +/// The messaging provider backing a relay channel. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RelayProvider { + Slack, +} + +impl RelayProvider { + /// Provider string used in proxy API routes and metadata. + pub fn as_str(&self) -> &'static str { + match self { + Self::Slack => "slack", + } + } + + /// The default channel name for this provider. + pub fn channel_name(&self) -> &'static str { + match self { + Self::Slack => DEFAULT_RELAY_NAME, + } + } +} + +/// Channel implementation that connects to a channel-relay SSE stream. +pub struct RelayChannel { + client: RelayClient, + provider: RelayProvider, + stream_token: Arc>, + team_id: String, + instance_id: String, + user_id: String, + /// SSE stream long-poll timeout in seconds. + stream_timeout_secs: u64, + /// Initial exponential backoff in milliseconds. + backoff_initial_ms: u64, + /// Maximum exponential backoff in milliseconds. + backoff_max_ms: u64, + /// Handle to the reconnect task for clean shutdown. + reconnect_handle: RwLock>>, + /// Handle to the SSE parser task for clean shutdown. + parser_handle: Arc>>>, + /// Maximum consecutive reconnect failures before giving up. + max_consecutive_failures: u64, +} + +impl RelayChannel { + /// Create a new relay channel for Slack (default provider). + pub fn new( + client: RelayClient, + stream_token: String, + team_id: String, + instance_id: String, + user_id: String, + ) -> Self { + Self::new_with_provider( + client, + RelayProvider::Slack, + stream_token, + team_id, + instance_id, + user_id, + ) + } + + /// Create a new relay channel with a specific provider. + pub fn new_with_provider( + client: RelayClient, + provider: RelayProvider, + stream_token: String, + team_id: String, + instance_id: String, + user_id: String, + ) -> Self { + Self { + client, + provider, + stream_token: Arc::new(RwLock::new(stream_token)), + team_id, + instance_id, + user_id, + stream_timeout_secs: 86400, + backoff_initial_ms: 1000, + backoff_max_ms: 60000, + reconnect_handle: RwLock::new(None), + parser_handle: Arc::new(RwLock::new(None)), + max_consecutive_failures: 50, + } + } + + /// Set backoff/timeout parameters from relay config values. + pub fn with_timeouts( + mut self, + stream_timeout_secs: u64, + backoff_initial_ms: u64, + backoff_max_ms: u64, + ) -> Self { + self.stream_timeout_secs = stream_timeout_secs; + self.backoff_initial_ms = backoff_initial_ms; + self.backoff_max_ms = backoff_max_ms; + self + } + + /// Set the maximum number of consecutive reconnect failures before giving up. + pub fn with_max_failures(mut self, max: u64) -> Self { + self.max_consecutive_failures = max; + self + } + + /// Build a provider-appropriate proxy body for sending a message. + fn build_send_body( + &self, + channel_id: &str, + text: &str, + thread_id: Option<&str>, + ) -> (String, serde_json::Value) { + match self.provider { + RelayProvider::Slack => { + let mut body = serde_json::json!({ + "channel": channel_id, + "text": text, + }); + if let Some(tid) = thread_id { + body["thread_ts"] = serde_json::Value::String(tid.to_string()); + } + ("chat.postMessage".to_string(), body) + } + } + } + + /// Send a message via the provider proxy. + async fn proxy_send( + &self, + team_id: &str, + method: &str, + body: serde_json::Value, + ) -> Result { + self.client + .proxy_provider( + self.provider.as_str(), + team_id, + method, + body, + Some(&self.instance_id), + ) + .await + } +} + +#[async_trait] +impl Channel for RelayChannel { + fn name(&self) -> &str { + self.provider.channel_name() + } + + async fn start(&self) -> Result { + let channel_name = self.name().to_string(); + let token = self.stream_token.read().await.clone(); + let (stream, initial_parser_handle) = self + .client + .connect_stream(&token, self.stream_timeout_secs) + .await + .map_err(|e| ChannelError::StartupFailed { + name: channel_name.clone(), + reason: e.to_string(), + })?; + + *self.parser_handle.write().await = Some(initial_parser_handle); + + let (tx, rx) = mpsc::channel(64); + + // Spawn the stream reader + reconnect task + let client = self.client.clone(); + let stream_token = Arc::clone(&self.stream_token); + let instance_id = self.instance_id.clone(); + let user_id = self.user_id.clone(); + let team_id = self.team_id.clone(); + let stream_timeout_secs = self.stream_timeout_secs; + let backoff_initial_ms = self.backoff_initial_ms; + let backoff_max_ms = self.backoff_max_ms; + let max_consecutive_failures = self.max_consecutive_failures; + let parser_handle = Arc::clone(&self.parser_handle); + let provider_str = self.provider.as_str().to_string(); + let relay_name = channel_name.clone(); + + let handle = tokio::spawn(async move { + use futures::StreamExt; + + let mut current_stream = stream; + let mut backoff_ms = backoff_initial_ms; + let mut consecutive_failures: u64 = 0; + + loop { + // Read events from the current stream + while let Some(event) = current_stream.next().await { + // Reset backoff and failure count on successful event + backoff_ms = backoff_initial_ms; + consecutive_failures = 0; + + // Validate required fields + if event.sender_id.is_empty() + || event.channel_id.is_empty() + || event.provider_scope.is_empty() + { + tracing::debug!( + event_type = %event.event_type, + sender_id = %event.sender_id, + channel_id = %event.channel_id, + "Relay: skipping event with missing required fields" + ); + continue; + } + + // Skip non-message events + if !event.is_message() { + tracing::debug!( + event_type = %event.event_type, + "Relay: skipping non-message event" + ); + continue; + } + + tracing::info!( + event_type = %event.event_type, + sender = %event.sender_id, + channel = %event.channel_id, + provider = %provider_str, + "Relay: received message from {}", provider_str + ); + + let msg = IncomingMessage::new(&relay_name, &event.sender_id, event.text()) + .with_user_name(event.display_name()) + .with_metadata(serde_json::json!({ + "team_id": event.team_id(), + "channel_id": event.channel_id, + "sender_id": event.sender_id, + "sender_name": event.display_name(), + "event_type": event.event_type, + "thread_id": event.thread_id, + "provider": event.provider, + })); + + let msg = if let Some(ref thread_id) = event.thread_id { + msg.with_thread(thread_id) + } else { + msg.with_thread(&event.channel_id) + }; + + if tx.send(msg).await.is_err() { + tracing::info!("Relay channel receiver dropped, stopping"); + return; + } + } + + // Stream ended, attempt reconnect with backoff + consecutive_failures += 1; + if consecutive_failures >= max_consecutive_failures { + tracing::error!( + channel = %relay_name, + failures = consecutive_failures, + "Relay channel giving up after {} consecutive failures", + consecutive_failures + ); + break; + } + + tracing::warn!( + backoff_ms = backoff_ms, + failures = consecutive_failures, + "Relay SSE stream ended, reconnecting..." + ); + tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await; + backoff_ms = (backoff_ms * 2).min(backoff_max_ms); + + // Try to reconnect + let token = stream_token.read().await.clone(); + match client.connect_stream(&token, stream_timeout_secs).await { + Ok((new_stream, new_parser)) => { + tracing::info!("Relay SSE stream reconnected"); + current_stream = new_stream; + // Abort old parser before replacing + if let Some(old) = parser_handle.write().await.take() { + old.abort(); + } + *parser_handle.write().await = Some(new_parser); + } + Err(RelayError::TokenExpired) => { + // Attempt token renewal + tracing::info!("Relay stream token expired, renewing..."); + match client.renew_token(&instance_id, &user_id).await { + Ok(new_token) => { + *stream_token.write().await = new_token.clone(); + match client.connect_stream(&new_token, stream_timeout_secs).await { + Ok((new_stream, new_parser)) => { + tracing::info!( + "Relay SSE stream reconnected with new token" + ); + current_stream = new_stream; + if let Some(old) = parser_handle.write().await.take() { + old.abort(); + } + *parser_handle.write().await = Some(new_parser); + } + Err(e) => { + tracing::error!( + error = %e, + "Failed to reconnect after token renewal" + ); + } + } + } + Err(e) => { + tracing::error!( + error = %e, + "Failed to renew relay stream token" + ); + } + } + } + Err(e) => { + tracing::error!(error = %e, "Failed to reconnect relay SSE stream"); + } + } + + // Check if the team is still valid (skip when team_id is unknown, + // e.g. when no DB store was available at activation time) + if !team_id.is_empty() { + match client.list_connections(&instance_id).await { + Ok(conns) => { + let has_team = + conns.iter().any(|c| c.team_id == team_id && c.connected); + if !has_team { + tracing::warn!( + team_id = %team_id, + "Team no longer connected, stopping relay channel" + ); + return; + } + } + Err(e) => { + tracing::warn!( + error = %e, + "Could not verify team connection, will retry next iteration" + ); + } + } + } + } + }); + + *self.reconnect_handle.write().await = Some(handle); + + let stream = tokio_stream::wrappers::ReceiverStream::new(rx); + Ok(Box::pin(stream)) + } + + async fn respond( + &self, + msg: &IncomingMessage, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + let channel_name = self.name().to_string(); + let metadata = &msg.metadata; + let team_id = metadata + .get("team_id") + .and_then(|v| v.as_str()) + .unwrap_or(&self.team_id); + let channel_id = metadata + .get("channel_id") + .and_then(|v| v.as_str()) + .ok_or_else(|| ChannelError::SendFailed { + name: channel_name.clone(), + reason: "Missing channel_id in message metadata".to_string(), + })?; + + // Determine thread_id from response or metadata + let thread_id = response + .thread_id + .as_deref() + .or_else(|| metadata.get("thread_id").and_then(|v| v.as_str())); + + let (method, body) = self.build_send_body(channel_id, &response.content, thread_id); + + self.proxy_send(team_id, &method, body) + .await + .map_err(|e| ChannelError::SendFailed { + name: channel_name, + reason: e.to_string(), + })?; + + Ok(()) + } + + /// Status updates are not forwarded to messaging providers to avoid noise. + async fn send_status( + &self, + _status: StatusUpdate, + _metadata: &serde_json::Value, + ) -> Result<(), ChannelError> { + Ok(()) + } + + async fn broadcast( + &self, + target: &str, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + let channel_name = self.name().to_string(); + + // Determine thread_id from response or metadata + let thread_id = response + .thread_id + .as_deref() + .or_else(|| response.metadata.get("thread_ts").and_then(|v| v.as_str())); + + let (method, body) = self.build_send_body(target, &response.content, thread_id); + + self.proxy_send(&self.team_id, &method, body) + .await + .map_err(|e| ChannelError::SendFailed { + name: channel_name, + reason: e.to_string(), + })?; + + Ok(()) + } + + async fn health_check(&self) -> Result<(), ChannelError> { + self.client + .list_connections(&self.instance_id) + .await + .map_err(|_| ChannelError::HealthCheckFailed { + name: self.name().to_string(), + })?; + Ok(()) + } + + fn conversation_context(&self, metadata: &serde_json::Value) -> HashMap { + let mut ctx = HashMap::new(); + + if let Some(sender) = metadata.get("sender_name").and_then(|v| v.as_str()) { + ctx.insert("sender".to_string(), sender.to_string()); + } + if let Some(sender_id) = metadata.get("sender_id").and_then(|v| v.as_str()) { + ctx.insert("sender_uuid".to_string(), sender_id.to_string()); + } + if let Some(channel_id) = metadata.get("channel_id").and_then(|v| v.as_str()) { + ctx.insert("group".to_string(), channel_id.to_string()); + } + ctx.insert("platform".to_string(), self.provider.as_str().to_string()); + + ctx + } + + async fn shutdown(&self) -> Result<(), ChannelError> { + if let Some(handle) = self.reconnect_handle.write().await.take() { + handle.abort(); + } + if let Some(handle) = self.parser_handle.write().await.take() { + handle.abort(); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_client() -> RelayClient { + RelayClient::new( + "http://localhost:3001".into(), + secrecy::SecretString::from("key".to_string()), + 30, + ) + .expect("client") + } + + #[test] + fn relay_channel_name() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ); + assert_eq!(channel.name(), DEFAULT_RELAY_NAME); + } + + #[test] + fn conversation_context_extracts_metadata() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ); + + let metadata = serde_json::json!({ + "sender_name": "bob", + "sender_id": "U123", + "channel_id": "C456", + }); + let ctx = channel.conversation_context(&metadata); + assert_eq!(ctx.get("sender"), Some(&"bob".to_string())); + assert_eq!(ctx.get("sender_uuid"), Some(&"U123".to_string())); + assert_eq!(ctx.get("platform"), Some(&"slack".to_string())); + } + + #[test] + fn metadata_shape_includes_event_type_and_sender_name() { + // Regression: metadata JSON must include event_type and sender_name + // for downstream routing (DM vs channel) and conversation_context(). + let metadata = serde_json::json!({ + "team_id": "T123", + "channel_id": "C456", + "sender_id": "U789", + "sender_name": "alice", + "event_type": "direct_message", + "thread_id": null, + "provider": "slack", + }); + // event_type must be present for DM-vs-channel routing + assert_eq!( + metadata.get("event_type").and_then(|v| v.as_str()), + Some("direct_message") + ); + // sender_name must be present for conversation_context + assert_eq!( + metadata.get("sender_name").and_then(|v| v.as_str()), + Some("alice") + ); + } + + #[test] + fn with_timeouts_sets_values() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ) + .with_timeouts(43200, 2000, 120000); + + assert_eq!(channel.stream_timeout_secs, 43200); + assert_eq!(channel.backoff_initial_ms, 2000); + assert_eq!(channel.backoff_max_ms, 120000); + } + + #[test] + fn build_send_body_slack() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ); + let (method, body) = channel.build_send_body("C456", "hello", Some("1234567.890")); + assert_eq!(method, "chat.postMessage"); + assert_eq!(body["channel"], "C456"); + assert_eq!(body["text"], "hello"); + assert_eq!(body["thread_ts"], "1234567.890"); + } + + #[test] + fn parser_handle_is_shared_arc() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ); + // parser_handle should be an Arc — cloning should give a second reference + let handle_clone = Arc::clone(&channel.parser_handle); + // Both point to the same allocation + assert!(Arc::ptr_eq(&channel.parser_handle, &handle_clone)); + } + + #[test] + fn with_max_failures_sets_value() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ) + .with_max_failures(10); + + assert_eq!(channel.max_consecutive_failures, 10); + } + + #[test] + fn default_max_failures_is_50() { + let channel = RelayChannel::new( + test_client(), + "token".into(), + "T123".into(), + "inst1".into(), + "user1".into(), + ); + assert_eq!(channel.max_consecutive_failures, 50); + } + + #[test] + fn empty_team_id_accepted_at_construction() { + // Regression: empty team_id (when no DB store is available) must not + // prevent channel construction or cause immediate shutdown. + let channel = RelayChannel::new( + test_client(), + "token".into(), + String::new(), // empty team_id + "inst1".into(), + "user1".into(), + ); + assert_eq!(channel.team_id, ""); + // The reconnect loop now skips team validation when team_id is empty, + // so the channel remains alive. + } +} diff --git a/src/channels/relay/client.rs b/src/channels/relay/client.rs new file mode 100644 index 00000000..d1c03a51 --- /dev/null +++ b/src/channels/relay/client.rs @@ -0,0 +1,549 @@ +//! HTTP client for the channel-relay service. +//! +//! Wraps reqwest for all channel-relay API calls: OAuth initiation, +//! SSE streaming, token renewal, and Slack API proxy. + +use std::pin::Pin; +use std::task::{Context, Poll}; + +use futures::Stream; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use tokio::sync::mpsc; + +/// Known relay event types. +pub mod event_types { + pub const MESSAGE: &str = "message"; + pub const DIRECT_MESSAGE: &str = "direct_message"; + pub const MENTION: &str = "mention"; +} + +/// A parsed SSE event from the channel-relay stream. +/// +/// Field names match the channel-relay `ChannelEvent` struct exactly. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ChannelEvent { + /// Unique event ID. + #[serde(default)] + pub id: String, + /// Event type enum from channel-relay (e.g., "direct_message", "message", "mention"). + pub event_type: String, + /// Provider (e.g., "slack"). + #[serde(default)] + pub provider: String, + /// Team/workspace ID (called `provider_scope` in channel-relay). + #[serde(alias = "team_id", default)] + pub provider_scope: String, + /// Channel or DM conversation ID. + #[serde(default)] + pub channel_id: String, + /// Sender user ID. + #[serde(default)] + pub sender_id: String, + /// Sender display name. + #[serde(default)] + pub sender_name: Option, + /// Message text content (called `content` in channel-relay). + #[serde(alias = "text", default)] + pub content: Option, + /// Thread ID (for threaded replies, called `thread_id` in channel-relay). + #[serde(alias = "thread_ts", default)] + pub thread_id: Option, + /// Full raw event data. + #[serde(default)] + pub raw: serde_json::Value, + /// Event timestamp (ISO 8601 from channel-relay). + #[serde(default)] + pub timestamp: Option, +} + +impl ChannelEvent { + /// Get the team_id (provider_scope). + pub fn team_id(&self) -> &str { + &self.provider_scope + } + + /// Get the message text content. + pub fn text(&self) -> &str { + self.content.as_deref().unwrap_or("") + } + + /// Get the sender name or fallback to sender_id. + pub fn display_name(&self) -> &str { + self.sender_name.as_deref().unwrap_or(&self.sender_id) + } + + /// Check if this is a message-like event that should be forwarded to the agent. + pub fn is_message(&self) -> bool { + matches!( + self.event_type.as_str(), + event_types::MESSAGE | event_types::DIRECT_MESSAGE | event_types::MENTION + ) + } +} + +/// Connection info returned by list_connections. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Connection { + pub provider: String, + pub team_id: String, + pub team_name: Option, + pub connected: bool, +} + +/// HTTP client for the channel-relay service. +#[derive(Clone)] +pub struct RelayClient { + http: reqwest::Client, + base_url: String, + api_key: SecretString, +} + +impl RelayClient { + /// Create a new relay client. + pub fn new( + base_url: String, + api_key: SecretString, + request_timeout_secs: u64, + ) -> Result { + let http = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(request_timeout_secs)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|e| RelayError::Network(format!("Failed to build HTTP client: {e}")))?; + + Ok(Self { + http, + base_url: base_url.trim_end_matches('/').to_string(), + api_key, + }) + } + + /// Initiate Slack OAuth flow via channel-relay. + /// + /// Calls `GET /oauth/slack/auth` with `redirect(Policy::none())` and + /// returns the `Location` header (Slack OAuth URL) without following it. + pub async fn initiate_oauth( + &self, + instance_id: &str, + user_id: &str, + callback_url: &str, + ) -> Result { + let resp = self + .http + .get(format!("{}/oauth/slack/auth", self.base_url)) + .header("X-API-Key", self.api_key.expose_secret()) + .query(&[ + ("instance_id", instance_id), + ("user_id", user_id), + ("callback", callback_url), + ]) + .send() + .await + .map_err(|e| RelayError::Network(e.to_string()))?; + + let status = resp.status(); + if status.is_redirection() { + let location = resp + .headers() + .get(reqwest::header::LOCATION) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) + .ok_or_else(|| { + RelayError::Protocol("Redirect response missing Location header".to_string()) + })?; + Ok(location) + } else if status.is_success() { + // Some relay implementations return the URL in JSON body instead + let body: serde_json::Value = resp + .json() + .await + .map_err(|e| RelayError::Protocol(e.to_string()))?; + body.get("auth_url") + .or_else(|| body.get("url")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .ok_or_else(|| RelayError::Protocol("Response missing auth_url field".to_string())) + } else { + let body = resp.text().await.unwrap_or_default(); + Err(RelayError::Api { + status: status.as_u16(), + message: body, + }) + } + } + + /// Connect to the SSE event stream. + /// + /// Returns a stream of parsed `ChannelEvent`s and the `JoinHandle` of the + /// background SSE parser task. The caller is responsible for reconnection + /// logic on stream end/error and for aborting the handle on shutdown. + pub async fn connect_stream( + &self, + stream_token: &str, + stream_timeout_secs: u64, + ) -> Result<(ChannelEventStream, tokio::task::JoinHandle<()>), RelayError> { + let resp = self + .http + .get(format!("{}/stream", self.base_url)) + .query(&[("token", stream_token)]) + .timeout(std::time::Duration::from_secs(stream_timeout_secs)) + .send() + .await + .map_err(|e| RelayError::Network(e.to_string()))?; + + let status = resp.status(); + if status == reqwest::StatusCode::UNAUTHORIZED { + return Err(RelayError::TokenExpired); + } + if !status.is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(RelayError::Api { + status: status.as_u16(), + message: body, + }); + } + + // Spawn a background task that reads the SSE stream and sends parsed events + let (tx, rx) = mpsc::channel(64); + let byte_stream = resp.bytes_stream(); + let handle = tokio::spawn(parse_sse_stream(byte_stream, tx)); + + Ok((ChannelEventStream { rx }, handle)) + } + + /// Renew an expired stream token. + /// + /// Calls `POST /stream/renew` with API key auth, returns a new stream token. + pub async fn renew_token( + &self, + instance_id: &str, + user_id: &str, + ) -> Result { + let resp = self + .http + .post(format!("{}/stream/renew", self.base_url)) + .header("X-API-Key", self.api_key.expose_secret()) + .json(&serde_json::json!({ + "instance_id": instance_id, + "user_id": user_id, + })) + .send() + .await + .map_err(|e| RelayError::Network(e.to_string()))?; + + let status = resp.status(); + if !status.is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(RelayError::Api { + status: status.as_u16(), + message: body, + }); + } + + let body: serde_json::Value = resp + .json() + .await + .map_err(|e| RelayError::Protocol(e.to_string()))?; + body.get("stream_token") + .or_else(|| body.get("token")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .ok_or_else(|| RelayError::Protocol("Response missing stream_token field".to_string())) + } + + /// Proxy an API call through channel-relay for any provider. + /// + /// Calls `POST /proxy/{provider}/{method}?team_id=X&instance_id=Y` with the given JSON body. + pub async fn proxy_provider( + &self, + provider: &str, + team_id: &str, + method: &str, + body: serde_json::Value, + instance_id: Option<&str>, + ) -> Result { + let mut query: Vec<(&str, &str)> = vec![("team_id", team_id)]; + if let Some(iid) = instance_id { + query.push(("instance_id", iid)); + } + let resp = self + .http + .post(format!("{}/proxy/{}/{}", self.base_url, provider, method)) + .header("X-API-Key", self.api_key.expose_secret()) + .query(&query) + .json(&body) + .send() + .await + .map_err(|e| RelayError::Network(e.to_string()))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(RelayError::Api { + status, + message: body, + }); + } + + resp.json() + .await + .map_err(|e| RelayError::Protocol(e.to_string())) + } + + /// List active connections for an instance. + pub async fn list_connections(&self, instance_id: &str) -> Result, RelayError> { + let resp = self + .http + .get(format!("{}/connections", self.base_url)) + .header("X-API-Key", self.api_key.expose_secret()) + .query(&[("instance_id", instance_id)]) + .send() + .await + .map_err(|e| RelayError::Network(e.to_string()))?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(RelayError::Api { + status, + message: body, + }); + } + + resp.json() + .await + .map_err(|e| RelayError::Protocol(e.to_string())) + } +} + +/// Async stream of parsed channel events from SSE. +pub struct ChannelEventStream { + rx: mpsc::Receiver, +} + +impl Stream for ChannelEventStream { + type Item = ChannelEvent; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.rx.poll_recv(cx) + } +} + +/// Parse SSE format from a reqwest bytes stream. +/// +/// SSE format: +/// ```text +/// event: message +/// data: {"key": "value"} +/// +/// ``` +/// Blank line terminates an event. +async fn parse_sse_stream( + byte_stream: impl futures::Stream> + Send + 'static, + tx: mpsc::Sender, +) { + use futures::StreamExt; + + let mut buffer = Vec::::new(); + let mut event_type = String::new(); + let mut data_lines = Vec::new(); + + let mut byte_stream = std::pin::pin!(byte_stream); + while let Some(chunk_result) = byte_stream.next().await { + let chunk = match chunk_result { + Ok(c) => c, + Err(e) => { + tracing::debug!(error = %e, "SSE stream chunk error"); + break; + } + }; + + buffer.extend_from_slice(&chunk); + + // Process complete lines (decode UTF-8 only on full lines to avoid + // corruption when multi-byte characters span chunk boundaries) + while let Some(newline_pos) = buffer.iter().position(|&b| b == b'\n') { + let line = String::from_utf8_lossy(&buffer[..newline_pos]) + .trim_end_matches('\r') + .to_string(); + buffer.drain(..=newline_pos); + + if line.is_empty() { + // Blank line = end of event + if !data_lines.is_empty() { + let data = data_lines.join("\n"); + if let Ok(mut event) = serde_json::from_str::(&data) { + if event.event_type.is_empty() && !event_type.is_empty() { + event.event_type = event_type.clone(); + } + if tx.send(event).await.is_err() { + return; // receiver dropped + } + } else { + tracing::debug!( + event_type = %event_type, + data_len = data.len(), + "Failed to parse SSE event data as ChannelEvent" + ); + } + } + event_type.clear(); + data_lines.clear(); + } else if let Some(value) = line.strip_prefix("event:") { + event_type = value.trim().to_string(); + } else if let Some(value) = line.strip_prefix("data:") { + data_lines.push(value.trim().to_string()); + } + // Ignore other fields (id:, retry:, comments) + } + } + + tracing::debug!("SSE stream ended"); +} + +/// Errors from relay client operations. +#[derive(Debug, thiserror::Error)] +pub enum RelayError { + #[error("Network error: {0}")] + Network(String), + + #[error("API error (HTTP {status}): {message}")] + Api { status: u16, message: String }, + + #[error("Protocol error: {0}")] + Protocol(String), + + #[error("Stream token expired")] + TokenExpired, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn channel_event_deserialize_minimal() { + let json = r#"{"event_type": "message", "content": "hello"}"#; + let event: ChannelEvent = serde_json::from_str(json).expect("parse failed"); + assert_eq!(event.event_type, "message"); + assert_eq!(event.text(), "hello"); + assert!(event.provider_scope.is_empty()); + } + + #[test] + fn channel_event_deserialize_relay_format() { + // Matches the actual channel-relay ChannelEvent serialization format. + let json = r#"{ + "id": "evt_123", + "event_type": "direct_message", + "provider": "slack", + "provider_scope": "T123", + "channel_id": "D456", + "sender_id": "U789", + "sender_name": "bob", + "content": "hi there", + "thread_id": "1234567890.123456", + "raw": {}, + "timestamp": "2026-03-09T21:00:00Z" + }"#; + let event: ChannelEvent = serde_json::from_str(json).expect("parse failed"); + assert_eq!(event.provider, "slack"); + assert_eq!(event.team_id(), "T123"); + assert_eq!(event.display_name(), "bob"); + assert_eq!(event.thread_id, Some("1234567890.123456".to_string())); + assert!(event.is_message()); + } + + #[test] + fn channel_event_is_message() { + let make = |et: &str| ChannelEvent { + id: String::new(), + event_type: et.to_string(), + provider: String::new(), + provider_scope: String::new(), + channel_id: String::new(), + sender_id: String::new(), + sender_name: None, + content: None, + thread_id: None, + raw: serde_json::Value::Null, + timestamp: None, + }; + assert!(make("message").is_message()); + assert!(make("direct_message").is_message()); + assert!(make("mention").is_message()); + assert!(!make("reaction").is_message()); + } + + #[test] + fn connection_deserialize() { + let json = r#"{"provider": "slack", "team_id": "T123", "team_name": "My Team", "connected": true}"#; + let conn: Connection = serde_json::from_str(json).expect("parse failed"); + assert_eq!(conn.provider, "slack"); + assert!(conn.connected); + } + + #[test] + fn relay_error_display() { + let err = RelayError::Network("timeout".into()); + assert_eq!(err.to_string(), "Network error: timeout"); + + let err = RelayError::Api { + status: 401, + message: "unauthorized".into(), + }; + assert_eq!(err.to_string(), "API error (HTTP 401): unauthorized"); + + let err = RelayError::TokenExpired; + assert_eq!(err.to_string(), "Stream token expired"); + } + + #[test] + fn event_type_constants_match_is_message() { + let make = |et: &str| ChannelEvent { + id: String::new(), + event_type: et.to_string(), + provider: String::new(), + provider_scope: String::new(), + channel_id: String::new(), + sender_id: String::new(), + sender_name: None, + content: None, + thread_id: None, + raw: serde_json::Value::Null, + timestamp: None, + }; + assert!(make(event_types::MESSAGE).is_message()); + assert!(make(event_types::DIRECT_MESSAGE).is_message()); + assert!(make(event_types::MENTION).is_message()); + } + + #[tokio::test] + async fn parse_sse_handles_multibyte_utf8_across_chunks() { + // The crab emoji (🦀) is 4 bytes: [0xF0, 0x9F, 0xA6, 0x80]. + // Split it across two chunks to verify no U+FFFD corruption. + let event_json = r#"{"event_type":"message","content":"hello 🦀 world","provider_scope":"T1","channel_id":"C1","sender_id":"U1"}"#; + let full = format!("event: message\ndata: {}\n\n", event_json); + let bytes = full.as_bytes(); + + // Find the crab emoji and split mid-character + let crab_pos = bytes + .windows(4) + .position(|w| w == [0xF0, 0x9F, 0xA6, 0x80]) + .expect("crab emoji not found"); + let split_at = crab_pos + 2; // split in the middle of the 4-byte emoji + + let chunk1 = bytes::Bytes::copy_from_slice(&bytes[..split_at]); + let chunk2 = bytes::Bytes::copy_from_slice(&bytes[split_at..]); + + let chunks: Vec> = vec![Ok(chunk1), Ok(chunk2)]; + let stream = futures::stream::iter(chunks); + + let (tx, mut rx) = mpsc::channel(8); + parse_sse_stream(stream, tx).await; + + let event = rx.recv().await.expect("should receive event"); + assert_eq!(event.text(), "hello 🦀 world"); + } +} diff --git a/src/channels/relay/mod.rs b/src/channels/relay/mod.rs new file mode 100644 index 00000000..1582319f --- /dev/null +++ b/src/channels/relay/mod.rs @@ -0,0 +1,12 @@ +//! Channel-relay integration for connecting to external messaging platforms +//! (Slack) via the channel-relay service. +//! +//! The relay service handles OAuth, credential storage, webhook ingestion, +//! and SSE event streaming. IronClaw consumes the SSE stream and sends +//! messages via the relay's proxy API. + +pub mod channel; +pub mod client; + +pub use channel::{DEFAULT_RELAY_NAME, RelayChannel}; +pub use client::RelayClient; diff --git a/src/channels/web/handlers/extensions.rs b/src/channels/web/handlers/extensions.rs index 078af7dc..3c490eac 100644 --- a/src/channels/web/handlers/extensions.rs +++ b/src/channels/web/handlers/extensions.rs @@ -46,6 +46,14 @@ pub async fn extensions_list_handler( } else { "configured".to_string() }) + } else if ext.kind == crate::extensions::ExtensionKind::ChannelRelay { + Some(if ext.active { + "active".to_string() + } else if ext.authenticated { + "configured".to_string() + } else { + "installed".to_string() + }) } else { None }; @@ -103,6 +111,7 @@ pub async fn extensions_install_handler( "mcp_server" => Some(crate::extensions::ExtensionKind::McpServer), "wasm_tool" => Some(crate::extensions::ExtensionKind::WasmTool), "wasm_channel" => Some(crate::extensions::ExtensionKind::WasmChannel), + "channel_relay" => Some(crate::extensions::ExtensionKind::ChannelRelay), _ => None, }); @@ -115,62 +124,6 @@ pub async fn extensions_install_handler( } } -pub async fn extensions_activate_handler( - State(state): State>, - Path(name): Path, -) -> Result, (StatusCode, String)> { - let ext_mgr = state.extension_manager.as_ref().ok_or(( - StatusCode::NOT_IMPLEMENTED, - "Extension manager not available (secrets store required)".to_string(), - ))?; - - match ext_mgr.activate(&name).await { - Ok(result) => { - // Activation just loads the WASM module. Auth (OAuth/manual) is - // triggered separately via save_setup_secrets or the auth endpoint. - Ok(Json(ActionResponse::ok(result.message))) - } - Err(activate_err) => { - let err_str = activate_err.to_string(); - let needs_auth = err_str.contains("authentication") - || err_str.contains("401") - || err_str.contains("Unauthorized"); - - if !needs_auth { - return Ok(Json(ActionResponse::fail(err_str))); - } - - // Activation failed due to auth; try authenticating first. - match ext_mgr.auth(&name, None).await { - Ok(auth_result) if auth_result.is_authenticated() => { - // Auth succeeded, retry activation. - match ext_mgr.activate(&name).await { - Ok(result) => Ok(Json(ActionResponse::ok(result.message))), - Err(e) => Ok(Json(ActionResponse::fail(e.to_string()))), - } - } - Ok(auth_result) => { - // Auth in progress (OAuth URL or awaiting manual token). - let mut resp = ActionResponse::fail( - auth_result - .instructions() - .map(String::from) - .unwrap_or_else(|| format!("'{}' requires authentication.", name)), - ); - resp.auth_url = auth_result.auth_url().map(String::from); - resp.awaiting_token = Some(auth_result.is_awaiting_token()); - resp.instructions = auth_result.instructions().map(String::from); - Ok(Json(resp)) - } - Err(auth_err) => Ok(Json(ActionResponse::fail(format!( - "Authentication failed: {}", - auth_err - )))), - } - } - } -} - pub async fn extensions_remove_handler( State(state): State>, Path(name): Path, diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 0fcf228e..b0e1d29e 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -97,6 +97,7 @@ impl GatewayChannel { skill_registry: None, skill_catalog: None, chat_rate_limiter: server::RateLimiter::new(30, 60), + oauth_rate_limiter: server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), @@ -133,6 +134,7 @@ impl GatewayChannel { skill_registry: self.state.skill_registry.clone(), skill_catalog: self.state.skill_catalog.clone(), chat_rate_limiter: server::RateLimiter::new(30, 60), + oauth_rate_limiter: server::RateLimiter::new(10, 60), registry_entries: self.state.registry_entries.clone(), cost_guard: self.state.cost_guard.clone(), routine_engine: Arc::clone(&self.state.routine_engine), diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 9c7561a1..48d3407c 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -28,6 +28,7 @@ use uuid::Uuid; use crate::agent::SessionManager; use crate::bootstrap::ironclaw_base_dir; use crate::channels::IncomingMessage; +use crate::channels::relay::DEFAULT_RELAY_NAME; use crate::channels::web::auth::{AuthState, auth_middleware}; use crate::channels::web::handlers::jobs::{ job_files_list_handler, job_files_read_handler, jobs_cancel_handler, jobs_detail_handler, @@ -164,6 +165,8 @@ pub struct GatewayState { pub scheduler: Option, /// Rate limiter for chat endpoints (30 messages per 60 seconds). pub chat_rate_limiter: RateLimiter, + /// Rate limiter for OAuth callback endpoints (10 requests per 60 seconds). + pub oauth_rate_limiter: RateLimiter, /// Registry catalog entries for the available extensions API. /// Populated at startup from `registry/` manifests, independent of extension manager. pub registry_entries: Vec, @@ -200,7 +203,11 @@ pub async fn start_server( // Public routes (no auth) let public = Router::new() .route("/api/health", get(health_handler)) - .route("/oauth/callback", get(oauth_callback_handler)); + .route("/oauth/callback", get(oauth_callback_handler)) + .route( + "/oauth/slack/callback", + get(slack_relay_oauth_callback_handler), + ); // Protected routes (require auth) let auth_state = AuthState { token: auth_token }; @@ -606,6 +613,208 @@ async fn oauth_callback_handler( axum::response::Html(html).into_response() } +/// OAuth callback for Slack via channel-relay. +/// +/// This is a PUBLIC route (no Bearer token required) because channel-relay +/// redirects the user's browser here after Slack OAuth completes. +/// Query params: `stream_token`, `provider`, `team_id`. +async fn slack_relay_oauth_callback_handler( + State(state): State>, + Query(params): Query>, +) -> impl IntoResponse { + // Rate limit + if !state.oauth_rate_limiter.check() { + return axum::response::Html( + "\ +

Too Many Requests

\ +

Please try again later.

\ + " + .to_string(), + ) + .into_response(); + } + + // Validate stream_token: required, non-empty, max 2048 bytes + let stream_token = match params.get("stream_token") { + Some(t) if !t.is_empty() && t.len() <= 2048 => t.clone(), + Some(t) if t.len() > 2048 => { + return axum::response::Html( + "\ +

Error

Invalid callback parameters.

" + .to_string(), + ) + .into_response(); + } + _ => { + return axum::response::Html( + "\ +

Error

Invalid callback parameters.

" + .to_string(), + ) + .into_response(); + } + }; + + // Validate team_id format: empty or T followed by alphanumeric (max 20 chars) + let team_id = params.get("team_id").cloned().unwrap_or_default(); + if !team_id.is_empty() { + let valid_team_id = team_id.len() <= 21 + && team_id.starts_with('T') + && team_id[1..].chars().all(|c| c.is_ascii_alphanumeric()); + if !valid_team_id { + return axum::response::Html( + "\ +

Error

Invalid callback parameters.

" + .to_string(), + ) + .into_response(); + } + } + + // Validate provider: must be "slack" (only supported provider) + let provider = params + .get("provider") + .cloned() + .unwrap_or_else(|| "slack".into()); + if provider != "slack" { + return axum::response::Html( + "\ +

Error

Invalid callback parameters.

" + .to_string(), + ) + .into_response(); + } + + let ext_mgr = match state.extension_manager.as_ref() { + Some(mgr) => mgr, + None => { + return axum::response::Html( + "\ +

Error

Extension manager not available.

" + .to_string(), + ) + .into_response(); + } + }; + + // Validate CSRF state parameter + let state_param = match params.get("state") { + Some(s) if !s.is_empty() && s.len() <= 128 => s.clone(), + _ => { + return axum::response::Html( + "\ +

Error

Invalid or expired authorization.

" + .to_string(), + ) + .into_response(); + } + }; + + let state_key = format!("relay:{}:oauth_state", DEFAULT_RELAY_NAME); + let stored_state = match ext_mgr + .secrets() + .get_decrypted(&state.user_id, &state_key) + .await + { + Ok(secret) => secret.expose().to_string(), + Err(_) => { + return axum::response::Html( + "\ +

Error

Invalid or expired authorization.

" + .to_string(), + ) + .into_response(); + } + }; + + if state_param != stored_state { + return axum::response::Html( + "\ +

Error

Invalid or expired authorization.

" + .to_string(), + ) + .into_response(); + } + + // Delete the nonce (one-time use) + let _ = ext_mgr.secrets().delete(&state.user_id, &state_key).await; + + let result: Result<(), String> = async { + // Store the stream token as a secret + let token_key = format!("relay:{}:stream_token", DEFAULT_RELAY_NAME); + let _ = ext_mgr.secrets().delete(&state.user_id, &token_key).await; + ext_mgr + .secrets() + .create( + &state.user_id, + crate::secrets::CreateSecretParams { + name: token_key, + value: secrecy::SecretString::from(stream_token), + provider: Some(provider.clone()), + expires_at: None, + }, + ) + .await + .map_err(|e| format!("Failed to store stream token: {}", e))?; + + // Store team_id in settings + if let Some(ref store) = state.store { + let team_id_key = format!("relay:{}:team_id", DEFAULT_RELAY_NAME); + let _ = store + .set_setting(&state.user_id, &team_id_key, &serde_json::json!(team_id)) + .await; + } + + // Activate the relay channel + ext_mgr + .activate_stored_relay(DEFAULT_RELAY_NAME) + .await + .map_err(|e| format!("Failed to activate relay channel: {}", e))?; + + Ok(()) + } + .await; + + let (success, message) = match &result { + Ok(()) => (true, "Slack connected successfully!".to_string()), + Err(e) => { + tracing::error!(error = %e, "Slack relay OAuth callback failed"); + ( + false, + "Connection failed. Check server logs for details.".to_string(), + ) + } + }; + + // Broadcast SSE event to notify the web UI + state.sse.broadcast(SseEvent::AuthCompleted { + extension_name: DEFAULT_RELAY_NAME.to_string(), + success, + message: message.clone(), + }); + + if success { + axum::response::Html( + "\ +

Slack Connected!

\ +

You can close this tab and return to IronClaw.

\ + \ + " + .to_string(), + ) + .into_response() + } else { + axum::response::Html(format!( + "\ +

Connection Failed

\ +

{}

\ + ", + message + )) + .into_response() + } +} + // --- Chat handlers --- /// Convert web gateway `ImageData` to `IncomingAttachment` objects. @@ -1639,13 +1848,13 @@ async fn extensions_activate_handler( Ok(Json(resp)) } Err(activate_err) => { - let err_str = activate_err.to_string(); - let needs_auth = err_str.contains("authentication") - || err_str.contains("401") - || err_str.contains("Unauthorized"); + let needs_auth = matches!( + &activate_err, + crate::extensions::ExtensionError::AuthRequired + ); if !needs_auth { - return Ok(Json(ActionResponse::fail(err_str))); + return Ok(Json(ActionResponse::fail(activate_err.to_string()))); } // Activation failed due to auth; try authenticating first. @@ -2481,6 +2690,7 @@ mod tests { skill_catalog: None, scheduler: None, chat_rate_limiter: RateLimiter::new(30, 60), + oauth_rate_limiter: RateLimiter::new(10, 60), registry_entries: vec![], cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), @@ -2803,4 +3013,177 @@ mod tests { .is_none() ); } + + // --- Slack relay OAuth CSRF tests --- + + fn test_relay_oauth_router(state: Arc) -> Router { + Router::new() + .route( + "/oauth/slack/callback", + get(slack_relay_oauth_callback_handler), + ) + .with_state(state) + } + + fn test_secrets_store() -> Arc { + Arc::new(crate::secrets::InMemorySecretsStore::new(Arc::new( + crate::secrets::SecretsCrypto::new(secrecy::SecretString::from( + "test-key-at-least-32-chars-long!!".to_string(), + )) + .expect("crypto"), + ))) + } + + fn test_ext_mgr( + secrets: Arc, + ) -> Arc { + let tool_registry = Arc::new(ToolRegistry::new()); + let mcp_sm = Arc::new(crate::tools::mcp::session::McpSessionManager::new()); + let mcp_pm = Arc::new(crate::tools::mcp::process::McpProcessManager::new()); + Arc::new(ExtensionManager::new( + mcp_sm, + mcp_pm, + secrets, + tool_registry, + None, + None, + std::path::PathBuf::from("/tmp/wasm_tools"), + std::path::PathBuf::from("/tmp/wasm_channels"), + None, + "test".to_string(), + None, + vec![], + )) + } + + #[tokio::test] + async fn test_relay_oauth_callback_missing_state_param() { + use axum::body::Body; + use tower::ServiceExt; + + let secrets = test_secrets_store(); + let ext_mgr = test_ext_mgr(secrets); + let state = test_gateway_state(Some(ext_mgr)); + let app = test_relay_oauth_router(state); + + // Callback without state param should be rejected + let req = axum::http::Request::builder() + .uri("/oauth/slack/callback?stream_token=tok123&team_id=T123&provider=slack") + .body(Body::empty()) + .expect("request"); + + let resp = ServiceExt::>::oneshot(app, req) + .await + .expect("response"); + + let body = axum::body::to_bytes(resp.into_body(), 1024 * 64) + .await + .expect("body"); + let html = String::from_utf8_lossy(&body); + assert!( + html.contains("Invalid or expired authorization"), + "Expected CSRF error, got: {}", + &html[..html.len().min(300)] + ); + } + + #[tokio::test] + async fn test_relay_oauth_callback_wrong_state_param() { + use axum::body::Body; + use tower::ServiceExt; + + let secrets = test_secrets_store(); + + // Store a valid nonce + secrets + .create( + "test", + crate::secrets::CreateSecretParams::new( + format!("relay:{}:oauth_state", DEFAULT_RELAY_NAME), + "correct-nonce-value", + ), + ) + .await + .expect("store nonce"); + + let ext_mgr = test_ext_mgr(secrets); + let state = test_gateway_state(Some(ext_mgr)); + let app = test_relay_oauth_router(state); + + // Callback with wrong state param + let req = axum::http::Request::builder() + .uri("/oauth/slack/callback?stream_token=tok123&team_id=T123&provider=slack&state=wrong-nonce") + .body(Body::empty()) + .expect("request"); + + let resp = ServiceExt::>::oneshot(app, req) + .await + .expect("response"); + + let body = axum::body::to_bytes(resp.into_body(), 1024 * 64) + .await + .expect("body"); + let html = String::from_utf8_lossy(&body); + assert!( + html.contains("Invalid or expired authorization"), + "Expected CSRF error for wrong nonce, got: {}", + &html[..html.len().min(300)] + ); + } + + #[tokio::test] + async fn test_relay_oauth_callback_correct_state_proceeds() { + use axum::body::Body; + use tower::ServiceExt; + + let secrets = test_secrets_store(); + let nonce = "valid-test-nonce-12345"; + + // Store the correct nonce + secrets + .create( + "test", + crate::secrets::CreateSecretParams::new( + format!("relay:{}:oauth_state", DEFAULT_RELAY_NAME), + nonce, + ), + ) + .await + .expect("store nonce"); + + let ext_mgr = test_ext_mgr(secrets.clone()); + let state = test_gateway_state(Some(ext_mgr)); + let app = test_relay_oauth_router(state); + + // Callback with correct state param — will pass CSRF check + // but may fail downstream (no real relay service) — that's OK, + // we just verify it doesn't return a CSRF error. + let req = axum::http::Request::builder() + .uri(format!( + "/oauth/slack/callback?stream_token=tok123&team_id=T123&provider=slack&state={}", + nonce + )) + .body(Body::empty()) + .expect("request"); + + let resp = ServiceExt::>::oneshot(app, req) + .await + .expect("response"); + + let body = axum::body::to_bytes(resp.into_body(), 1024 * 64) + .await + .expect("body"); + let html = String::from_utf8_lossy(&body); + // Should NOT contain the CSRF error message + assert!( + !html.contains("Invalid or expired authorization"), + "Should have passed CSRF check, got: {}", + &html[..html.len().min(300)] + ); + + // Verify the nonce was consumed (deleted) + let state_key = format!("relay:{}:oauth_state", DEFAULT_RELAY_NAME); + let exists = secrets.exists("test", &state_key).await.unwrap_or(true); + assert!(!exists, "CSRF nonce should be deleted after use"); + } } diff --git a/src/channels/web/static/app.js b/src/channels/web/static/app.js index 3090c515..c64f491e 100644 --- a/src/channels/web/static/app.js +++ b/src/channels/web/static/app.js @@ -2350,8 +2350,8 @@ function renderExtensionCard(ext) { activeLabel.textContent = ext.active ? 'Active' : 'Installed'; actions.appendChild(activeLabel); - // MCP servers may be installed but inactive — show Activate button - if (ext.kind === 'mcp_server' && !ext.active) { + // MCP servers and channel-relay extensions may be installed but inactive — show Activate button + if ((ext.kind === 'mcp_server' || ext.kind === 'channel_relay') && !ext.active) { const activateBtn = document.createElement('button'); activateBtn.className = 'btn-ext activate'; activateBtn.textContent = 'Activate'; diff --git a/src/channels/web/test_helpers.rs b/src/channels/web/test_helpers.rs index 053dd84e..981eacdd 100644 --- a/src/channels/web/test_helpers.rs +++ b/src/channels/web/test_helpers.rs @@ -82,6 +82,7 @@ impl TestGatewayBuilder { skill_catalog: None, scheduler: None, chat_rate_limiter: RateLimiter::new(30, 60), + oauth_rate_limiter: RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/src/channels/web/ws.rs b/src/channels/web/ws.rs index 1736ae7e..d24529ef 100644 --- a/src/channels/web/ws.rs +++ b/src/channels/web/ws.rs @@ -509,6 +509,7 @@ mod tests { skill_registry: None, skill_catalog: None, chat_rate_limiter: crate::channels::web::server::RateLimiter::new(30, 60), + oauth_rate_limiter: crate::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/src/config/mod.rs b/src/config/mod.rs index 77b05a13..1bdd446e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -14,6 +14,7 @@ mod heartbeat; pub(crate) mod helpers; mod hygiene; pub(crate) mod llm; +pub mod relay; mod routines; mod safety; mod sandbox; @@ -38,6 +39,7 @@ pub use self::embeddings::EmbeddingsConfig; pub use self::heartbeat::HeartbeatConfig; pub use self::hygiene::HygieneConfig; pub use self::llm::default_session_path; +pub use self::relay::RelayConfig; pub use self::routines::RoutineConfig; pub use self::safety::SafetyConfig; pub use self::sandbox::{ClaudeCodeConfig, SandboxModeConfig}; @@ -85,6 +87,9 @@ pub struct Config { pub skills: SkillsConfig, pub transcription: TranscriptionConfig, pub observability: crate::observability::ObservabilityConfig, + /// Channel-relay integration (Slack via external relay service). + /// Present only when both `CHANNEL_RELAY_URL` and `CHANNEL_RELAY_API_KEY` are set. + pub relay: Option, } impl Config { @@ -157,6 +162,7 @@ impl Config { }, transcription: TranscriptionConfig::default(), observability: crate::observability::ObservabilityConfig::default(), + relay: None, } } @@ -310,6 +316,7 @@ impl Config { observability: crate::observability::ObservabilityConfig { backend: std::env::var("OBSERVABILITY_BACKEND").unwrap_or_else(|_| "none".into()), }, + relay: RelayConfig::from_env(), }) } } diff --git a/src/config/relay.rs b/src/config/relay.rs new file mode 100644 index 00000000..d45de188 --- /dev/null +++ b/src/config/relay.rs @@ -0,0 +1,157 @@ +//! Channel-relay service configuration. + +use secrecy::SecretString; + +/// Configuration for connecting to a channel-relay service. +#[derive(Clone)] +pub struct RelayConfig { + /// Base URL of the channel-relay service (e.g., `http://localhost:3001`). + pub url: String, + /// API key for authenticated channel-relay endpoints. + pub api_key: SecretString, + /// Override for the OAuth callback URL (e.g., a tunnel URL). + pub callback_url: Option, + /// Override for the instance identifier. + pub instance_id: Option, + /// HTTP request timeout in seconds (default: 30). + pub request_timeout_secs: u64, + /// SSE stream long-poll timeout in seconds (default: 86400 = 24 h). + pub stream_timeout_secs: u64, + /// Initial exponential backoff in milliseconds (default: 1000). + pub backoff_initial_ms: u64, + /// Maximum exponential backoff in milliseconds (default: 60000). + pub backoff_max_ms: u64, +} + +impl std::fmt::Debug for RelayConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RelayConfig") + .field("url", &self.url) + .field("api_key", &"[REDACTED]") + .field("callback_url", &self.callback_url) + .field("instance_id", &self.instance_id) + .field("request_timeout_secs", &self.request_timeout_secs) + .field("stream_timeout_secs", &self.stream_timeout_secs) + .field("backoff_initial_ms", &self.backoff_initial_ms) + .field("backoff_max_ms", &self.backoff_max_ms) + .finish() + } +} + +impl RelayConfig { + /// Load relay config from environment variables. + /// + /// Returns `None` if either `CHANNEL_RELAY_URL` or `CHANNEL_RELAY_API_KEY` + /// is not set, making the relay integration opt-in. + pub fn from_env() -> Option { + Self::from_env_reader(|key| std::env::var(key).ok()) + } + + /// Build a config for tests without touching the process environment. + pub fn from_values(url: impl Into, api_key: impl Into) -> Self { + Self { + url: url.into(), + api_key: SecretString::from(api_key.into()), + callback_url: None, + instance_id: None, + request_timeout_secs: 30, + stream_timeout_secs: 86400, + backoff_initial_ms: 1000, + backoff_max_ms: 60000, + } + } + + /// Internal constructor that reads values through a closure, enabling safe testing. + fn from_env_reader(env: impl Fn(&str) -> Option) -> Option { + let url = env("CHANNEL_RELAY_URL")?; + let api_key = SecretString::from(env("CHANNEL_RELAY_API_KEY")?); + Some(Self { + url, + api_key, + callback_url: env("IRONCLAW_OAUTH_CALLBACK_URL"), + instance_id: env("IRONCLAW_INSTANCE_ID"), + request_timeout_secs: env("RELAY_REQUEST_TIMEOUT_SECS") + .and_then(|v| v.parse().ok()) + .unwrap_or(30), + stream_timeout_secs: env("RELAY_STREAM_TIMEOUT_SECS") + .and_then(|v| v.parse().ok()) + .unwrap_or(86400), + backoff_initial_ms: env("RELAY_BACKOFF_INITIAL_MS") + .and_then(|v| v.parse().ok()) + .unwrap_or(1000), + backoff_max_ms: env("RELAY_BACKOFF_MAX_MS") + .and_then(|v| v.parse().ok()) + .unwrap_or(60000), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn from_env_reader_returns_none_when_unset() { + let config = RelayConfig::from_env_reader(|_| None); + assert!(config.is_none()); + } + + #[test] + fn from_env_reader_loads_defaults() { + let config = RelayConfig::from_env_reader(|key| match key { + "CHANNEL_RELAY_URL" => Some("http://localhost:3001".into()), + "CHANNEL_RELAY_API_KEY" => Some("test-key".into()), + _ => None, + }) + .expect("config should be Some"); + + assert_eq!(config.url, "http://localhost:3001"); + assert_eq!(config.request_timeout_secs, 30); + assert_eq!(config.stream_timeout_secs, 86400); + assert_eq!(config.backoff_initial_ms, 1000); + assert_eq!(config.backoff_max_ms, 60000); + assert!(config.callback_url.is_none()); + assert!(config.instance_id.is_none()); + } + + #[test] + fn from_env_reader_loads_overrides() { + let config = RelayConfig::from_env_reader(|key| match key { + "CHANNEL_RELAY_URL" => Some("http://relay:3001".into()), + "CHANNEL_RELAY_API_KEY" => Some("secret".into()), + "IRONCLAW_OAUTH_CALLBACK_URL" => Some("https://tunnel.example.com".into()), + "IRONCLAW_INSTANCE_ID" => Some("my-instance".into()), + "RELAY_REQUEST_TIMEOUT_SECS" => Some("60".into()), + "RELAY_STREAM_TIMEOUT_SECS" => Some("43200".into()), + "RELAY_BACKOFF_INITIAL_MS" => Some("2000".into()), + "RELAY_BACKOFF_MAX_MS" => Some("120000".into()), + _ => None, + }) + .expect("config should be Some"); + + assert_eq!( + config.callback_url.as_deref(), + Some("https://tunnel.example.com") + ); + assert_eq!(config.instance_id.as_deref(), Some("my-instance")); + assert_eq!(config.request_timeout_secs, 60); + assert_eq!(config.stream_timeout_secs, 43200); + assert_eq!(config.backoff_initial_ms, 2000); + assert_eq!(config.backoff_max_ms, 120000); + } + + #[test] + fn from_values_builds_with_defaults() { + let config = RelayConfig::from_values("http://localhost:3001", "key"); + assert_eq!(config.url, "http://localhost:3001"); + assert_eq!(config.request_timeout_secs, 30); + } + + #[test] + fn debug_redacts_api_key() { + let config = RelayConfig::from_values("http://localhost:3001", "super-secret"); + let debug = format!("{:?}", config); + assert!(debug.contains("[REDACTED]")); + assert!(!debug.contains("super-secret")); + } +} diff --git a/src/extensions/discovery.rs b/src/extensions/discovery.rs index b58101bc..64cdf104 100644 --- a/src/extensions/discovery.rs +++ b/src/extensions/discovery.rs @@ -250,6 +250,7 @@ fn extract_source(source: &ExtensionSource) -> String { ExtensionSource::Discovered { url } => url.clone(), ExtensionSource::WasmDownload { wasm_url, .. } => wasm_url.clone(), ExtensionSource::WasmBuildable { source_dir, .. } => source_dir.clone(), + ExtensionSource::ChannelRelay { relay_url } => relay_url.clone(), } } diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 85d1ce74..8b9747fb 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -84,6 +84,8 @@ pub struct ExtensionManager { // WASM channel hot-activation infrastructure (set post-construction) channel_runtime: RwLock>, + /// Channel manager for hot-adding relay channels (set independently of WASM runtime). + relay_channel_manager: RwLock>>, // Shared secrets: Arc, @@ -97,6 +99,8 @@ pub struct ExtensionManager { store: Option>, /// Names of WASM channels that were successfully loaded at startup. active_channel_names: RwLock>, + /// Installed channel-relay extensions (no on-disk artifact, tracked in memory). + installed_relay_extensions: RwLock>, /// Last activation error for each WASM channel (ephemeral, cleared on success). activation_errors: RwLock>, /// SSE broadcast sender (set post-construction via `set_sse_sender()`). @@ -111,6 +115,9 @@ pub struct ExtensionManager { /// Gateway auth token for authenticating with the platform token exchange proxy. /// Read once at construction from `GATEWAY_AUTH_TOKEN` env var. gateway_token: Option, + /// Relay config captured at startup. Used by `auth_channel_relay` and + /// `activate_channel_relay` instead of re-reading env vars. + relay_config: Option, } /// Sanitize a URL for logging by removing query parameters and credentials. @@ -169,6 +176,7 @@ impl ExtensionManager { wasm_tools_dir, wasm_channels_dir, channel_runtime: RwLock::new(None), + relay_channel_manager: RwLock::new(None), secrets, tool_registry, hooks, @@ -177,13 +185,24 @@ impl ExtensionManager { user_id, store, active_channel_names: RwLock::new(HashSet::new()), + installed_relay_extensions: RwLock::new(HashSet::new()), activation_errors: RwLock::new(HashMap::new()), sse_sender: RwLock::new(None), pending_oauth_flows: crate::cli::oauth_defaults::new_pending_oauth_registry(), gateway_token: std::env::var("GATEWAY_AUTH_TOKEN").ok(), + relay_config: crate::config::RelayConfig::from_env(), } } + /// Get the relay config stored at startup. + fn relay_config(&self) -> Result<&crate::config::RelayConfig, ExtensionError> { + self.relay_config.as_ref().ok_or_else(|| { + ExtensionError::Config( + "CHANNEL_RELAY_URL and CHANNEL_RELAY_API_KEY must be set".to_string(), + ) + }) + } + /// Configure the channel runtime infrastructure for hot-activating WASM channels. /// /// Call after construction (and after wrapping in `Arc`) once the channel @@ -197,6 +216,8 @@ impl ExtensionManager { wasm_channel_router: Arc, wasm_channel_owner_ids: std::collections::HashMap, ) { + // Also store the channel manager for relay channel activation. + *self.relay_channel_manager.write().await = Some(Arc::clone(&channel_manager)); *self.channel_runtime.write().await = Some(ChannelRuntimeState { channel_manager, wasm_channel_runtime, @@ -206,6 +227,58 @@ impl ExtensionManager { }); } + /// Set just the channel manager for relay channel hot-activation. + /// + /// Call this when WASM channel runtime is not available but relay channels + /// still need to be hot-added. + pub async fn set_relay_channel_manager(&self, channel_manager: Arc) { + *self.relay_channel_manager.write().await = Some(channel_manager); + } + + /// Check if a channel name corresponds to a relay extension (has stored stream token). + pub async fn is_relay_channel(&self, name: &str) -> bool { + self.secrets + .exists(&self.user_id, &format!("relay:{}:stream_token", name)) + .await + .unwrap_or(false) + } + + /// Restore persisted relay channels after startup. + /// + /// Loads the persisted active channel list, filters to relay types (those with + /// a stored stream token), and activates each via `activate_stored_relay()`. + /// Skips channels that are already active. Call this after `set_relay_channel_manager()`. + pub async fn restore_relay_channels(&self) { + let persisted = self.load_persisted_active_channels().await; + let already_active = self.active_channel_names.read().await.clone(); + + for name in &persisted { + if already_active.contains(name) { + continue; + } + if !self.is_relay_channel(name).await { + continue; + } + match self.activate_stored_relay(name).await { + Ok(_) => { + tracing::debug!(channel = %name, "Restored persisted relay channel"); + } + Err(e) => { + tracing::warn!( + channel = %name, + error = %e, + "Failed to restore persisted relay channel" + ); + } + } + } + } + + /// Access the secrets store (used by OAuth callback handlers). + pub fn secrets(&self) -> &Arc { + &self.secrets + } + /// Register channel names that were loaded at startup. /// Called after WASM channels are loaded so `list()` reports accurate active status. pub async fn set_active_channels(&self, names: Vec) { @@ -345,6 +418,12 @@ impl ExtensionManager { ExtensionKind::WasmChannel => { self.install_wasm_channel_from_url(name, url, None).await } + ExtensionKind::ChannelRelay => { + // ChannelRelay extensions are installed from registry, not by URL + Err(ExtensionError::InstallFailed( + "Channel relay extensions cannot be installed by URL".to_string(), + )) + } } .map_err(|e| { let sanitized = sanitize_url_for_logging(url); @@ -377,6 +456,7 @@ impl ExtensionManager { ExtensionKind::McpServer => self.auth_mcp(name, token).await, ExtensionKind::WasmTool => self.auth_wasm_tool(name, token).await, ExtensionKind::WasmChannel => self.auth_wasm_channel(name, token).await, + ExtensionKind::ChannelRelay => self.auth_channel_relay(name, token).await, } } @@ -389,6 +469,7 @@ impl ExtensionManager { ExtensionKind::McpServer => self.activate_mcp(name).await, ExtensionKind::WasmTool => self.activate_wasm_tool(name).await, ExtensionKind::WasmChannel => self.activate_wasm_channel(name).await, + ExtensionKind::ChannelRelay => self.activate_channel_relay(name).await, } } @@ -560,6 +641,41 @@ impl ExtensionManager { } } + // List channel-relay extensions + if kind_filter.is_none() || kind_filter == Some(ExtensionKind::ChannelRelay) { + let installed = self.installed_relay_extensions.read().await; + let active_names = self.active_channel_names.read().await; + for name in installed.iter() { + let active = active_names.contains(name); + let has_token = self + .secrets + .exists(&self.user_id, &format!("relay:{}:stream_token", name)) + .await + .unwrap_or(false); + let registry_entry = self + .registry + .get_with_kind(name, Some(ExtensionKind::ChannelRelay)) + .await; + let display_name = registry_entry.as_ref().map(|e| e.display_name.clone()); + let description = registry_entry.as_ref().map(|e| e.description.clone()); + extensions.push(InstalledExtension { + name: name.clone(), + kind: ExtensionKind::ChannelRelay, + display_name, + description, + url: None, + authenticated: has_token, + active, + tools: Vec::new(), + needs_setup: false, + has_auth: true, + installed: true, + activation_error: None, + version: None, + }); + } + } + // Append available-but-not-installed registry entries if include_available { let installed_names: std::collections::HashSet<(String, ExtensionKind)> = extensions @@ -698,6 +814,37 @@ impl ExtensionManager { name )) } + ExtensionKind::ChannelRelay => { + // Remove from installed set + self.installed_relay_extensions.write().await.remove(name); + + // Remove from active channels + self.active_channel_names.write().await.remove(name); + self.persist_active_channels().await; + + // Remove stored stream token + let _ = self + .secrets + .delete(&self.user_id, &format!("relay:{}:stream_token", name)) + .await; + + // Shut down the channel (check both runtime paths for WASM+relay and relay-only modes) + let mut shut_down = false; + if let Some(ref rt) = *self.channel_runtime.read().await + && let Some(channel) = rt.channel_manager.get_channel(name).await + { + let _ = channel.shutdown().await; + shut_down = true; + } + if !shut_down + && let Some(ref cm) = *self.relay_channel_manager.read().await + && let Some(channel) = cm.get_channel(name).await + { + let _ = channel.shutdown().await; + } + + Ok(format!("Removed channel relay '{}'", name)) + } } } @@ -785,12 +932,12 @@ impl ExtensionManager { &self.wasm_channels_dir, crate::tools::wasm::WIT_CHANNEL_VERSION, ), - ExtensionKind::McpServer => { + ExtensionKind::McpServer | ExtensionKind::ChannelRelay => { return UpgradeOutcome { name: name.to_string(), kind, status: "failed".to_string(), - detail: "MCP servers cannot be upgraded this way".to_string(), + detail: "This extension type cannot be upgraded this way".to_string(), }; } }; @@ -811,7 +958,7 @@ impl ExtensionManager { .ok() .and_then(|c| c.wit_version) } - ExtensionKind::McpServer => None, + ExtensionKind::McpServer | ExtensionKind::ChannelRelay => None, }; wit } @@ -971,6 +1118,14 @@ impl ExtensionManager { }); Ok(info) } + ExtensionKind::ChannelRelay => { + let info = serde_json::json!({ + "name": name, + "kind": "channel_relay", + "active": self.active_channel_names.read().await.contains(name), + }); + Ok(info) + } } } @@ -1135,6 +1290,21 @@ impl ExtensionManager { "WASM channel entry has no download URL or build info".to_string(), )), }, + ExtensionKind::ChannelRelay => { + // No download needed — just mark as installed. + self.installed_relay_extensions + .write() + .await + .insert(entry.name.clone()); + Ok(InstallResult { + name: entry.name.clone(), + kind: ExtensionKind::ChannelRelay, + message: format!( + "'{}' installed. Click Activate to connect your workspace.", + entry.display_name + ), + }) + } } } @@ -1494,6 +1664,7 @@ impl ExtensionManager { ExtensionKind::WasmTool => "WASM tool", ExtensionKind::WasmChannel => "WASM channel", ExtensionKind::McpServer => "MCP server", + ExtensionKind::ChannelRelay => "channel relay", }; tracing::info!( @@ -3033,7 +3204,192 @@ impl ExtensionManager { }) } + // ── Channel-relay extension methods ────────────────────────────────── + + /// Derive a stable instance ID from the relay config and user_id. + fn relay_instance_id(&self, config: &crate::config::RelayConfig) -> String { + config.instance_id.clone().unwrap_or_else(|| { + uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_DNS, self.user_id.as_bytes()).to_string() + }) + } + + /// Authenticate a channel-relay extension. + /// + /// For Slack: initiates OAuth flow (redirect-based). + /// For Telegram: accepts a bot token, registers it with channel-relay, + /// and stores the returned stream token. + async fn auth_channel_relay( + &self, + name: &str, + _token: Option<&str>, + ) -> Result { + // Check if already authenticated (stream token exists) + let token_key = format!("relay:{}:stream_token", name); + if self + .secrets + .exists(&self.user_id, &token_key) + .await + .unwrap_or(false) + { + return Ok(AuthResult::authenticated(name, ExtensionKind::ChannelRelay)); + } + + // Use relay config captured at startup + let relay_config = self.relay_config()?; + + let instance_id = self.relay_instance_id(relay_config); + let user_id_uuid = std::env::var("IRONCLAW_USER_ID").unwrap_or_else(|_| { + uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_DNS, self.user_id.as_bytes()).to_string() + }); + + let client = crate::channels::relay::RelayClient::new( + relay_config.url.clone(), + relay_config.api_key.clone(), + relay_config.request_timeout_secs, + ) + .map_err(|e| ExtensionError::Config(e.to_string()))?; + + // OAuth redirect flow + let callback_base = self + .tunnel_url + .clone() + .or_else(|| relay_config.callback_url.clone()) + .unwrap_or_else(|| { + let host = std::env::var("GATEWAY_HOST").unwrap_or_else(|_| "127.0.0.1".into()); + let port = std::env::var("GATEWAY_PORT").unwrap_or_else(|_| "3001".into()); + format!("http://{}:{}", host, port) + }); + + // Generate CSRF nonce for OAuth state parameter + let state_nonce = uuid::Uuid::new_v4().to_string(); + let state_key = format!("relay:{}:oauth_state", name); + // Delete any stale nonce before storing the new one + let _ = self.secrets.delete(&self.user_id, &state_key).await; + self.secrets + .create( + &self.user_id, + CreateSecretParams::new(&state_key, &state_nonce), + ) + .await + .map_err(|e| ExtensionError::AuthFailed(format!("Failed to store OAuth state: {e}")))?; + + let callback_url = format!( + "{}/oauth/slack/callback?state={}", + callback_base, state_nonce + ); + + match client + .initiate_oauth(&instance_id, &user_id_uuid, &callback_url) + .await + { + Ok(auth_url) => Ok(AuthResult::awaiting_authorization( + name, + ExtensionKind::ChannelRelay, + auth_url, + "redirect".to_string(), + )), + Err(e) => Err(ExtensionError::AuthFailed(e.to_string())), + } + } + + /// Activate a channel-relay extension. + async fn activate_channel_relay(&self, name: &str) -> Result { + let token_key = format!("relay:{}:stream_token", name); + let team_id_key = format!("relay:{}:team_id", name); + + // Check if we have a stream token + let stream_token = match self.secrets.get_decrypted(&self.user_id, &token_key).await { + Ok(secret) => secret.expose().to_string(), + Err(_) => { + return Err(ExtensionError::AuthRequired); + } + }; + + // Get team_id from settings + let team_id = if let Some(ref store) = self.store { + store + .get_setting(&self.user_id, &team_id_key) + .await + .ok() + .flatten() + .and_then(|v| v.as_str().map(|s| s.to_string())) + .unwrap_or_default() + } else { + String::new() + }; + + // Use relay config captured at startup + let relay_config = self.relay_config()?; + + let instance_id = self.relay_instance_id(relay_config); + + let client = crate::channels::relay::RelayClient::new( + relay_config.url.clone(), + relay_config.api_key.clone(), + relay_config.request_timeout_secs, + ) + .map_err(|e| ExtensionError::ActivationFailed(e.to_string()))?; + + let channel = crate::channels::relay::RelayChannel::new_with_provider( + client, + crate::channels::relay::channel::RelayProvider::Slack, + stream_token, + team_id, + instance_id, + self.user_id.clone(), + ) + .with_timeouts( + relay_config.stream_timeout_secs, + relay_config.backoff_initial_ms, + relay_config.backoff_max_ms, + ); + + // Hot-add to channel manager + let cm_guard = self.relay_channel_manager.read().await; + let channel_mgr = cm_guard.as_ref().ok_or_else(|| { + ExtensionError::ActivationFailed("Channel manager not initialized".to_string()) + })?; + + channel_mgr + .hot_add(Box::new(channel)) + .await + .map_err(|e| ExtensionError::ActivationFailed(e.to_string()))?; + + // Mark as active + self.active_channel_names + .write() + .await + .insert(name.to_string()); + self.persist_active_channels().await; + + // Broadcast status + let status_msg = "Slack connected via channel relay".to_string(); + self.broadcast_extension_status(name, "active", Some(&status_msg)) + .await; + + Ok(ActivateResult { + name: name.to_string(), + kind: ExtensionKind::ChannelRelay, + tools_loaded: Vec::new(), + message: status_msg, + }) + } + + /// Activate a channel-relay extension from stored credentials (for startup reconnect). + pub async fn activate_stored_relay(&self, name: &str) -> Result<(), ExtensionError> { + self.installed_relay_extensions + .write() + .await + .insert(name.to_string()); + self.activate_channel_relay(name).await?; + Ok(()) + } + /// Determine what kind of installed extension this is. + /// + /// This is a read-only check — it never modifies `installed_relay_extensions`. + /// To mark a relay extension as installed, use `activate_stored_relay()` or + /// the explicit install flow. async fn determine_installed_kind(&self, name: &str) -> Result { // Check MCP servers first if self.get_mcp_server(name).await.is_ok() { @@ -3052,8 +3408,22 @@ impl ExtensionManager { return Ok(ExtensionKind::WasmChannel); } + // Check channel-relay extensions (installed in memory or has stored token) + if self.installed_relay_extensions.read().await.contains(name) { + return Ok(ExtensionKind::ChannelRelay); + } + // Also check if there's a stored stream token (persisted across restarts) + if self + .secrets + .exists(&self.user_id, &format!("relay:{}:stream_token", name)) + .await + .unwrap_or(false) + { + return Ok(ExtensionKind::ChannelRelay); + } + Err(ExtensionError::NotInstalled(format!( - "'{}' is not installed as an MCP server, WASM tool, or WASM channel", + "'{}' is not installed as an MCP server, WASM tool, WASM channel, or channel relay", name ))) } @@ -4136,6 +4506,95 @@ mod tests { unsafe { std::env::remove_var("ICTEST6_TOKEN") }; } + #[tokio::test] + async fn test_determine_installed_kind_does_not_auto_install_relay() { + // Regression: determine_installed_kind used to auto-insert into + // installed_relay_extensions when a ChannelRelay registry entry existed, + // even though the user never installed it. It should be read-only. + let dir = tempfile::tempdir().expect("temp dir"); + let mgr = make_test_manager(None, dir.path().to_path_buf()); + + // The manager has no relay extensions installed + assert!( + mgr.installed_relay_extensions.read().await.is_empty(), + "Should start with no installed relay extensions" + ); + + // Calling determine_installed_kind for a non-installed name returns NotInstalled + let result = mgr.determine_installed_kind("slack-relay").await; + assert!(result.is_err(), "Should return NotInstalled"); + + // Crucially: installed_relay_extensions must still be empty + assert!( + mgr.installed_relay_extensions.read().await.is_empty(), + "determine_installed_kind must not modify installed_relay_extensions" + ); + } + + #[tokio::test] + async fn test_is_relay_channel_detects_stored_token() { + let dir = tempfile::tempdir().expect("temp dir"); + let mgr = make_test_manager(None, dir.path().to_path_buf()); + + // No token stored → not a relay channel + assert!(!mgr.is_relay_channel("slack-relay").await); + + // Store a stream token + mgr.secrets + .create( + "test", + crate::secrets::CreateSecretParams::new("relay:slack-relay:stream_token", "tok123"), + ) + .await + .expect("store token"); + + // Now it's detected as a relay channel + assert!(mgr.is_relay_channel("slack-relay").await); + } + + #[tokio::test] + async fn test_remove_relay_shuts_down_via_relay_channel_manager() { + // Regression: remove() only checked channel_runtime for shutdown, missing + // relay-only mode where only relay_channel_manager is set. + let dir = tempfile::tempdir().expect("temp dir"); + let mgr = make_test_manager(None, dir.path().to_path_buf()); + + // Set up relay channel manager with a stub channel + let cm = Arc::new(crate::channels::ChannelManager::new()); + let (stub, _tx) = crate::testing::StubChannel::new("slack-relay"); + cm.add(Box::new(stub)).await; + mgr.set_relay_channel_manager(Arc::clone(&cm)).await; + + // Mark as installed + store a token so determine_installed_kind finds it + mgr.installed_relay_extensions + .write() + .await + .insert("slack-relay".to_string()); + mgr.secrets + .create( + "test", + crate::secrets::CreateSecretParams::new("relay:slack-relay:stream_token", "tok123"), + ) + .await + .expect("store token"); + + // Verify channel exists before removal + assert!(cm.get_channel("slack-relay").await.is_some()); + + // Remove should succeed and shut down the channel + let result = mgr.remove("slack-relay").await; + assert!(result.is_ok(), "remove should succeed: {:?}", result.err()); + + // installed_relay_extensions should be cleared + assert!( + !mgr.installed_relay_extensions + .read() + .await + .contains("slack-relay"), + "Should be removed from installed set" + ); + } + #[test] fn test_sanitize_url_with_query_params() { let url = "https://api.example.com/path?api_key=secret123&token=abc"; diff --git a/src/extensions/mod.rs b/src/extensions/mod.rs index 011d9571..2c591e0c 100644 --- a/src/extensions/mod.rs +++ b/src/extensions/mod.rs @@ -37,6 +37,8 @@ pub enum ExtensionKind { WasmTool, /// WASM channel module with hot-activation support. WasmChannel, + /// External channel via channel-relay service (Slack, etc.). + ChannelRelay, } impl std::fmt::Display for ExtensionKind { @@ -45,6 +47,7 @@ impl std::fmt::Display for ExtensionKind { ExtensionKind::McpServer => write!(f, "mcp_server"), ExtensionKind::WasmTool => write!(f, "wasm_tool"), ExtensionKind::WasmChannel => write!(f, "wasm_channel"), + ExtensionKind::ChannelRelay => write!(f, "channel_relay"), } } } @@ -99,6 +102,8 @@ pub enum ExtensionSource { }, /// Discovered online (not yet validated for a specific source type). Discovered { url: String }, + /// External channel via channel-relay service. + ChannelRelay { relay_url: String }, } /// Hint about what authentication method is needed. @@ -116,6 +121,8 @@ pub enum AuthHint { CapabilitiesAuth, /// No authentication needed. None, + /// OAuth via channel-relay service. + ChannelRelayOAuth, } /// Where a search result came from. @@ -499,6 +506,9 @@ pub enum ExtensionError { #[error("Activation failed: {0}")] ActivationFailed(String), + #[error("Authentication required")] + AuthRequired, + #[error("Installation failed: {0}")] InstallFailed(String), @@ -976,6 +986,7 @@ mod tests { ExtensionError::Config("missing key".into()), "Config error: missing key", ), + (ExtensionError::AuthRequired, "Authentication required"), ( ExtensionError::Other("something broke".into()), "something broke", diff --git a/src/extensions/registry.rs b/src/extensions/registry.rs index 32dd4c2b..35a45862 100644 --- a/src/extensions/registry.rs +++ b/src/extensions/registry.rs @@ -224,8 +224,16 @@ fn score_entry(entry: &RegistryEntry, tokens: &[String]) -> u32 { } /// Well-known extensions that ship with ironclaw. -fn builtin_entries() -> Vec { - vec![ +/// +/// If `relay_url` is provided, a channel-relay Slack entry is included in the list. +/// Pass `None` when the relay is not configured. +pub fn builtin_entries() -> Vec { + builtin_entries_with_relay(std::env::var("CHANNEL_RELAY_URL").ok()) +} + +/// Well-known extensions, with an optional relay URL for the channel-relay entry. +pub fn builtin_entries_with_relay(relay_url: Option) -> Vec { + let mut entries = vec![ // -- MCP Servers -- RegistryEntry { name: "notion".to_string(), @@ -415,7 +423,29 @@ fn builtin_entries() -> Vec { // WASM channels (telegram, slack, discord, whatsapp) come from the embedded // registry catalog (registry/channels/*.json) with WasmDownload URLs pointing // to GitHub release artifacts. See new_with_catalog() for merging. - ] + ]; + + // Conditionally add channel-relay entries when relay URL is configured + if let Some(relay_url) = relay_url { + entries.push(RegistryEntry { + name: crate::channels::relay::DEFAULT_RELAY_NAME.to_string(), + display_name: "Slack".to_string(), + kind: ExtensionKind::ChannelRelay, + description: "Connect Slack workspace via channel relay".to_string(), + keywords: vec![ + "slack".into(), + "chat".into(), + "messaging".into(), + "relay".into(), + ], + source: ExtensionSource::ChannelRelay { relay_url }, + fallback_source: None, + auth_hint: AuthHint::ChannelRelayOAuth, + version: None, + }); + } + + entries } #[cfg(test)] @@ -935,4 +965,30 @@ mod tests { // The first catalog entry added is the channel. assert_eq!(entry.unwrap().kind, ExtensionKind::WasmChannel); } + + #[test] + fn test_builtin_entries_with_relay_none_excludes_relay() { + let entries = super::builtin_entries_with_relay(None); + assert!( + !entries + .iter() + .any(|e| e.kind == ExtensionKind::ChannelRelay), + "No ChannelRelay entry when relay URL is None" + ); + } + + #[test] + fn test_builtin_entries_with_relay_some_includes_relay() { + let entries = + super::builtin_entries_with_relay(Some("http://relay.example.com".to_string())); + let relay = entries + .iter() + .find(|e| e.kind == ExtensionKind::ChannelRelay); + assert!(relay.is_some(), "ChannelRelay entry should be present"); + if let ExtensionSource::ChannelRelay { relay_url } = &relay.unwrap().source { + assert_eq!(relay_url, "http://relay.example.com"); + } else { + panic!("Expected ChannelRelay source"); + } + } } diff --git a/src/main.rs b/src/main.rs index 581e3500..2490bb11 100644 --- a/src/main.rs +++ b/src/main.rs @@ -564,30 +564,41 @@ async fn async_main() -> anyhow::Result<()> { .await; tracing::debug!("Channel runtime wired into extension manager for hot-activation"); - // Auto-activate channels that were active in a previous session. + // Auto-activate WASM channels that were active in a previous session. + // Relay channels are handled separately below via restore_relay_channels(). let persisted = ext_mgr.load_persisted_active_channels().await; for name in &persisted { - if !active_at_startup.contains(name) { - match ext_mgr.activate(name).await { - Ok(result) => { - tracing::debug!( - channel = %name, - message = %result.message, - "Auto-activated persisted channel" - ); - } - Err(e) => { - tracing::warn!( - channel = %name, - error = %e, - "Failed to auto-activate persisted channel" - ); - } + if active_at_startup.contains(name) || ext_mgr.is_relay_channel(name).await { + continue; + } + match ext_mgr.activate(name).await { + Ok(result) => { + tracing::debug!( + channel = %name, + message = %result.message, + "Auto-activated persisted WASM channel" + ); + } + Err(e) => { + tracing::warn!( + channel = %name, + error = %e, + "Failed to auto-activate persisted WASM channel" + ); } } } } + // Ensure the relay channel manager is always set (even without WASM runtime), + // then restore any persisted relay channels. + if let Some(ref ext_mgr) = components.extension_manager { + ext_mgr + .set_relay_channel_manager(Arc::clone(&channels)) + .await; + ext_mgr.restore_relay_channels().await; + } + // Wire SSE sender into extension manager for broadcasting status events. if let Some(ref ext_mgr) = components.extension_manager && let Some(ref sender) = sse_sender diff --git a/tests/openai_compat_integration.rs b/tests/openai_compat_integration.rs index 501fa1aa..c6dd9a11 100644 --- a/tests/openai_compat_integration.rs +++ b/tests/openai_compat_integration.rs @@ -209,6 +209,7 @@ async fn start_test_server_with_provider( skill_registry: None, skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), + oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), @@ -699,6 +700,7 @@ async fn test_no_llm_provider_returns_503() { skill_registry: None, skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), + oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/tests/relay_integration.rs b/tests/relay_integration.rs new file mode 100644 index 00000000..8479cd67 --- /dev/null +++ b/tests/relay_integration.rs @@ -0,0 +1,323 @@ +//! Integration tests for the channel-relay client and channel. +//! +//! Uses real HTTP servers on random ports (no mock framework). + +use std::convert::Infallible; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use axum::{ + Json, Router, + extract::Query, + http::StatusCode, + response::sse::{Event, KeepAlive, Sse}, + routing::{get, post}, +}; +use futures::stream; +use ironclaw::channels::relay::client::{RelayClient, RelayError}; +use secrecy::SecretString; +use serde::Deserialize; +use tokio::net::TcpListener; + +/// Start an axum server on a random port, returning the base URL. +async fn start_server(app: Router) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{}", addr) +} + +fn test_client(base_url: &str) -> RelayClient { + RelayClient::new( + base_url.to_string(), + SecretString::from("test-api-key".to_string()), + 5, + ) + .expect("client build") +} + +// ── SSE stream mock ───────────────────────────────────────────────────── + +#[tokio::test] +async fn test_sse_stream_receives_events() { + let app = Router::new().route( + "/stream", + get( + |Query(params): Query>| async move { + // Verify token is passed + assert!(params.contains_key("token")); + + let events = vec![ + Ok::<_, Infallible>( + Event::default().event("message").data( + serde_json::json!({ + "event_type": "message", + "provider": "slack", + "provider_scope": "T123", + "channel_id": "C456", + "sender_id": "U789", + "content": "hello world" + }) + .to_string(), + ), + ), + Ok(Event::default().event("message").data( + serde_json::json!({ + "event_type": "direct_message", + "provider": "slack", + "provider_scope": "T123", + "channel_id": "D001", + "sender_id": "U789", + "content": "dm text" + }) + .to_string(), + )), + ]; + + Sse::new(stream::iter(events)).keep_alive(KeepAlive::default()) + }, + ), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + let (mut event_stream, handle) = client.connect_stream("test-token", 30).await.unwrap(); + + use futures::StreamExt; + let first = event_stream.next().await.expect("first event"); + assert_eq!(first.event_type, "message"); + assert_eq!(first.text(), "hello world"); + assert_eq!(first.team_id(), "T123"); + + let second = event_stream.next().await.expect("second event"); + assert_eq!(second.event_type, "direct_message"); + assert_eq!(second.text(), "dm text"); + + handle.abort(); +} + +// ── Token renewal flow ────────────────────────────────────────────────── + +#[tokio::test] +async fn test_token_expired_returns_error() { + let app = Router::new().route("/stream", get(|| async { StatusCode::UNAUTHORIZED })); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + match client.connect_stream("expired-token", 30).await { + Err(RelayError::TokenExpired) => {} // expected + Err(other) => panic!("expected TokenExpired, got: {other}"), + Ok(_) => panic!("expected error, got Ok"), + } +} + +#[tokio::test] +async fn test_token_renewal() { + let call_count = std::sync::Arc::new(AtomicUsize::new(0)); + let call_count_clone = call_count.clone(); + + let app = Router::new().route( + "/stream/renew", + post(move |Json(body): Json| { + let count = call_count_clone.clone(); + async move { + count.fetch_add(1, Ordering::SeqCst); + assert!(body.get("instance_id").is_some()); + assert!(body.get("user_id").is_some()); + Json(serde_json::json!({ + "stream_token": "renewed-token-123" + })) + } + }), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + let new_token = client.renew_token("inst-1", "user-1").await.unwrap(); + assert_eq!(new_token, "renewed-token-123"); + assert_eq!(call_count.load(Ordering::SeqCst), 1); +} + +// ── Proxy call ────────────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct ProxyQuery { + team_id: String, +} + +#[tokio::test] +async fn test_proxy_provider_sends_correct_payload() { + let app = Router::new().route( + "/proxy/slack/chat.postMessage", + post( + |Query(q): Query, Json(body): Json| async move { + assert_eq!(q.team_id, "T123"); + assert_eq!(body["channel"], "C456"); + assert_eq!(body["text"], "Hello from test"); + Json(serde_json::json!({"ok": true})) + }, + ), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + let body = serde_json::json!({ + "channel": "C456", + "text": "Hello from test", + }); + let resp = client + .proxy_provider("slack", "T123", "chat.postMessage", body, None) + .await + .unwrap(); + assert_eq!(resp["ok"], true); +} + +// ── List connections ──────────────────────────────────────────────────── + +#[tokio::test] +async fn test_list_connections() { + let app = Router::new().route( + "/connections", + get(|| async { + Json(serde_json::json!([ + {"provider": "slack", "team_id": "T123", "team_name": "Test Team", "connected": true}, + {"provider": "slack", "team_id": "T456", "team_name": "Other", "connected": false}, + ])) + }), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + let conns = client.list_connections("inst-1").await.unwrap(); + assert_eq!(conns.len(), 2); + assert!(conns[0].connected); + assert!(!conns[1].connected); +} + +// ── API key header ────────────────────────────────────────────────────── + +#[tokio::test] +async fn test_api_key_sent_in_header() { + let app = Router::new().route( + "/connections", + get(|headers: axum::http::HeaderMap| async move { + let key = headers + .get("X-API-Key") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + assert_eq!(key, "test-api-key"); + Json(serde_json::json!([])) + }), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + let _ = client.list_connections("inst-1").await.unwrap(); +} + +// ── Client builder error propagation ──────────────────────────────────── + +#[test] +fn test_relay_client_new_succeeds() { + let client = RelayClient::new( + "http://localhost:9999".to_string(), + SecretString::from("key".to_string()), + 30, + ); + assert!(client.is_ok()); +} + +// ── SSE UTF-8 chunk boundary ──────────────────────────────────────────── + +/// Verify that multi-byte UTF-8 characters split across SSE chunks are +/// not corrupted (no U+FFFD replacement characters). +#[tokio::test] +async fn test_sse_stream_preserves_multibyte_utf8_across_chunks() { + use std::sync::atomic::{AtomicBool, Ordering}; + + let sent = std::sync::Arc::new(AtomicBool::new(false)); + let sent_clone = sent.clone(); + + let app = Router::new().route( + "/stream", + get(move |_: Query>| { + let sent = sent_clone.clone(); + async move { + // Build SSE payload with emoji that will be split mid-character + let event_data = serde_json::json!({ + "event_type": "message", + "provider": "slack", + "provider_scope": "T1", + "channel_id": "C1", + "sender_id": "U1", + "content": "hello 🦀 world" + }); + let payload = format!("event: message\ndata: {}\n\n", event_data); + let bytes = payload.into_bytes(); + + // Split in the middle of the 4-byte crab emoji + let crab_pos = bytes + .windows(4) + .position(|w| w == [0xF0, 0x9F, 0xA6, 0x80]) + .unwrap(); + let split_at = crab_pos + 2; + + let chunk1 = bytes[..split_at].to_vec(); + let chunk2 = bytes[split_at..].to_vec(); + + sent.store(true, Ordering::SeqCst); + + let events = vec![ + Ok::<_, Infallible>(axum::body::Bytes::from(chunk1)), + Ok(axum::body::Bytes::from(chunk2)), + ]; + + axum::response::Response::builder() + .header("content-type", "text/event-stream") + .body(axum::body::Body::from_stream(stream::iter(events))) + .unwrap() + } + }), + ); + + let base_url = start_server(app).await; + let client = test_client(&base_url); + + let (mut event_stream, handle) = client.connect_stream("tok", 30).await.unwrap(); + + use futures::StreamExt; + let event = event_stream.next().await.expect("should get event"); + assert_eq!( + event.text(), + "hello 🦀 world", + "emoji should not be corrupted" + ); + assert!(sent.load(Ordering::SeqCst)); + + handle.abort(); +} + +// ── Channel event field validation ────────────────────────────────────── + +#[test] +fn test_channel_event_missing_fields_detected() { + use ironclaw::channels::relay::client::ChannelEvent; + + // Event with empty sender_id should be detectable + let json = r#"{"event_type": "message", "provider_scope": "T1", "channel_id": "C1", "sender_id": "", "content": "test"}"#; + let event: ChannelEvent = serde_json::from_str(json).unwrap(); + assert!(event.sender_id.is_empty()); + + // Event with all fields present + let json = r#"{"event_type": "message", "provider_scope": "T1", "channel_id": "C1", "sender_id": "U1", "content": "test"}"#; + let event: ChannelEvent = serde_json::from_str(json).unwrap(); + assert!(!event.sender_id.is_empty()); + assert!(!event.channel_id.is_empty()); + assert!(!event.provider_scope.is_empty()); +} diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index 6f66e19e..51e39d8d 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -57,6 +57,7 @@ async fn start_test_server() -> ( skill_registry: None, skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), + oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)),