From 33cb2bed3e65b35ea807aece552a296f52c95df7 Mon Sep 17 00:00:00 2001 From: dave Date: Sat, 25 Apr 2026 18:57:35 +0000 Subject: [PATCH] huskies: merge 627_refactor_migrate_slack_discord_and_whatsapp_transports_to_services_bundle --- server/src/chat/transport/slack/commands.rs | 125 +++++++++----------- server/src/chat/transport/slack/mod.rs | 14 +-- server/src/main.rs | 20 +--- 3 files changed, 68 insertions(+), 91 deletions(-) diff --git a/server/src/chat/transport/slack/commands.rs b/server/src/chat/transport/slack/commands.rs index 0f96a121..21d74154 100644 --- a/server/src/chat/transport/slack/commands.rs +++ b/server/src/chat/transport/slack/commands.rs @@ -1,20 +1,17 @@ //! Slack incoming message dispatch and slash command handling. use serde::{Deserialize, Serialize}; -use std::collections::HashMap; use std::collections::HashSet; -use std::path::PathBuf; -use std::sync::{Arc, Mutex}; -use tokio::sync::{Mutex as TokioMutex, oneshot}; +use std::sync::Arc; use super::format::markdown_to_slack; use super::history::{SlackConversationHistory, save_slack_history}; use super::meta::SlackTransport; -use crate::agents::AgentPool; use crate::chat::ChatTransport; use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation}; use crate::chat::util::is_permission_approval; -use crate::http::context::{PermissionDecision, PermissionForward}; +use crate::http::context::PermissionDecision; +use crate::services::Services; use crate::slog; // ── Slash command types ───────────────────────────────────────────────── @@ -64,26 +61,18 @@ pub(super) fn slash_command_to_bot_keyword(command: &str) -> Option<&'static str /// Shared context for the Slack webhook handler, injected via Poem's `Data` extractor. pub struct SlackWebhookContext { + /// Shared services bundle (project root, agent pool, bot identity, permissions). + pub services: Arc, + /// Slack signing secret for verifying incoming webhook requests. pub signing_secret: String, + /// Slack Web API transport for sending/editing messages. pub transport: Arc, - pub project_root: PathBuf, - pub agents: Arc, - pub bot_name: String, - /// The bot's "user ID" for command dispatch. - pub bot_user_id: String, - pub ambient_rooms: Arc>>, /// Per-channel conversation history for LLM passthrough. pub history: SlackConversationHistory, /// Maximum number of conversation entries to keep per channel. pub history_size: usize, /// Allowed channel IDs (messages from other channels are ignored). pub channel_ids: HashSet, - /// Permission requests from the MCP `prompt_permission` tool arrive here. - pub perm_rx: Arc>>, - /// Pending permission replies keyed by channel ID. - pub pending_perm_replies: Arc>>>, - /// Seconds before an unanswered permission prompt is auto-denied. - pub permission_timeout_secs: u64, } // ── Incoming message dispatch ─────────────────────────────────────────── @@ -99,7 +88,7 @@ pub(super) async fn handle_incoming_message( // If there is a pending permission prompt for this channel, interpret the // message as a yes/no response instead of starting a new command/LLM flow. { - let mut pending = ctx.pending_perm_replies.lock().await; + let mut pending = ctx.services.pending_perm_replies.lock().await; if let Some(tx) = pending.remove(channel) { let decision = if is_permission_approval(message) { PermissionDecision::Approve @@ -119,11 +108,11 @@ pub(super) async fn handle_incoming_message( } let dispatch = CommandDispatch { - bot_name: &ctx.bot_name, - bot_user_id: &ctx.bot_user_id, - project_root: &ctx.project_root, - agents: &ctx.agents, - ambient_rooms: &ctx.ambient_rooms, + bot_name: &ctx.services.bot_name, + bot_user_id: &ctx.services.bot_user_id, + project_root: &ctx.services.project_root, + agents: &ctx.services.agents, + ambient_rooms: &ctx.services.ambient_rooms, room_id: channel, }; @@ -139,8 +128,8 @@ pub(super) async fn handle_incoming_message( // Check for async commands (htop, delete). if let Some(htop_cmd) = crate::chat::transport::matrix::htop::extract_htop_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) { use crate::chat::transport::matrix::htop::HtopCommand; slog!("[slack] Handling htop command from {user} in {channel}"); @@ -154,7 +143,7 @@ pub(super) async fn handle_incoming_message( HtopCommand::Start { duration_secs } => { // On Slack, htop uses native message editing for live updates. let snapshot = crate::chat::transport::matrix::htop::build_htop_message( - &ctx.agents, + &ctx.services.agents, 0, duration_secs, ); @@ -168,7 +157,7 @@ pub(super) async fn handle_incoming_message( }; // Spawn a background task that edits the message periodically. let transport = Arc::clone(&ctx.transport); - let agents = Arc::clone(&ctx.agents); + let agents = Arc::clone(&ctx.services.agents); let ch = channel.to_string(); tokio::spawn(async move { let interval = std::time::Duration::from_secs(2); @@ -194,22 +183,22 @@ pub(super) async fn handle_incoming_message( if let Some(del_cmd) = crate::chat::transport::matrix::delete::extract_delete_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) { let response = match del_cmd { crate::chat::transport::matrix::delete::DeleteCommand::Delete { story_number } => { slog!("[slack] Handling delete command from {user}: story {story_number}"); crate::chat::transport::matrix::delete::handle_delete( - &ctx.bot_name, + &ctx.services.bot_name, &story_number, - &ctx.project_root, - &ctx.agents, + &ctx.services.project_root, + &ctx.services.agents, ) .await } crate::chat::transport::matrix::delete::DeleteCommand::BadArgs => { - format!("Usage: `{} delete `", ctx.bot_name) + format!("Usage: `{} delete `", ctx.services.bot_name) } }; let response = markdown_to_slack(&response); @@ -219,8 +208,8 @@ pub(super) async fn handle_incoming_message( if crate::chat::transport::matrix::rebuild::extract_rebuild_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) .is_some() { @@ -228,9 +217,9 @@ pub(super) async fn handle_incoming_message( let ack = "Rebuilding server… this may take a moment."; let _ = ctx.transport.send_message(channel, ack, "").await; let response = crate::chat::transport::matrix::rebuild::handle_rebuild( - &ctx.bot_name, - &ctx.project_root, - &ctx.agents, + &ctx.services.bot_name, + &ctx.services.project_root, + &ctx.services.agents, ) .await; let response = markdown_to_slack(&response); @@ -240,8 +229,8 @@ pub(super) async fn handle_incoming_message( if let Some(rmtree_cmd) = crate::chat::transport::matrix::rmtree::extract_rmtree_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) { let response = match rmtree_cmd { crate::chat::transport::matrix::rmtree::RmtreeCommand::Rmtree { story_number } => { @@ -249,15 +238,15 @@ pub(super) async fn handle_incoming_message( "[slack] Handling rmtree command from {user} in {channel}: story {story_number}" ); crate::chat::transport::matrix::rmtree::handle_rmtree( - &ctx.bot_name, + &ctx.services.bot_name, &story_number, - &ctx.project_root, - &ctx.agents, + &ctx.services.project_root, + &ctx.services.agents, ) .await } crate::chat::transport::matrix::rmtree::RmtreeCommand::BadArgs => { - format!("Usage: `{} rmtree `", ctx.bot_name) + format!("Usage: `{} rmtree `", ctx.services.bot_name) } }; let response = markdown_to_slack(&response); @@ -267,8 +256,8 @@ pub(super) async fn handle_incoming_message( if crate::chat::transport::matrix::reset::extract_reset_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) .is_some() { @@ -280,7 +269,7 @@ pub(super) async fn handle_incoming_message( .or_insert_with(RoomConversation::default); conv.session_id = None; conv.entries.clear(); - save_slack_history(&ctx.project_root, &guard); + save_slack_history(&ctx.services.project_root, &guard); } let _ = ctx .transport @@ -291,8 +280,8 @@ pub(super) async fn handle_incoming_message( if let Some(start_cmd) = crate::chat::transport::matrix::start::extract_start_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) { let response = match start_cmd { crate::chat::transport::matrix::start::StartCommand::Start { @@ -303,16 +292,16 @@ pub(super) async fn handle_incoming_message( "[slack] Handling start command from {user} in {channel}: story {story_number}" ); crate::chat::transport::matrix::start::handle_start( - &ctx.bot_name, + &ctx.services.bot_name, &story_number, agent_hint.as_deref(), - &ctx.project_root, - &ctx.agents, + &ctx.services.project_root, + &ctx.services.agents, ) .await } crate::chat::transport::matrix::start::StartCommand::BadArgs => { - format!("Usage: `{} start `", ctx.bot_name) + format!("Usage: `{} start `", ctx.services.bot_name) } }; let response = markdown_to_slack(&response); @@ -322,8 +311,8 @@ pub(super) async fn handle_incoming_message( if let Some(assign_cmd) = crate::chat::transport::matrix::assign::extract_assign_command( message, - &ctx.bot_name, - &ctx.bot_user_id, + &ctx.services.bot_name, + &ctx.services.bot_user_id, ) { let response = match assign_cmd { crate::chat::transport::matrix::assign::AssignCommand::Assign { @@ -334,16 +323,16 @@ pub(super) async fn handle_incoming_message( "[slack] Handling assign command from {user} in {channel}: story {story_number} model {model}" ); crate::chat::transport::matrix::assign::handle_assign( - &ctx.bot_name, + &ctx.services.bot_name, &story_number, &model, - &ctx.project_root, - &ctx.agents, + &ctx.services.project_root, + &ctx.services.agents, ) .await } crate::chat::transport::matrix::assign::AssignCommand::BadArgs => { - format!("Usage: `{} assign `", ctx.bot_name) + format!("Usage: `{} assign `", ctx.services.bot_name) } }; let response = markdown_to_slack(&response); @@ -374,7 +363,7 @@ async fn handle_llm_message( guard.get(channel).and_then(|conv| conv.session_id.clone()) }; - let bot_name = &ctx.bot_name; + let bot_name = &ctx.services.bot_name; let prompt = format!( "[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}" ); @@ -404,7 +393,7 @@ async fn handle_llm_message( let sent_any_chunk = Arc::new(AtomicBool::new(false)); let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk); - let project_root_str = ctx.project_root.to_string_lossy().to_string(); + let project_root_str = ctx.services.project_root.to_string_lossy().to_string(); let chat_fut = provider.chat_stream( &prompt, &project_root_str, @@ -426,7 +415,7 @@ async fn handle_llm_message( tokio::pin!(chat_fut); // Lock the permission receiver for the duration of this chat session. - let mut perm_rx_guard = ctx.perm_rx.lock().await; + let mut perm_rx_guard = ctx.services.perm_rx.lock().await; let result = loop { tokio::select! { @@ -444,16 +433,16 @@ async fn handle_llm_message( // Store the response sender so the incoming message handler // can resolve it when the user replies yes/no. - ctx.pending_perm_replies + ctx.services.pending_perm_replies .lock() .await .insert(channel.to_string(), perm_fwd.response_tx); // Spawn a timeout task: auto-deny if the user does not respond. - let pending = Arc::clone(&ctx.pending_perm_replies); + let pending = Arc::clone(&ctx.services.pending_perm_replies); let timeout_channel = channel.to_string(); let timeout_transport = Arc::clone(&ctx.transport); - let timeout_secs = ctx.permission_timeout_secs; + let timeout_secs = ctx.services.permission_timeout_secs; tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; if let Some(tx) = pending.lock().await.remove(&timeout_channel) { @@ -534,7 +523,7 @@ async fn handle_llm_message( conv.entries.drain(..excess); } - save_slack_history(&ctx.project_root, &guard); + save_slack_history(&ctx.services.project_root, &guard); } } @@ -543,6 +532,8 @@ async fn handle_llm_message( #[cfg(test)] mod tests { use super::*; + use std::collections::HashMap; + use std::sync::Mutex; // ── Slash command types ──────────────────────────────────────────── diff --git a/server/src/chat/transport/slack/mod.rs b/server/src/chat/transport/slack/mod.rs index 2aee55fe..73362a67 100644 --- a/server/src/chat/transport/slack/mod.rs +++ b/server/src/chat/transport/slack/mod.rs @@ -201,19 +201,19 @@ pub async fn slash_command_receive( // Build a synthetic message that the command registry can parse. // The format is " " so strip_bot_mention + dispatch works. let synthetic_message = if payload.text.is_empty() { - format!("{} {keyword}", ctx.bot_name) + format!("{} {keyword}", ctx.services.bot_name) } else { - format!("{} {keyword} {}", ctx.bot_name, payload.text) + format!("{} {keyword} {}", ctx.services.bot_name, payload.text) }; use crate::chat::commands::{CommandDispatch, try_handle_command}; let dispatch = CommandDispatch { - bot_name: &ctx.bot_name, - bot_user_id: &ctx.bot_user_id, - project_root: &ctx.project_root, - agents: &ctx.agents, - ambient_rooms: &ctx.ambient_rooms, + bot_name: &ctx.services.bot_name, + bot_user_id: &ctx.services.bot_user_id, + project_root: &ctx.services.project_root, + agents: &ctx.services.agents, + ambient_rooms: &ctx.services.ambient_rooms, room_id: &payload.channel_id, }; diff --git a/server/src/main.rs b/server/src/main.rs index 21c6b77a..e2ed2770 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -565,7 +565,6 @@ async fn main() -> Result<(), std::io::Error> { // bundle (AppContext + Matrix bot) and the webhook-based transports. let perm_rx = Arc::new(tokio::sync::Mutex::new(perm_rx)); let perm_rx_for_whatsapp = Arc::clone(&perm_rx); - let perm_rx_for_slack = Arc::clone(&perm_rx); let perm_rx_for_discord = Arc::clone(&perm_rx); // Capture project root, agents Arc, and reconciliation sender before ctx @@ -666,30 +665,17 @@ async fn main() -> Result<(), std::io::Error> { let transport = Arc::new(chat::transport::slack::SlackTransport::new( cfg.slack_bot_token.clone().unwrap_or_default(), )); - let bot_name = cfg - .display_name - .clone() - .unwrap_or_else(|| "Assistant".to_string()); let root = startup_root.clone().unwrap(); let history = chat::transport::slack::load_slack_history(&root); let channel_ids: std::collections::HashSet = cfg.slack_channel_ids.iter().cloned().collect(); Arc::new(chat::transport::slack::SlackWebhookContext { + services: Arc::clone(&services), signing_secret: cfg.slack_signing_secret.clone().unwrap_or_default(), transport, - project_root: root, - agents: Arc::clone(&startup_agents), - bot_name, - bot_user_id: "slack-bot".to_string(), - ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), history: std::sync::Arc::new(tokio::sync::Mutex::new(history)), history_size: cfg.history_size, channel_ids, - perm_rx: perm_rx_for_slack, - pending_perm_replies: Arc::new(tokio::sync::Mutex::new( - std::collections::HashMap::new(), - )), - permission_timeout_secs: cfg.permission_timeout_secs, }) }); @@ -744,7 +730,7 @@ async fn main() -> Result<(), std::io::Error> { Some(Arc::new(BotShutdownNotifier::new( Arc::clone(&ctx.transport) as Arc, channels, - ctx.bot_name.clone(), + ctx.services.bot_name.clone(), ))) } else if let Some(ref ctx) = discord_ctx { let channels: Vec = ctx.channel_ids.iter().cloned().collect(); @@ -785,7 +771,7 @@ async fn main() -> Result<(), std::io::Error> { } if let Some(ref ctx) = slack_ctx { let transport = Arc::clone(&ctx.transport) as Arc; - let bot_name = ctx.bot_name.clone(); + let bot_name = ctx.services.bot_name.clone(); let channels: Vec = ctx.channel_ids.iter().cloned().collect(); tokio::spawn(async move { if channels.is_empty() {