feat: add channel-relay integration for Slack (#790)

* fix(ci): secrets can't be used in step if conditions [skip-regression-check] (#787)

GitHub Actions step-level `if:` doesn't have access to `secrets` context.
Replace `if: secrets.X != ''` with `continue-on-error: true` and let
the Set token step handle the fallback.

Co-authored-by: Claude Sonnet 4.6 <[email protected]>

* feat: add channel-relay integration for Slack via external relay service

- Add RelayChannel and RelayClient for connecting to channel-relay SSE streams
- Add RelayConfig with env-based configuration (CHANNEL_RELAY_URL, CHANNEL_RELAY_API_KEY)
- Add channel-relay extension lifecycle: install, OAuth auth, activate with hot-add
- Add proxy message sending through channel-relay for Slack chat.postMessage
- Add extension registry entry for Slack relay with OAuth auth hint
- Add relay integration test with mock SSE server
- Wire relay channel into app startup with reconnect on stored credentials
- Add AuthRequired extension error variant for cleaner auth flow detection

[skip-regression-check]

* chore: apply cargo fmt

* fix: remove remaining Telegram test references in relay channel

* fix: address PR #790 review feedback — parser handle leak, CSRF, circuit breaker

- Fix parser handle leak on reconnect by sharing Arc<RwLock> instead of
  creating a local copy in start() (shutdown now aborts the correct task)
- Add CSRF state nonce to OAuth flow: generate in auth_channel_relay,
  validate in slack_relay_oauth_callback_handler, one-time use
- Remove dead proxy_slack method, update integration test to use
  proxy_provider
- Add reconnect circuit breaker (max_consecutive_failures, default 50)
- Fix stale docs (Telegram refs), extract event_types constants

---------

Co-authored-by: Henry Park <[email protected]>
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
Pierre LE GUEN
2026-03-10 16:34:54 -07:00
committed by GitHub
co-authored by Henry Park Claude Sonnet 4.6
parent 3a841b30d8
commit b0214fef41
22 changed files with 2707 additions and 89 deletions
+10 -1
View File
@@ -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) =
+37
View File
@@ -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"));
}
}
+1
View File
@@ -30,6 +30,7 @@
mod channel;
mod http;
mod manager;
pub mod relay;
mod repl;
mod signal;
pub mod wasm;
+642
View File
@@ -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.
}
}
+549
View File
@@ -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");
}
}
+12
View File
@@ -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;
+9 -56
View File
@@ -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>,
+2
View File
@@ -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),
+389 -6
View File
@@ -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");
}
}
+2 -2
View File
@@ -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';
+1
View File
@@ -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)),
+1
View File
@@ -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)),
+7
View File
@@ -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(),
})
}
}
+157
View File
@@ -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"));
}
}
+1
View File
@@ -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(),
}
}
+463 -4
View File
@@ -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";
+11
View File
@@ -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",
+59 -3
View File
@@ -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");
}
}
}
+28 -17
View File
@@ -564,30 +564,41 @@ async fn async_main() -> anyhow::Result<()> {
.await;
tracing::debug!("Channel runtime wired into extension manager for hot-activation");
// Auto-activate channels that were active in a previous session.
// Auto-activate WASM channels that were active in a previous session.
// Relay channels are handled separately below via restore_relay_channels().
let persisted = ext_mgr.load_persisted_active_channels().await;
for name in &persisted {
if !active_at_startup.contains(name) {
match ext_mgr.activate(name).await {
Ok(result) => {
tracing::debug!(
channel = %name,
message = %result.message,
"Auto-activated persisted channel"
);
}
Err(e) => {
tracing::warn!(
channel = %name,
error = %e,
"Failed to auto-activate persisted channel"
);
}
if active_at_startup.contains(name) || ext_mgr.is_relay_channel(name).await {
continue;
}
match ext_mgr.activate(name).await {
Ok(result) => {
tracing::debug!(
channel = %name,
message = %result.message,
"Auto-activated persisted WASM channel"
);
}
Err(e) => {
tracing::warn!(
channel = %name,
error = %e,
"Failed to auto-activate persisted WASM channel"
);
}
}
}
}
// Ensure the relay channel manager is always set (even without WASM runtime),
// then restore any persisted relay channels.
if let Some(ref ext_mgr) = components.extension_manager {
ext_mgr
.set_relay_channel_manager(Arc::clone(&channels))
.await;
ext_mgr.restore_relay_channels().await;
}
// Wire SSE sender into extension manager for broadcasting status events.
if let Some(ref ext_mgr) = components.extension_manager
&& let Some(ref sender) = sse_sender
+2
View File
@@ -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)),
+323
View File
@@ -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());
}
+1
View File
@@ -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)),