Merge pull request #907 from nearai/staging-promote/b0214fef-22930316561

chore: promote staging to main (2026-03-11 00:16 UTC)

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<dyn crate::secrets::SecretsStore + Send + Sync> = if let Some(ref s) =

src/channels/manager.rs

@@ -56,6 +56,17 @@ impl ChannelManager {
     /// the agent loop.
     pub async fn hot_add(&self, channel: Box<dyn Channel>) -> 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"));
+    }
 }

src/channels/mod.rs

@@ -30,6 +30,7 @@
 mod channel;
 mod http;
 mod manager;
+pub mod relay;
 mod repl;
 mod signal;
 pub mod wasm;

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<RwLock<String>>,
+    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<Option<tokio::task::JoinHandle<()>>>,
+    /// Handle to the SSE parser task for clean shutdown.
+    parser_handle: Arc<RwLock<Option<tokio::task::JoinHandle<()>>>>,
+    /// 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<serde_json::Value, RelayError> {
+        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<MessageStream, ChannelError> {
+        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<String, String> {
+        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.
+    }
+}

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<String>,
+    /// Message text content (called `content` in channel-relay).
+    #[serde(alias = "text", default)]
+    pub content: Option<String>,
+    /// Thread ID (for threaded replies, called `thread_id` in channel-relay).
+    #[serde(alias = "thread_ts", default)]
+    pub thread_id: Option<String>,
+    /// Full raw event data.
+    #[serde(default)]
+    pub raw: serde_json::Value,
+    /// Event timestamp (ISO 8601 from channel-relay).
+    #[serde(default)]
+    pub timestamp: Option<String>,
+}
+
+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<String>,
+    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<Self, RelayError> {
+        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<String, RelayError> {
+        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<String, RelayError> {
+        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<serde_json::Value, RelayError> {
+        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<Vec<Connection>, 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<ChannelEvent>,
+}
+
+impl Stream for ChannelEventStream {
+    type Item = ChannelEvent;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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<Item = Result<bytes::Bytes, reqwest::Error>> + Send + 'static,
+    tx: mpsc::Sender<ChannelEvent>,
+) {
+    use futures::StreamExt;
+
+    let mut buffer = Vec::<u8>::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::<ChannelEvent>(&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<Result<bytes::Bytes, reqwest::Error>> = 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");
+    }
+}

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;

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<Arc<GatewayState>>,
-    Path(name): Path<String>,
-) -> Result<Json<ActionResponse>, (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<Arc<GatewayState>>,
     Path(name): Path<String>,

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),

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<crate::tools::builtin::SchedulerSlot>,
     /// 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<crate::extensions::RegistryEntry>,
@@ -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<Arc<GatewayState>>,
+    Query(params): Query<std::collections::HashMap<String, String>>,
+) -> impl IntoResponse {
+    // Rate limit
+    if !state.oauth_rate_limiter.check() {
+        return axum::response::Html(
+            "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+             <h2>Too Many Requests</h2>\
+             <p>Please try again later.</p>\
+             </body></html>"
+                .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(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Invalid callback parameters.</p></body></html>"
+                    .to_string(),
+            )
+            .into_response();
+        }
+        _ => {
+            return axum::response::Html(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Invalid callback parameters.</p></body></html>"
+                    .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(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Invalid callback parameters.</p></body></html>"
+                    .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(
+            "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+             <h2>Error</h2><p>Invalid callback parameters.</p></body></html>"
+                .to_string(),
+        )
+        .into_response();
+    }
+
+    let ext_mgr = match state.extension_manager.as_ref() {
+        Some(mgr) => mgr,
+        None => {
+            return axum::response::Html(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Extension manager not available.</p></body></html>"
+                    .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(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Invalid or expired authorization.</p></body></html>"
+                    .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(
+                "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+                 <h2>Error</h2><p>Invalid or expired authorization.</p></body></html>"
+                    .to_string(),
+            )
+            .into_response();
+        }
+    };
+
+    if state_param != stored_state {
+        return axum::response::Html(
+            "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+             <h2>Error</h2><p>Invalid or expired authorization.</p></body></html>"
+                .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(
+            "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+             <h2>Slack Connected!</h2>\
+             <p>You can close this tab and return to IronClaw.</p>\
+             <script>window.close()</script>\
+             </body></html>"
+                .to_string(),
+        )
+        .into_response()
+    } else {
+        axum::response::Html(format!(
+            "<html><body style='font-family: system-ui; text-align: center; padding: 60px;'>\
+             <h2>Connection Failed</h2>\
+             <p>{}</p>\
+             </body></html>",
+            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<GatewayState>) -> Router {
+        Router::new()
+            .route(
+                "/oauth/slack/callback",
+                get(slack_relay_oauth_callback_handler),
+            )
+            .with_state(state)
+    }
+
+    fn test_secrets_store() -> Arc<dyn crate::secrets::SecretsStore + Send + Sync> {
+        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<dyn crate::secrets::SecretsStore + Send + Sync>,
+    ) -> Arc<ExtensionManager> {
+        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::<axum::http::Request<Body>>::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::<axum::http::Request<Body>>::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::<axum::http::Request<Body>>::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");
+    }
 }

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';

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)),

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)),

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<RelayConfig>,
 }
 
 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(),
         })
     }
 }

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<String>,
+    /// Override for the instance identifier.
+    pub instance_id: Option<String>,
+    /// 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> {
+        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<String>, api_key: impl Into<String>) -> 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<String>) -> Option<Self> {
+        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"));
+    }
+}

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(),
     }
 }
 

