crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

huskies: merge 638_refactor_migrate_discord_transport_to_services_bundle

Browse Source
This commit is contained in:
dave
2026-04-25 19:28:50 +00:00
parent 33cb2bed3e
commit e4dd4bbe2c
3 changed files with 68 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/chat/transport/discord/commands.rs
+59 -68
View File
@@ -1,15 +1,13 @@
//! Discord incoming message dispatch and command handling. //! Discord incoming message dispatch and command handling.
use std::collections::{HashMap, HashSet}; use std::collections::HashSet;
use std::path::PathBuf; use std::sync::Arc;
use std::sync::{Arc, Mutex};
use tokio::sync::{Mutex as TokioMutex, oneshot};
use crate::agents::AgentPool;
use crate::chat::ChatTransport; use crate::chat::ChatTransport;
use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation}; use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation};
use crate::chat::util::is_permission_approval; 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; use crate::slog;
use super::format::markdown_to_discord; use super::format::markdown_to_discord;
@@ -21,15 +19,10 @@ use super::meta::DiscordTransport;
/// Shared context for the Discord bot, used by both the Gateway listener /// Shared context for the Discord bot, used by both the Gateway listener
/// and any future webhook handlers. /// and any future webhook handlers.
pub struct DiscordContext { pub struct DiscordContext {
/// Shared services bundle (project root, agent pool, bot identity, permissions).
pub services: Arc<Services>,
pub bot_token: String, pub bot_token: String,
pub transport: Arc<DiscordTransport>, pub transport: Arc<DiscordTransport>,
pub project_root: PathBuf,
pub agents: Arc<AgentPool>,
pub bot_name: String,
/// The bot's Discord user ID (set dynamically from READY event, but
/// also stored here for command dispatch).
pub bot_user_id: String,
pub ambient_rooms: Arc<Mutex<HashSet<String>>>,
/// Per-channel conversation history for LLM passthrough. /// Per-channel conversation history for LLM passthrough.
pub history: DiscordConversationHistory, pub history: DiscordConversationHistory,
/// Maximum number of conversation entries to keep per channel. /// Maximum number of conversation entries to keep per channel.
@@ -39,12 +32,6 @@ pub struct DiscordContext {
/// Allowed Discord user IDs. When non-empty, only listed users can /// Allowed Discord user IDs. When non-empty, only listed users can
/// interact with the bot. When empty, all users are allowed. /// interact with the bot. When empty, all users are allowed.
pub allowed_users: HashSet<String>, pub allowed_users: HashSet<String>,
/// Permission requests from the MCP `prompt_permission` tool arrive here.
pub perm_rx: Arc<TokioMutex<tokio::sync::mpsc::UnboundedReceiver<PermissionForward>>>,
/// Pending permission replies keyed by channel ID.
pub pending_perm_replies: Arc<TokioMutex<HashMap<String, oneshot::Sender<PermissionDecision>>>>,
/// Seconds before an unanswered permission prompt is auto-denied.
pub permission_timeout_secs: u64,
} }
// ── Incoming message dispatch ─────────────────────────────────────────── // ── Incoming message dispatch ───────────────────────────────────────────
@@ -60,7 +47,7 @@ pub(super) async fn handle_incoming_message(
// If there is a pending permission prompt for this channel, interpret the // If there is a pending permission prompt for this channel, interpret the
// message as a yes/no response. // message as a yes/no response.
{ {
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) { if let Some(tx) = pending.remove(channel) {
let decision = if is_permission_approval(message) { let decision = if is_permission_approval(message) {
PermissionDecision::Approve PermissionDecision::Approve
@@ -80,11 +67,11 @@ pub(super) async fn handle_incoming_message(
} }
let dispatch = CommandDispatch { let dispatch = CommandDispatch {
bot_name: &ctx.bot_name, bot_name: &ctx.services.bot_name,
bot_user_id: &ctx.bot_user_id, bot_user_id: &ctx.services.bot_user_id,
project_root: &ctx.project_root, project_root: &ctx.services.project_root,
agents: &ctx.agents, agents: &ctx.services.agents,
ambient_rooms: &ctx.ambient_rooms, ambient_rooms: &ctx.services.ambient_rooms,
room_id: channel, room_id: channel,
}; };
@@ -100,8 +87,8 @@ pub(super) async fn handle_incoming_message(
// Check for async commands (htop, delete). // Check for async commands (htop, delete).
if let Some(htop_cmd) = crate::chat::transport::matrix::htop::extract_htop_command( if let Some(htop_cmd) = crate::chat::transport::matrix::htop::extract_htop_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) { ) {
use crate::chat::transport::matrix::htop::HtopCommand; use crate::chat::transport::matrix::htop::HtopCommand;
slog!("[discord] Handling htop command from {user} in {channel}"); slog!("[discord] Handling htop command from {user} in {channel}");
@@ -114,7 +101,7 @@ pub(super) async fn handle_incoming_message(
} }
HtopCommand::Start { duration_secs } => { HtopCommand::Start { duration_secs } => {
let snapshot = crate::chat::transport::matrix::htop::build_htop_message( let snapshot = crate::chat::transport::matrix::htop::build_htop_message(
&ctx.agents, &ctx.services.agents,
0, 0,
duration_secs, duration_secs,
); );
@@ -127,7 +114,7 @@ pub(super) async fn handle_incoming_message(
} }
}; };
let transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>; let transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>;
let agents = Arc::clone(&ctx.agents); let agents = Arc::clone(&ctx.services.agents);
let ch = channel.to_string(); let ch = channel.to_string();
tokio::spawn(async move { tokio::spawn(async move {
let interval = std::time::Duration::from_secs(2); let interval = std::time::Duration::from_secs(2);
@@ -153,22 +140,22 @@ pub(super) async fn handle_incoming_message(
if let Some(del_cmd) = crate::chat::transport::matrix::delete::extract_delete_command( if let Some(del_cmd) = crate::chat::transport::matrix::delete::extract_delete_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) { ) {
let response = match del_cmd { let response = match del_cmd {
crate::chat::transport::matrix::delete::DeleteCommand::Delete { story_number } => { crate::chat::transport::matrix::delete::DeleteCommand::Delete { story_number } => {
slog!("[discord] Handling delete command from {user}: story {story_number}"); slog!("[discord] Handling delete command from {user}: story {story_number}");
crate::chat::transport::matrix::delete::handle_delete( crate::chat::transport::matrix::delete::handle_delete(
&ctx.bot_name, &ctx.services.bot_name,
&story_number, &story_number,
&ctx.project_root, &ctx.services.project_root,
&ctx.agents, &ctx.services.agents,
) )
.await .await
} }
crate::chat::transport::matrix::delete::DeleteCommand::BadArgs => { crate::chat::transport::matrix::delete::DeleteCommand::BadArgs => {
format!("Usage: `{} delete <number>`", ctx.bot_name) format!("Usage: `{} delete <number>`", ctx.services.bot_name)
} }
}; };
let response = markdown_to_discord(&response); let response = markdown_to_discord(&response);
@@ -178,8 +165,8 @@ pub(super) async fn handle_incoming_message(
if crate::chat::transport::matrix::rebuild::extract_rebuild_command( if crate::chat::transport::matrix::rebuild::extract_rebuild_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) )
.is_some() .is_some()
{ {
@@ -187,9 +174,9 @@ pub(super) async fn handle_incoming_message(
let ack = "Rebuilding server… this may take a moment."; let ack = "Rebuilding server… this may take a moment.";
let _ = ctx.transport.send_message(channel, ack, "").await; let _ = ctx.transport.send_message(channel, ack, "").await;
let response = crate::chat::transport::matrix::rebuild::handle_rebuild( let response = crate::chat::transport::matrix::rebuild::handle_rebuild(
&ctx.bot_name, &ctx.services.bot_name,
&ctx.project_root, &ctx.services.project_root,
&ctx.agents, &ctx.services.agents,
) )
.await; .await;
let response = markdown_to_discord(&response); let response = markdown_to_discord(&response);
@@ -199,8 +186,8 @@ pub(super) async fn handle_incoming_message(
if let Some(rmtree_cmd) = crate::chat::transport::matrix::rmtree::extract_rmtree_command( if let Some(rmtree_cmd) = crate::chat::transport::matrix::rmtree::extract_rmtree_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) { ) {
let response = match rmtree_cmd { let response = match rmtree_cmd {
crate::chat::transport::matrix::rmtree::RmtreeCommand::Rmtree { story_number } => { crate::chat::transport::matrix::rmtree::RmtreeCommand::Rmtree { story_number } => {
@@ -208,15 +195,15 @@ pub(super) async fn handle_incoming_message(
"[discord] Handling rmtree command from {user} in {channel}: story {story_number}" "[discord] Handling rmtree command from {user} in {channel}: story {story_number}"
); );
crate::chat::transport::matrix::rmtree::handle_rmtree( crate::chat::transport::matrix::rmtree::handle_rmtree(
&ctx.bot_name, &ctx.services.bot_name,
&story_number, &story_number,
&ctx.project_root, &ctx.services.project_root,
&ctx.agents, &ctx.services.agents,
) )
.await .await
} }
crate::chat::transport::matrix::rmtree::RmtreeCommand::BadArgs => { crate::chat::transport::matrix::rmtree::RmtreeCommand::BadArgs => {
format!("Usage: `{} rmtree <number>`", ctx.bot_name) format!("Usage: `{} rmtree <number>`", ctx.services.bot_name)
} }
}; };
let response = markdown_to_discord(&response); let response = markdown_to_discord(&response);
@@ -226,8 +213,8 @@ pub(super) async fn handle_incoming_message(
if crate::chat::transport::matrix::reset::extract_reset_command( if crate::chat::transport::matrix::reset::extract_reset_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) )
.is_some() .is_some()
{ {
@@ -239,7 +226,7 @@ pub(super) async fn handle_incoming_message(
.or_insert_with(RoomConversation::default); .or_insert_with(RoomConversation::default);
conv.session_id = None; conv.session_id = None;
conv.entries.clear(); conv.entries.clear();
save_discord_history(&ctx.project_root, &guard); save_discord_history(&ctx.services.project_root, &guard);
} }
let _ = ctx let _ = ctx
.transport .transport
@@ -250,8 +237,8 @@ pub(super) async fn handle_incoming_message(
if let Some(start_cmd) = crate::chat::transport::matrix::start::extract_start_command( if let Some(start_cmd) = crate::chat::transport::matrix::start::extract_start_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) { ) {
let response = match start_cmd { let response = match start_cmd {
crate::chat::transport::matrix::start::StartCommand::Start { crate::chat::transport::matrix::start::StartCommand::Start {
@@ -262,16 +249,16 @@ pub(super) async fn handle_incoming_message(
"[discord] Handling start command from {user} in {channel}: story {story_number}" "[discord] Handling start command from {user} in {channel}: story {story_number}"
); );
crate::chat::transport::matrix::start::handle_start( crate::chat::transport::matrix::start::handle_start(
&ctx.bot_name, &ctx.services.bot_name,
&story_number, &story_number,
agent_hint.as_deref(), agent_hint.as_deref(),
&ctx.project_root, &ctx.services.project_root,
&ctx.agents, &ctx.services.agents,
) )
.await .await
} }
crate::chat::transport::matrix::start::StartCommand::BadArgs => { crate::chat::transport::matrix::start::StartCommand::BadArgs => {
format!("Usage: `{} start <number>`", ctx.bot_name) format!("Usage: `{} start <number>`", ctx.services.bot_name)
} }
}; };
let response = markdown_to_discord(&response); let response = markdown_to_discord(&response);
@@ -281,8 +268,8 @@ pub(super) async fn handle_incoming_message(
if let Some(assign_cmd) = crate::chat::transport::matrix::assign::extract_assign_command( if let Some(assign_cmd) = crate::chat::transport::matrix::assign::extract_assign_command(
message, message,
&ctx.bot_name, &ctx.services.bot_name,
&ctx.bot_user_id, &ctx.services.bot_user_id,
) { ) {
let response = match assign_cmd { let response = match assign_cmd {
crate::chat::transport::matrix::assign::AssignCommand::Assign { crate::chat::transport::matrix::assign::AssignCommand::Assign {
@@ -293,16 +280,16 @@ pub(super) async fn handle_incoming_message(
"[discord] Handling assign command from {user} in {channel}: story {story_number} model {model}" "[discord] Handling assign command from {user} in {channel}: story {story_number} model {model}"
); );
crate::chat::transport::matrix::assign::handle_assign( crate::chat::transport::matrix::assign::handle_assign(
&ctx.bot_name, &ctx.services.bot_name,
&story_number, &story_number,
&model, &model,
&ctx.project_root, &ctx.services.project_root,
&ctx.agents, &ctx.services.agents,
) )
.await .await
} }
crate::chat::transport::matrix::assign::AssignCommand::BadArgs => { crate::chat::transport::matrix::assign::AssignCommand::BadArgs => {
format!("Usage: `{} assign <number> <model>`", ctx.bot_name) format!("Usage: `{} assign <number> <model>`", ctx.services.bot_name)
} }
}; };
let response = markdown_to_discord(&response); let response = markdown_to_discord(&response);
@@ -328,7 +315,7 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
guard.get(channel).and_then(|conv| conv.session_id.clone()) 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!( let prompt = format!(
"[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}" "[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}"
); );
@@ -358,7 +345,7 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
let sent_any_chunk = Arc::new(AtomicBool::new(false)); let sent_any_chunk = Arc::new(AtomicBool::new(false));
let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk); 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( let chat_fut = provider.chat_stream(
&prompt, &prompt,
&project_root_str, &project_root_str,
@@ -380,7 +367,7 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
tokio::pin!(chat_fut); tokio::pin!(chat_fut);
// Lock the permission receiver for the duration of this chat session. // 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 { let result = loop {
tokio::select! { tokio::select! {
@@ -396,16 +383,17 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
let formatted = markdown_to_discord(&prompt_msg); let formatted = markdown_to_discord(&prompt_msg);
let _ = ctx.transport.send_message(channel, &formatted, "").await; let _ = ctx.transport.send_message(channel, &formatted, "").await;
ctx.pending_perm_replies ctx.services
.pending_perm_replies
.lock() .lock()
.await .await
.insert(channel.to_string(), perm_fwd.response_tx); .insert(channel.to_string(), perm_fwd.response_tx);
// Spawn a timeout task: auto-deny if the user does not respond. // 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_channel = channel.to_string();
let timeout_transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>; let timeout_transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>;
let timeout_secs = ctx.permission_timeout_secs; let timeout_secs = ctx.services.permission_timeout_secs;
tokio::spawn(async move { tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await;
if let Some(tx) = pending.lock().await.remove(&timeout_channel) { if let Some(tx) = pending.lock().await.remove(&timeout_channel) {
@@ -486,7 +474,7 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
conv.entries.drain(..excess); conv.entries.drain(..excess);
} }
save_discord_history(&ctx.project_root, &guard); save_discord_history(&ctx.services.project_root, &guard);
} }
} }
@@ -495,6 +483,9 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
use tokio::sync::Mutex as TokioMutex;
fn test_agents() -> Arc<crate::agents::AgentPool> { fn test_agents() -> Arc<crate::agents::AgentPool> {
Arc::new(crate::agents::AgentPool::new_test(3000)) Arc::new(crate::agents::AgentPool::new_test(3000))
server/src/chat/transport/discord/gateway.rs
+6 -1
View File
@@ -350,7 +350,12 @@ fn dispatch_message(
.as_ref() .as_ref()
.is_some_and(|bid| msg.mentions.iter().any(|m| m.id == *bid)); .is_some_and(|bid| msg.mentions.iter().any(|m| m.id == *bid));
let in_ambient = ctx.ambient_rooms.lock().unwrap().contains(&msg.channel_id); let in_ambient = ctx
.services
.ambient_rooms
.lock()
.unwrap()
.contains(&msg.channel_id);
if !bot_mentioned && !in_ambient { if !bot_mentioned && !in_ambient {
return; return;
server/src/main.rs
+3 -17
View File
@@ -565,7 +565,6 @@ async fn main() -> Result<(), std::io::Error> {
// bundle (AppContext + Matrix bot) and the webhook-based transports. // bundle (AppContext + Matrix bot) and the webhook-based transports.
let perm_rx = Arc::new(tokio::sync::Mutex::new(perm_rx)); let perm_rx = Arc::new(tokio::sync::Mutex::new(perm_rx));
let perm_rx_for_whatsapp = Arc::clone(&perm_rx); let perm_rx_for_whatsapp = Arc::clone(&perm_rx);
let perm_rx_for_discord = Arc::clone(&perm_rx);
// Capture project root, agents Arc, and reconciliation sender before ctx // Capture project root, agents Arc, and reconciliation sender before ctx
// is consumed by build_routes. // is consumed by build_routes.
@@ -688,10 +687,6 @@ async fn main() -> Result<(), std::io::Error> {
let transport = Arc::new(chat::transport::discord::DiscordTransport::new( let transport = Arc::new(chat::transport::discord::DiscordTransport::new(
cfg.discord_bot_token.clone().unwrap_or_default(), cfg.discord_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 root = startup_root.clone().unwrap();
let history = chat::transport::discord::load_discord_history(&root); let history = chat::transport::discord::load_discord_history(&root);
let channel_ids: std::collections::HashSet<String> = let channel_ids: std::collections::HashSet<String> =
@@ -699,22 +694,13 @@ async fn main() -> Result<(), std::io::Error> {
let allowed_users: std::collections::HashSet<String> = let allowed_users: std::collections::HashSet<String> =
cfg.discord_allowed_users.iter().cloned().collect(); cfg.discord_allowed_users.iter().cloned().collect();
Arc::new(chat::transport::discord::DiscordContext { Arc::new(chat::transport::discord::DiscordContext {
services: Arc::clone(&services),
bot_token: cfg.discord_bot_token.clone().unwrap_or_default(), bot_token: cfg.discord_bot_token.clone().unwrap_or_default(),
transport, transport,
project_root: root,
agents: Arc::clone(&startup_agents),
bot_name,
bot_user_id: "discord-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: std::sync::Arc::new(tokio::sync::Mutex::new(history)),
history_size: cfg.history_size, history_size: cfg.history_size,
channel_ids, channel_ids,
allowed_users, allowed_users,
perm_rx: perm_rx_for_discord,
pending_perm_replies: Arc::new(tokio::sync::Mutex::new(
std::collections::HashMap::new(),
)),
permission_timeout_secs: cfg.permission_timeout_secs,
}) })
}); });
@@ -737,7 +723,7 @@ async fn main() -> Result<(), std::io::Error> {
Some(Arc::new(BotShutdownNotifier::new( Some(Arc::new(BotShutdownNotifier::new(
Arc::clone(&ctx.transport) as Arc<dyn crate::chat::ChatTransport>, Arc::clone(&ctx.transport) as Arc<dyn crate::chat::ChatTransport>,
channels, channels,
ctx.bot_name.clone(), ctx.services.bot_name.clone(),
))) )))
} else { } else {
None None
@@ -783,7 +769,7 @@ async fn main() -> Result<(), std::io::Error> {
} }
if let Some(ref ctx) = discord_ctx { if let Some(ref ctx) = discord_ctx {
let transport = Arc::clone(&ctx.transport) as Arc<dyn crate::chat::ChatTransport>; let transport = Arc::clone(&ctx.transport) as Arc<dyn crate::chat::ChatTransport>;
let bot_name = ctx.bot_name.clone(); let bot_name = ctx.services.bot_name.clone();
let channels: Vec<String> = ctx.channel_ids.iter().cloned().collect(); let channels: Vec<String> = ctx.channel_ids.iter().cloned().collect();
tokio::spawn(async move { tokio::spawn(async move {
if channels.is_empty() { if channels.is_empty() {