diff --git a/Cargo.lock b/Cargo.lock index 8fe9f11..fb47514 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -734,8 +734,10 @@ dependencies = [ "async-trait", "chrono", "futures", + "gyandoot-agent", "gyandoot-core", "gyandoot-memory", + "gyandoot-platform", "gyandoot-providers", "reqwest", "serde", diff --git a/crates/gyandoot-core/src/config.rs b/crates/gyandoot-core/src/config.rs index b41ff6c..1d61197 100644 --- a/crates/gyandoot-core/src/config.rs +++ b/crates/gyandoot-core/src/config.rs @@ -131,9 +131,12 @@ pub struct PlatformSection { /// Default platform (e.g. "cli", "telegram", "discord"). #[serde(default = "default_platform_default")] pub default: String, - /// Platform-specific config (flattened for extensibility). - #[serde(default, flatten)] - pub platforms: HashMap, + /// Telegram bot configuration. + #[serde(default)] + pub telegram: Option, + /// Discord bot configuration. + #[serde(default)] + pub discord: Option, } fn default_platform_default() -> String { @@ -144,11 +147,31 @@ impl Default for PlatformSection { fn default() -> Self { Self { default: default_platform_default(), - platforms: HashMap::new(), + telegram: None, + discord: None, } } } +/// Telegram bot config as written in [platform.telegram] section. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TelegramPlatformCfg { + /// Bot token from @BotFather. + #[serde(default)] + pub bot_token: String, +} + +/// Discord bot config as written in [platform.discord] section. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DiscordPlatformCfg { + /// Bot token from Discord Developer Portal. + #[serde(default)] + pub bot_token: String, + /// Optional guild ID for channel discovery. + #[serde(default)] + pub guild_id: Option, +} + /// `[memory]` section. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MemorySection { @@ -343,8 +366,14 @@ impl From for AgentConfig { }, platform: PlatformRootConfig { default: cfg.platform.default, - telegram: None, - discord: None, + telegram: cfg.platform.telegram.map(|t| TelegramConfig { + bot_token: t.bot_token, + allowed_users: Vec::new(), + }), + discord: cfg.platform.discord.map(|d| DiscordConfig { + bot_token: d.bot_token, + guild_id: d.guild_id, + }), webhook: None, }, memory: MemoryRootConfig { diff --git a/crates/gyandoot-core/src/lib.rs b/crates/gyandoot-core/src/lib.rs index f6e5bb2..a996a71 100644 --- a/crates/gyandoot-core/src/lib.rs +++ b/crates/gyandoot-core/src/lib.rs @@ -11,11 +11,11 @@ pub mod service; pub use config::{ AgentConfig, AgentSection, AnthropicConfig, BrowserSection, ConfigLoader, - GatewaySection, GyandootConfig, IdentitySection, + DiscordPlatformCfg, GatewaySection, GyandootConfig, IdentitySection, KanbanRootConfig, MemoryRootConfig, MemorySection, OllamaConfig, OpenAIConfig, PlatformRootConfig, PlatformSection, ProviderRootConfig, ProvidersSection, SecurityRootConfig, SkillsRootConfig, TelegramConfig, - WebhookConfig, DiscordConfig, load_config_from_file, + TelegramPlatformCfg, WebhookConfig, DiscordConfig, load_config_from_file, }; pub use error::*; pub use types::*; \ No newline at end of file diff --git a/crates/gyandoot-gateway/Cargo.toml b/crates/gyandoot-gateway/Cargo.toml index b942de2..80fdef3 100644 --- a/crates/gyandoot-gateway/Cargo.toml +++ b/crates/gyandoot-gateway/Cargo.toml @@ -9,6 +9,8 @@ description = "GyanDoot v2 — Gateway for platform sessions, routing, streaming gyandoot-core = { path = "../gyandoot-core" } gyandoot-memory = { path = "../gyandoot-memory" } gyandoot-providers = { path = "../gyandoot-providers" } +gyandoot-platform = { path = "../gyandoot-platform" } +gyandoot-agent = { path = "../gyandoot-agent" } tokio.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/gyandoot-gateway/src/lib.rs b/crates/gyandoot-gateway/src/lib.rs index b0e439a..94bd3ce 100644 --- a/crates/gyandoot-gateway/src/lib.rs +++ b/crates/gyandoot-gateway/src/lib.rs @@ -2,14 +2,26 @@ //! //! Gateway for platform sessions, message routing, and streaming delivery. //! Connects platform adapters (Telegram, Discord, Webhook) to the agent core. +//! +//! ## Architecture +//! +//! ```text +//! Telegram ──┐ +//! Discord ──┤──→ mpsc::Receiver ──→ AgentLoop ──→ response → platform.send() +//! Webhook ──┘ +//! ``` use std::collections::HashMap; use std::sync::Arc; -use tokio::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::sync::{Mutex, mpsc}; use uuid::Uuid; use chrono::Utc; use serde::{Deserialize, Serialize}; -use tracing::info; +use tracing::{error, info, warn}; + +use gyandoot_agent::agent_loop::AgentLoop; +use gyandoot_platform::{IncomingMessage, OutgoingMessage, Platform}; /// Unique session identifier. pub type SessionId = Uuid; @@ -97,7 +109,7 @@ impl AccessControl { pub fn check_user(&self, username: &str) -> bool { match &self.allowed_users { Some(users) => users.iter().any(|u| u.eq_ignore_ascii_case(username)), - None => true, // empty = allow all + None => true, } } @@ -115,19 +127,31 @@ impl Default for AccessControl { } } +/// Multi-platform adapter list. +type PlatformList = Arc>>>; + /// The main gateway runner — orchestrates platform connections and message flow. pub struct GatewayRunner { pub sessions: Arc>>, pub access_control: AccessControl, - running: Arc>, + running: Arc, + platforms: PlatformList, + /// Receivers from all platform pollers (merged into one). + rx: Option>, } impl GatewayRunner { - pub fn new() -> Self { + /// Create a new gateway with platforms. + /// + /// Takes ownership of platform adapters. At least one must be provided + /// (though you can add more later with `add_platform`). + pub fn new(platforms: Vec>) -> Self { Self { sessions: Arc::new(Mutex::new(HashMap::new())), access_control: AccessControl::new(), - running: Arc::new(Mutex::new(false)), + running: Arc::new(AtomicBool::new(false)), + platforms: Arc::new(Mutex::new(platforms)), + rx: None, } } @@ -136,21 +160,185 @@ impl GatewayRunner { self } - pub async fn start(&self) -> anyhow::Result<()> { - let mut running = self.running.lock().await; - *running = true; - info!("Gateway started"); + /// Add a platform adapter after construction. + pub async fn add_platform(&self, platform: Box) { + let mut platforms = self.platforms.lock().await; + platforms.push(platform); + info!("Platform adapter added (total: {})", platforms.len()); + } + + /// Start all platform adapters and begin routing messages. + /// + /// This spawns polling loops for each platform and merges their message + /// receivers. Does NOT block — call `serve(agent)` to process messages. + pub async fn start(&mut self) -> anyhow::Result<()> { + self.running.store(true, Ordering::SeqCst); + + let platforms = self.platforms.lock().await; + if platforms.is_empty() { + info!("Gateway started with no platform adapters"); + return Ok(()); + } + + // Start each platform and collect receivers + let mut all_rxs = Vec::new(); + for platform in platforms.iter() { + match platform.start().await { + Ok(rx) => { + info!("Platform '{}' started successfully", platform.name()); + all_rxs.push((platform.name().to_string(), rx)); + } + Err(e) => { + warn!("Platform '{}' failed to start: {}", platform.name(), e); + // Non-fatal — other platforms may still work + } + } + } + + if all_rxs.is_empty() { + warn!("No platform adapters started successfully"); + return Ok(()); + } + + // Merge all receivers into one via a select loop + let (tx, rx) = mpsc::channel::(1024); + self.rx = Some(rx); + + let running = Arc::clone(&self.running); + tokio::spawn(async move { + loop { + if !running.load(Ordering::SeqCst) { + break; + } + + // Check all platform receivers + for (name, rx) in &mut all_rxs { + match rx.try_recv() { + Ok(msg) => { + if tx.send(msg).await.is_err() { + warn!("Gateway message channel closed"); + return; + } + } + Err(mpsc::error::TryRecvError::Disconnected) => { + warn!("Platform '{name}' receiver disconnected"); + } + Err(mpsc::error::TryRecvError::Empty) => {} + } + } + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }); + + info!("Gateway started with {} platform(s)", platforms.len()); Ok(()) } - pub async fn stop(&self) { - let mut running = self.running.lock().await; - *running = false; + /// Run the gateway message processing loop with an agent. + /// + /// Blocks until the gateway is stopped, forwarding incoming messages + /// to the agent and routing responses back through the correct platform. + pub async fn serve(&mut self, agent: &AgentLoop) -> anyhow::Result<()> { + // Ensure the gateway is running + if !self.running.load(Ordering::SeqCst) { + self.start().await?; + } + + let mut rx = self.rx.take().ok_or_else(|| { + anyhow::anyhow!("Gateway not started — call start() first or use serve() directly") + })?; + + info!("Gateway serve loop started — listening for messages"); + + // Init agent session + agent.init_session().await.map_err(|e| { + anyhow::anyhow!("Agent session init failed: {}", e) + })?; + + let platforms = Arc::clone(&self.platforms); + let sessions = Arc::clone(&self.sessions); + + while self.running.load(Ordering::SeqCst) { + tokio::select! { + Some(msg) = rx.recv() => { + // Access control + if !self.access_control.check_user(&msg.username) { + warn!("Access denied for user: {}", msg.username); + continue; + } + if !self.access_control.check_channel(&msg.channel_id) { + warn!("Access denied for channel: {}", msg.channel_id); + continue; + } + + // Create/update session + { + let source = SessionSource { + platform: msg.platform.clone(), + chat_id: msg.channel_id.clone(), + thread_id: None, + }; + let mut sessions = sessions.lock().await; + // Find existing session for this source or create new + let existing = sessions.values_mut().find(|s| { + s.source.platform == msg.platform + && s.source.chat_id == msg.channel_id + }); + match existing { + Some(session) => { + session.last_active = Utc::now(); + session.message_count += 1; + } + None => { + let session = GatewaySession::new(source); + sessions.insert(session.id, session); + } + } + } + + // Send typing indicator + let platforms_guard = platforms.lock().await; + if let Some(p) = platforms_guard.iter().find(|p| p.name() == msg.platform) { + let _ = p.send_typing(&msg.channel_id).await; + } + + // Process with agent + let response = match agent.session_turn(&msg.text).await { + Ok(text) => text, + Err(e) => { + error!("Agent error: {}", e); + format!("⚠️ Error processing your message: {}", e) + } + }; + + // Send response back through the same platform + if let Some(p) = platforms_guard.iter().find(|p| p.name() == msg.platform) { + let outgoing = OutgoingMessage { + channel_id: msg.channel_id.clone(), + text: response, + reply_to: Some(msg.message_id), + }; + if let Err(e) = p.send(outgoing).await { + error!("Failed to send response on {}: {}", msg.platform, e); + } + } + } + else => break, + } + } + + info!("Gateway serve loop finished"); + Ok(()) + } + + pub fn stop(&self) { + self.running.store(false, Ordering::SeqCst); info!("Gateway stopped"); } - pub async fn is_running(&self) -> bool { - *self.running.lock().await + pub fn is_running(&self) -> bool { + self.running.load(Ordering::SeqCst) } pub async fn create_session(&self, source: SessionSource) -> SessionId { @@ -167,12 +355,6 @@ impl GatewayRunner { } } -impl Default for GatewayRunner { - fn default() -> Self { - Self::new() - } -} - #[cfg(test)] mod tests { use super::*; @@ -210,11 +392,11 @@ mod tests { #[tokio::test] async fn test_gateway_lifecycle() { - let gw = GatewayRunner::new(); - assert!(!gw.is_running().await); + let mut gw = GatewayRunner::new(vec![]); + assert!(!gw.is_running()); gw.start().await.unwrap(); - assert!(gw.is_running().await); - gw.stop().await; - assert!(!gw.is_running().await); + assert!(gw.is_running()); + gw.stop(); + assert!(!gw.is_running()); } } \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 8ca464c..e7db7ae 100644 --- a/src/main.rs +++ b/src/main.rs @@ -364,19 +364,53 @@ fn load_and_convert_config(path: &std::path::Path) -> Result anyhow::Result<()> { - tracing::info!( - "Starting gateway server (profile={}, bind={}:{})", - profile_name, host, port - ); +async fn cmd_serve(profile_name: &str, _host: &str, _port: u16) -> anyhow::Result<()> { + tracing::info!("Starting gateway server (profile={})", profile_name); - let gateway = gyandoot_gateway::GatewayRunner::new(); + let config = load_agent_config(); + let agent = gyandoot_agent::agent_loop::AgentLoop::new(config.clone()); + + // Build platform adapters from config + let mut platforms: Vec> = Vec::new(); + + // Telegram + if let Some(tg) = &config.platform.telegram { + if !tg.bot_token.is_empty() { + platforms.push(Box::new( + gyandoot_platform::telegram::TelegramPlatform::new(&tg.bot_token), + )); + tracing::info!("Telegram platform added"); + } + } + + // Discord + if let Some(dc) = &config.platform.discord { + if !dc.bot_token.is_empty() { + platforms.push(Box::new( + gyandoot_platform::discord::DiscordPlatform::new(&dc.bot_token, dc.guild_id), + )); + tracing::info!("Discord platform added"); + } + } + + // Gateway + let mut gateway = gyandoot_gateway::GatewayRunner::new(platforms); gateway.start().await?; - tracing::info!("Gateway running on {}:{} — press Ctrl+C to stop", host, port); + tracing::info!("Gateway started — listening on platforms"); + + tokio::select! { + _ = tokio::signal::ctrl_c() => { + tracing::info!("Shutting down gateway..."); + gateway.stop(); + } + result = gateway.serve(&agent) => { + match result { + Ok(()) => tracing::info!("Gateway serve completed"), + Err(e) => tracing::error!("Gateway serve error: {}", e), + } + } + } - // Keep running until signal - tokio::signal::ctrl_c().await?; - gateway.stop().await; Ok(()) }