src/extensions/manager.rs

@@ -84,6 +84,8 @@ pub struct ExtensionManager {
 
     // WASM channel hot-activation infrastructure (set post-construction)
     channel_runtime: RwLock<Option<ChannelRuntimeState>>,
+    /// Channel manager for hot-adding relay channels (set independently of WASM runtime).
+    relay_channel_manager: RwLock<Option<Arc<ChannelManager>>>,
 
     // Shared
     secrets: Arc<dyn SecretsStore + Send + Sync>,
@@ -97,6 +99,8 @@ pub struct ExtensionManager {
     store: Option<Arc<dyn crate::db::Database>>,
     /// Names of WASM channels that were successfully loaded at startup.
     active_channel_names: RwLock<HashSet<String>>,
+    /// Installed channel-relay extensions (no on-disk artifact, tracked in memory).
+    installed_relay_extensions: RwLock<HashSet<String>>,
     /// Last activation error for each WASM channel (ephemeral, cleared on success).
     activation_errors: RwLock<HashMap<String, String>>,
     /// 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<String>,
+    /// Relay config captured at startup. Used by `auth_channel_relay` and
+    /// `activate_channel_relay` instead of re-reading env vars.
+    relay_config: Option<crate::config::RelayConfig>,
 }
 
 /// 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<WasmChannelRouter>,
         wasm_channel_owner_ids: std::collections::HashMap<String, i64>,
     ) {
+        // 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<ChannelManager>) {
+        *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<dyn SecretsStore + Send + Sync> {
+        &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<String>) {
@@ -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<AuthResult, ExtensionError> {
+        // 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<ActivateResult, ExtensionError> {
+        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<ExtensionKind, ExtensionError> {
         // 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";

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",

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<RegistryEntry> {
-    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<RegistryEntry> {
+    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<String>) -> Vec<RegistryEntry> {
+    let mut entries = vec![
         // -- MCP Servers --
         RegistryEntry {
             name: "notion".to_string(),
@@ -415,7 +423,29 @@ fn builtin_entries() -> Vec<RegistryEntry> {
         // 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");
+        }
+    }
 }

src/main.rs

@@ -564,28 +564,39 @@ 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) {
+            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 channel"
+                        "Auto-activated persisted WASM channel"
                     );
                 }
                 Err(e) => {
                     tracing::warn!(
                         channel = %name,
                         error = %e,
-                            "Failed to auto-activate persisted channel"
+                        "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.

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)),

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<std::collections::HashMap<String, String>>| 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<serde_json::Value>| {
+            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<ProxyQuery>, Json(body): Json<serde_json::Value>| 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<std::collections::HashMap<String, String>>| {
+            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());
+}

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)),