diff --git a/src/channels/feishu.rs b/src/channels/feishu.rs index 218c10e..cb656e0 100644 --- a/src/channels/feishu.rs +++ b/src/channels/feishu.rs @@ -8,7 +8,8 @@ use futures_util::{SinkExt, StreamExt}; use prost::{Message as ProstMessage, bytes::Bytes}; use regex::Regex; use serde::Deserialize; -use tokio::sync::{RwLock, broadcast}; +use tokio::sync::{Mutex, RwLock, broadcast}; +use tokio::task::JoinHandle; use crate::bus::{MediaItem, MessageBus, OutboundMessage}; use crate::channels::base::{Channel, ChannelError}; @@ -152,6 +153,7 @@ pub struct FeishuChannel { http_client: reqwest::Client, running: Arc>, shutdown_tx: Arc>>>, + run_task: Arc>>>, connected: Arc>, /// Cached tenant access token with proactive refresh. tenant_token: Arc>>, @@ -185,6 +187,7 @@ impl FeishuChannel { http_client: reqwest::Client::new(), running: Arc::new(RwLock::new(false)), shutdown_tx: Arc::new(RwLock::new(None)), + run_task: Arc::new(Mutex::new(None)), connected: Arc::new(RwLock::new(false)), tenant_token: Arc::new(RwLock::new(None)), seen_message_ids: Arc::new(RwLock::new(HashMap::new())), @@ -1953,6 +1956,14 @@ impl Channel for FeishuChannel { )); } + let mut run_task = self.run_task.lock().await; + if run_task.as_ref().is_some_and(|task| !task.is_finished()) { + return Ok(()); + } + if let Some(finished) = run_task.take() { + let _ = finished.await; + } + *self.running.write().await = true; let (shutdown_tx, _) = broadcast::channel(1); @@ -1960,7 +1971,7 @@ impl Channel for FeishuChannel { let channel = self.clone(); let bus = bus.clone(); - tokio::spawn(async move { + *run_task = Some(tokio::spawn(async move { let mut consecutive_failures = 0; let max_failures = 3; @@ -1994,7 +2005,7 @@ impl Channel for FeishuChannel { *channel.running.write().await = false; tracing::info!("Feishu channel stopped"); - }); + })); tracing::info!("Feishu channel started"); Ok(()) @@ -2007,6 +2018,12 @@ impl Channel for FeishuChannel { let _ = tx.send(()); } + if let Some(task) = self.run_task.lock().await.take() { + task.await.map_err(|error| { + ChannelError::Other(format!("Feishu channel task failed to join: {error}")) + })?; + } + Ok(()) } diff --git a/src/channels/manager.rs b/src/channels/manager.rs index d089ea9..0dd3727 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; -use crate::bus::{MessageBus, OutboundMessage}; +use crate::bus::MessageBus; use crate::channels::base::{Channel, ChannelError}; use crate::channels::feishu::FeishuChannel; use crate::config::Config; @@ -16,14 +16,6 @@ pub struct ChannelManager { } impl ChannelManager { - pub fn new(cli_chat_channel: Arc) -> Self { - Self { - channels: Arc::new(RwLock::new(HashMap::new())), - cli_chat_channel, - bus: MessageBus::new(100), - } - } - pub fn with_bus( cli_chat_channel: Arc, bus: Arc, @@ -91,27 +83,53 @@ impl ChannelManager { } pub async fn start_all(&self) -> Result<(), ChannelError> { - let channels = self.channels.read().await; + let channels: Vec<_> = self + .channels + .read() + .await + .iter() + .map(|(name, channel)| (name.clone(), channel.clone())) + .collect(); let bus = self.bus.clone(); - for (name, channel) in channels.iter() { + let mut failures = Vec::new(); + for (name, channel) in channels { tracing::info!(channel = %name, "Starting channel"); if let Err(e) = channel.start(bus.clone()).await { tracing::error!(channel = %name, error = %e, "Failed to start channel"); + failures.push(format!("{name}: {e}")); } } - Ok(()) + if failures.is_empty() { + Ok(()) + } else { + Err(ChannelError::Other(format!( + "failed to start channels: {}", + failures.join("; ") + ))) + } } pub async fn stop_all(&self) -> Result<(), ChannelError> { - let mut channels = self.channels.write().await; - for (name, channel) in channels.iter() { + let channels: Vec<_> = { + let mut registered = self.channels.write().await; + registered.drain().collect() + }; + let mut failures = Vec::new(); + for (name, channel) in channels { tracing::info!(channel = %name, "Stopping channel"); if let Err(e) = channel.stop().await { tracing::error!(channel = %name, error = %e, "Error stopping channel"); + failures.push(format!("{name}: {e}")); } } - channels.clear(); - Ok(()) + if failures.is_empty() { + Ok(()) + } else { + Err(ChannelError::Other(format!( + "failed to stop channels: {}", + failures.join("; ") + ))) + } } pub async fn get_channel(&self, name: &str) -> Option> { @@ -122,17 +140,104 @@ impl ChannelManager { pub async fn list_channel_names(&self) -> Vec { self.channels.read().await.keys().cloned().collect() } +} - /// Dispatch an outbound message to the appropriate channel - pub async fn dispatch(&self, msg: OutboundMessage) -> Result<(), ChannelError> { - let channel_name = &msg.channel; - if let Some(channel) = self.get_channel(channel_name).await { - channel.send(msg).await - } else { - Err(ChannelError::Other(format!( - "Channel not found: {}", - channel_name - ))) +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + + struct TestChannel { + name: &'static str, + fail_start: bool, + fail_stop: bool, + running: AtomicBool, + starts: AtomicUsize, + stops: AtomicUsize, + } + + impl TestChannel { + fn new(name: &'static str, fail_start: bool, fail_stop: bool) -> Self { + Self { + name, + fail_start, + fail_stop, + running: AtomicBool::new(false), + starts: AtomicUsize::new(0), + stops: AtomicUsize::new(0), + } } } + + #[async_trait] + impl Channel for TestChannel { + fn name(&self) -> &str { + self.name + } + + fn is_running(&self) -> bool { + self.running.load(Ordering::SeqCst) + } + + async fn start(&self, _bus: Arc) -> Result<(), ChannelError> { + self.starts.fetch_add(1, Ordering::SeqCst); + if self.fail_start { + return Err(ChannelError::ConnectionError("unavailable".into())); + } + self.running.store(true, Ordering::SeqCst); + Ok(()) + } + + async fn stop(&self) -> Result<(), ChannelError> { + self.stops.fetch_add(1, Ordering::SeqCst); + self.running.store(false, Ordering::SeqCst); + if self.fail_stop { + return Err(ChannelError::Other("stop failed".into())); + } + Ok(()) + } + + async fn send(&self, _msg: crate::bus::OutboundMessage) -> Result<(), ChannelError> { + Ok(()) + } + } + + fn manager() -> ChannelManager { + ChannelManager::with_bus( + Arc::new(crate::channels::CliChatChannel::new()), + MessageBus::new(8), + ) + } + + #[tokio::test] + async fn start_all_reports_channel_failures_and_starts_the_rest() { + let manager = manager(); + let healthy = Arc::new(TestChannel::new("healthy", false, false)); + let broken = Arc::new(TestChannel::new("broken", true, false)); + manager.register_channel("healthy", healthy.clone()).await; + manager.register_channel("broken", broken.clone()).await; + + let error = manager.start_all().await.unwrap_err().to_string(); + + assert!(error.contains("broken")); + assert_eq!(healthy.starts.load(Ordering::SeqCst), 1); + assert_eq!(broken.starts.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn stop_all_reports_failures_and_unregisters_every_channel() { + let manager = manager(); + let healthy = Arc::new(TestChannel::new("healthy", false, false)); + let broken = Arc::new(TestChannel::new("broken", false, true)); + manager.register_channel("healthy", healthy.clone()).await; + manager.register_channel("broken", broken.clone()).await; + + let error = manager.stop_all().await.unwrap_err().to_string(); + + assert!(error.contains("broken")); + assert_eq!(healthy.stops.load(Ordering::SeqCst), 1); + assert_eq!(broken.stops.load(Ordering::SeqCst), 1); + assert!(manager.list_channel_names().await.is_empty()); + } } diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 0990e87..ee1613c 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use tokio::net::TcpListener; use crate::bus::{ControlMessage, MessageBus, OutboundDispatcher}; -use crate::channels::base::{Channel, ChannelError}; +use crate::channels::base::ChannelError; use crate::channels::{ChannelManager, CliChatChannel}; use crate::config::{Config, ensure_workspace_dir, expand_path}; use crate::logging; @@ -198,12 +198,6 @@ impl GatewayState { let bus_for_outbound = bus.clone(); let session_manager = self.session_manager.clone(); - // Start CLI Chat Channel (it's already registered in ChannelManager) - let cli_chat_channel = self.cli_chat_channel(); - if let Err(e) = cli_chat_channel.start(bus.clone()).await { - tracing::error!(error = %e, "Failed to start CLI chat channel"); - } - // Spawn unified message processor // This handles both inbound AI messages and control messages in one loop self.task_supervisor.spawn("message-processor", async move {