From e601c76dd062312685ba1ec7f75a48afed53e120 Mon Sep 17 00:00:00 2001 From: chaodu-agent <274062505+chaodu-agent@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:49:56 +0000 Subject: [PATCH] fix(discord): keep multibot threads responsive --- crates/openab-core/src/adapter.rs | 344 +++++++++++++++++------------- crates/openab-core/src/discord.rs | 272 ++++++++++++++++++++++- src/main.rs | 1 + 3 files changed, 459 insertions(+), 158 deletions(-) diff --git a/crates/openab-core/src/adapter.rs b/crates/openab-core/src/adapter.rs index 844cefbab..31a32b0a6 100644 --- a/crates/openab-core/src/adapter.rs +++ b/crates/openab-core/src/adapter.rs @@ -1,6 +1,6 @@ use anyhow::Result; use async_trait::async_trait; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use std::sync::Arc; use tracing::{error, warn}; @@ -13,12 +13,35 @@ use crate::reactions::StatusReactionController; // --- Output directive parsing --- -/// Parsed directives from agent output header block. -/// Consecutive `[[key:value]]` lines at the start of output are directives. +/// A validated bot-to-bot handoff request. Presentation updates must never be +/// interpreted as this control-plane message; only a complete turn may create it. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct HandoffRequest { + pub schema: String, + pub source_bot_id: String, + pub target_bot_id: String, + pub event_id: String, + pub origin_channel_id: String, + pub origin_thread_id: Option, + pub origin_message_id: Option, + pub expires_at: u64, + pub hop_count: u8, + pub payload: HandoffPayload, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct HandoffPayload { + pub text: String, +} + +pub const MULTIBOT_HANDOFF_SCHEMA: &str = "openab.multibot.handoff.v1"; + #[derive(Default, Debug)] pub struct OutputDirectives { /// Message ID to reply to (Discord: message_reference) pub reply_to: Option, + /// Target Discord bot ID for a complete control-plane handoff. + pub handoff_target_bot_id: Option, } /// Parse `[[key:value]]` directives from the beginning of agent output. @@ -43,6 +66,19 @@ pub fn parse_output_directives(content: &str) -> (OutputDirectives, String) { directives.reply_to = Some(v.to_string()); } } + "handoff" => { + let v = value.trim(); + // Discord snowflake IDs are decimal u64 values. Keep the + // directive strict so arbitrary text can never become a + // routing target. + if !v.is_empty() + && v.len() <= 20 + && v.chars().all(|c| c.is_ascii_digit()) + && v.parse::().is_ok() + { + directives.handoff_target_bot_id = Some(v.to_string()); + } + } _ => { tracing::debug!(key = key.trim(), "unknown output directive ignored"); } @@ -423,6 +459,26 @@ pub trait ChatAdapter: Send + Sync + 'static { /// response may stream, but subsequent ones will correctly use send-once. fn use_streaming(&self, other_bot_present: bool) -> bool; + /// Set the identity used as the source of control-plane handoffs. + /// Platforms without bot-to-bot control messages may ignore this. + fn set_bot_identity(&self, _bot_id: &str) {} + + /// Return the configured source identity for a control-plane handoff. + fn bot_identity(&self) -> Option { + None + } + + /// Send one complete, structured bot-to-bot handoff. The default is fail-closed: + /// a platform must explicitly implement this instead of accidentally routing + /// an incomplete presentation message as control input. + async fn send_handoff( + &self, + _channel: &ChannelRef, + _request: &HandoffRequest, + ) -> Result { + Err(anyhow::anyhow!("structured handoff not supported")) + } + /// Whether to send the "…" placeholder message before streaming starts. /// Default: true. Platforms using drafts (e.g. Telegram Rich Messages) can /// return false to suppress the placeholder. @@ -933,12 +989,13 @@ impl AdapterRouter { } } } else if let Some(tx) = &buf_tx { - let _ = tx.send(compose_display( + let display = compose_display( &tool_lines, &text_buf, true, tool_display, - )); + ); + let _ = tx.send(presentation_text(adapter.platform(), &display)); } } AcpEvent::Thinking => { @@ -982,12 +1039,13 @@ impl AdapterRouter { } // Post+edit live update (no-op under native streaming: buf_tx is None). if let Some(tx) = &buf_tx { - let _ = tx.send(compose_display( + let display = compose_display( &tool_lines, &text_buf, true, tool_display, - )); + ); + let _ = tx.send(presentation_text(adapter.platform(), &display)); } } AcpEvent::ToolDone { id, title, status } => { @@ -1027,12 +1085,13 @@ impl AdapterRouter { }); } if let Some(tx) = &buf_tx { - let _ = tx.send(compose_display( + let display = compose_display( &tool_lines, &text_buf, true, tool_display, - )); + ); + let _ = tx.send(presentation_text(adapter.platform(), &display)); } } AcpEvent::ConfigUpdate { options } => { @@ -1105,17 +1164,11 @@ impl AdapterRouter { }; let final_content = markdown::convert_tables(&final_content, table_mode); - let chunks = if adapter.platform() == "discord" { - let mentions = extract_mentions(&final_content); - let mention_reserve = mention_footer_len(&mentions); - let chunks = format::split_message( - &final_content, - message_limit.saturating_sub(mention_reserve), - ); - propagate_mentions_to_chunks(chunks, &mentions, message_limit) - } else { - format::split_message(&final_content, message_limit) - }; + // Final presentation is always inert. A complete handoff, if + // requested, uses the unsanitized payload below and is sent + // through the explicit control-plane API exactly once. + let presentation_content = presentation_text(adapter.platform(), &final_content); + let chunks = format::split_message(&presentation_content, message_limit); // Track delivery health across all final write paths. Any failure // here means the user's view is incomplete; we propagate Err at the // end of the closure so dispatch surfaces set_error (❌) instead of @@ -1125,7 +1178,82 @@ impl AdapterRouter { if assistant_status { let _ = adapter.set_status(&thread_channel, "").await; } - if native { + if let Some(target_bot_id) = directives.handoff_target_bot_id.clone() { + // A handoff is a control-plane operation: it is emitted only + // after the ACP turn is complete, never from a stream delta. + let handoff_result = async { + if adapter.platform() != "discord" { + return Err(anyhow::anyhow!( + "structured handoff is only supported on Discord" + )); + } + let source_bot_id = adapter.bot_identity().ok_or_else(|| { + anyhow::anyhow!("Discord bot identity unavailable for handoff") + })?; + let payload_text = sanitize_discord_mentions(&final_content); + let request = HandoffRequest { + schema: MULTIBOT_HANDOFF_SCHEMA.to_string(), + source_bot_id, + target_bot_id, + event_id: uuid::Uuid::new_v4().to_string(), + origin_channel_id: thread_channel.channel_id.clone(), + origin_thread_id: thread_channel + .parent_id + .as_ref() + .map(|_| thread_channel.channel_id.clone()), + origin_message_id: None, + expires_at: chrono::Utc::now().timestamp().max(0) as u64 + 300, + hop_count: 0, + payload: HandoffPayload { text: payload_text }, + }; + adapter.send_handoff(&thread_channel, &request).await + } + .await; + + match handoff_result { + Ok(_) => { + if let Some(msg) = placeholder_msg { + let _ = adapter.delete_message(&msg).await; + } + } + Err(e) => { + tracing::error!(error = ?e, "structured handoff failed closed"); + delivery_failed = true; + // Keep the human-facing result visible, but it is + // sanitized and therefore cannot become control input. + if let Some(msg) = placeholder_msg { + if let Some(first) = chunks.first() { + if let Err(edit_error) = adapter.edit_message(&msg, first).await { + tracing::warn!(error = ?edit_error, "handoff fallback placeholder edit failed"); + delivery_failed = true; + if let Err(send_error) = + adapter.send_message(&thread_channel, first).await + { + tracing::warn!(error = ?send_error, "handoff fallback first send failed"); + } + } + for chunk in chunks.iter().skip(1) { + if let Err(send_error) = + adapter.send_message(&thread_channel, chunk).await + { + tracing::warn!(error = ?send_error, "handoff fallback overflow send failed"); + delivery_failed = true; + } + } + } + } else { + for chunk in &chunks { + if let Err(send_error) = + adapter.send_message(&thread_channel, chunk).await + { + tracing::warn!(error = ?send_error, "handoff fallback send failed"); + delivery_failed = true; + } + } + } + } + } + } else if native { if let Some(msg) = &native_msg { if !native_pending.is_empty() { if let Err(e) = @@ -1214,34 +1342,6 @@ impl AdapterRouter { tracing::warn!(error = ?e, platform = %thread_channel.platform, message_id = %msg.message_id, "delete placeholder failed; placeholder will remain visible"); } } - } else if adapter.platform() == "discord" - && contains_bot_mention(&final_content) - { - // Discord-specific: bot mention detected. Delete placeholder - // and send as new message so Discord emits MESSAGE_CREATE — - // otherwise the mentioned bot won't receive the gateway - // event since MESSAGE_UPDATE skips notifications (#1110). - let mut send_ok = false; - if let Some(first) = chunks.first() { - match adapter.send_message(&thread_channel, first).await { - Ok(_) => { - send_ok = true; - } - Err(e) => { - tracing::warn!(error = ?e, platform = %thread_channel.platform, message_id = %msg.message_id, "discord bot-mention first chunk send failed"); - delivery_failed = true; - } - } - } - for chunk in chunks.iter().skip(1) { - if let Err(e) = adapter.send_message(&thread_channel, chunk).await { - tracing::warn!(error = ?e, platform = %thread_channel.platform, message_id = %msg.message_id, "streaming overflow chunk send failed"); - delivery_failed = true; - } - } - if send_ok { - let _ = adapter.delete_message(&msg).await; - } } else { // Normal streaming: edit first chunk into placeholder, send rest. // If placeholder is a dummy "draft" ref (no real message), send as @@ -1331,16 +1431,12 @@ impl AdapterRouter { } } -/// Returns true if `content` contains a Discord user/bot mention (`<@123>`, `<@!123>`) -/// or a role mention (`<@&123>`). -/// Used to detect cross-bot mentions so the streaming path can switch from -/// edit (MESSAGE_UPDATE, no mention notification) to delete+send (MESSAGE_CREATE). +#[cfg(test)] fn contains_bot_mention(content: &str) -> bool { let mut i = 0; let bytes = content.as_bytes(); while i + 2 < bytes.len() { if bytes[i] == b'<' && bytes[i + 1] == b'@' { - // Skip optional '!' (nickname mention) or '&' (role mention) let start = if i + 2 < bytes.len() && (bytes[i + 2] == b'!' || bytes[i + 2] == b'&') { @@ -1363,6 +1459,16 @@ fn contains_bot_mention(content: &str) -> bool { false } +/// Neutralize Discord mentions in human-facing presentation text. This is applied +/// to every cumulative streaming edit and every ordinary final send, so a partial +/// output or an overflow chunk can never become a bot-to-bot control event. +fn sanitize_discord_mentions(content: &str) -> String { + content + .replace("<@", "<\u{200b}@") + .replace("@everyone", "@\u{200b}everyone") + .replace("@here", "@\u{200b}here") +} + /// Flatten a tool-call title into a single line safe for inline-code spans. fn sanitize_title(title: &str) -> String { title @@ -1371,6 +1477,14 @@ fn sanitize_title(title: &str) -> String { .replace('`', "'") } +fn presentation_text(platform: &str, content: &str) -> String { + if platform == "discord" { + sanitize_discord_mentions(content) + } else { + content.to_string() + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ToolState { Running, @@ -1590,98 +1704,6 @@ fn compose_display( out } -fn extract_mentions(content: &str) -> Vec { - let mut mentions = Vec::new(); - let mut in_fence = false; - - for line in content.split('\n') { - if line.starts_with("```") { - in_fence = !in_fence; - continue; - } - if in_fence { - continue; - } - - let bytes = line.as_bytes(); - let mut i = 0; - while i + 2 < bytes.len() { - if bytes[i] == b'<' && bytes[i + 1] == b'@' { - let (prefix_end, is_role) = if i + 2 < bytes.len() && bytes[i + 2] == b'&' { - (i + 3, true) - } else if i + 2 < bytes.len() && bytes[i + 2] == b'!' { - (i + 3, false) - } else { - (i + 2, false) - }; - if prefix_end < bytes.len() && bytes[prefix_end].is_ascii_digit() { - if let Some(end) = line[prefix_end..].find('>') { - if line[prefix_end..prefix_end + end] - .chars() - .all(|c| c.is_ascii_digit()) - { - let uid = &line[prefix_end..prefix_end + end]; - let normalized = if is_role { - format!("<@&{uid}>") - } else { - format!("<@{uid}>") - }; - if !mentions.contains(&normalized) { - mentions.push(normalized); - } - i = prefix_end + end + 1; - continue; - } - } - } - i = prefix_end; - } else { - i += 1; - } - } - } - mentions -} - -fn mention_footer_len(mentions: &[String]) -> usize { - if mentions.is_empty() { - return 0; - } - 1 + mentions.iter().map(|m| m.len()).sum::() + mentions.len().saturating_sub(1) -} - -fn propagate_mentions_to_chunks( - chunks: Vec, - mentions: &[String], - limit: usize, -) -> Vec { - if mentions.is_empty() || chunks.len() <= 1 { - return chunks; - } - chunks - .into_iter() - .map(|chunk| { - let missing: Vec<&String> = mentions - .iter() - .filter(|m| !chunk.contains(m.as_str())) - .collect(); - if missing.is_empty() { - chunk - } else { - let footer = format!( - "\n{}", - missing.iter().map(|m| m.as_str()).collect::>().join(" ") - ); - if chunk.chars().count() + footer.chars().count() <= limit { - format!("{chunk}{footer}") - } else { - chunk - } - } - }) - .collect() -} - #[cfg(test)] mod tests { use super::*; @@ -2272,7 +2294,7 @@ mod tests { #[cfg(test)] mod directive_tests { use super::parse_output_directives; - use super::{classify_empty_turn, SILENT_FAILURE_MSG}; + use super::{classify_empty_turn, presentation_text, sanitize_discord_mentions, SILENT_FAILURE_MSG}; use crate::acp::TurnResult; #[test] @@ -2426,6 +2448,34 @@ mod directive_tests { assert_eq!(content, "看看 [[這個]] 怎麼樣"); } + #[test] + fn parse_handoff_directive_extracts_valid_target() { + let (directives, content) = parse_output_directives( + "[[handoff:1490365068863606784]]\nComplete request", + ); + assert_eq!(directives.handoff_target_bot_id.as_deref(), Some("1490365068863606784")); + assert_eq!(content, "Complete request"); + } + + #[test] + fn parse_handoff_directive_rejects_non_numeric_target() { + let (directives, content) = parse_output_directives("[[handoff:bot-b]]\nRequest"); + assert_eq!(directives.handoff_target_bot_id, None); + assert_eq!(content, "Request"); + } + + #[test] + fn discord_presentation_neutralizes_all_mention_forms() { + let input = "<@123> <@!456> <@&789> @everyone @here"; + let output = sanitize_discord_mentions(input); + assert!(!output.contains("<@123>")); + assert!(!output.contains("<@!456>")); + assert!(!output.contains("<@&789>")); + assert!(!output.contains("@everyone")); + assert!(!output.contains("@here")); + assert_eq!(presentation_text("slack", input), input); + } + // --- classify_empty_turn: adapter-level finalization tests --- #[test] diff --git a/crates/openab-core/src/discord.rs b/crates/openab-core/src/discord.rs index 609bd7459..fb02cebee 100644 --- a/crates/openab-core/src/discord.rs +++ b/crates/openab-core/src/discord.rs @@ -1,6 +1,9 @@ use crate::acp::protocol::{ConfigOption, UsageReport}; use crate::acp::ContentBlock; -use crate::adapter::{AdapterRouter, ChannelRef, ChatAdapter, MessageRef, SenderContext}; +use crate::adapter::{ + AdapterRouter, ChannelRef, ChatAdapter, HandoffRequest, MessageRef, SenderContext, + MULTIBOT_HANDOFF_SCHEMA, +}; use crate::bot_turns::{BotTurnTracker, TurnAction, TurnSeverity, BOT_TURN_LIMIT_WARNING_PREFIX}; use crate::config::{AllowBots, AllowUsers, SttConfig}; use crate::dispatch::DispatchTarget; @@ -10,9 +13,10 @@ use crate::remind::{self, ReminderStore}; use crate::trust::l3_gate_applies; use async_trait::async_trait; use serenity::builder::{ - CreateActionRow, CreateAttachment, CreateButton, CreateCommand, CreateCommandOption, - CreateEmbed, CreateEmbedFooter, CreateInteractionResponse, CreateInteractionResponseFollowup, - CreateInteractionResponseMessage, CreateSelectMenu, CreateSelectMenuKind, + CreateActionRow, CreateAllowedMentions, CreateAttachment, CreateButton, CreateCommand, + CreateCommandOption, CreateEmbed, CreateEmbedFooter, CreateInteractionResponse, + CreateInteractionResponseFollowup, CreateInteractionResponseMessage, CreateMessage, + CreateSelectMenu, CreateSelectMenuKind, CreateSelectMenuOption, CreateThread, EditChannel, EditMessage, GetMessages, }; use serenity::http::Http; @@ -64,11 +68,15 @@ const THREAD_EXPORT_MESSAGE_LIMIT: usize = 5000; pub struct DiscordAdapter { http: Arc, + bot_id: Arc>, } impl DiscordAdapter { pub fn new(http: Arc) -> Self { - Self { http } + Self { + http, + bot_id: Arc::new(OnceLock::new()), + } } /// Resolve the effective Discord channel ID from a ChannelRef. @@ -94,7 +102,17 @@ impl ChatAdapter for DiscordAdapter { content: &str, ) -> anyhow::Result { let ch_id: u64 = Self::resolve_channel(channel).parse()?; - let msg = ChannelId::new(ch_id).say(&self.http, content).await?; + let msg = ChannelId::new(ch_id) + .send_message( + &self.http, + CreateMessage::new() + .content(content) + // Presentation messages are never allowed to ping users, + // roles, @everyone, or @here. Control-plane handoffs use + // the explicit send_handoff path below. + .allowed_mentions(CreateAllowedMentions::new().replied_user(false)), + ) + .await?; Ok(MessageRef { channel: channel.clone(), message_id: msg.id.to_string(), @@ -113,8 +131,9 @@ impl ChatAdapter for DiscordAdapter { // Invalid message ID, fall back to plain send return self.send_message(channel, content).await; } - let builder = serenity::builder::CreateMessage::new() + let builder = CreateMessage::new() .content(content) + .allowed_mentions(CreateAllowedMentions::new().replied_user(false)) .reference_message((ChannelId::new(ch_id), MessageId::new(msg_id))); match ChannelId::new(ch_id) .send_message(&self.http, builder) @@ -148,14 +167,61 @@ impl ChatAdapter for DiscordAdapter { .edit_message( &self.http, MessageId::new(msg_id), - EditMessage::new().content(content), + EditMessage::new() + .content(content) + .allowed_mentions(CreateAllowedMentions::new().replied_user(false)), ) .await?; Ok(()) } - fn use_streaming(&self, other_bot_present: bool) -> bool { - !other_bot_present + fn use_streaming(&self, _other_bot_present: bool) -> bool { + // Discord presentation writes are sanitized and explicitly disable + // allowed mentions, so MESSAGE_UPDATE edits cannot trigger another bot. + // Keep live progress visible even in a multibot thread; only the + // structured handoff path below can emit an executable MESSAGE_CREATE. + true + } + + fn set_bot_identity(&self, bot_id: &str) { + let _ = self.bot_id.set(bot_id.to_string()); + } + + fn bot_identity(&self) -> Option { + self.bot_id.get().cloned() + } + + async fn send_handoff( + &self, + channel: &ChannelRef, + request: &HandoffRequest, + ) -> anyhow::Result { + if request.schema != MULTIBOT_HANDOFF_SCHEMA { + anyhow::bail!("unsupported handoff schema: {}", request.schema); + } + let target_id: u64 = request.target_bot_id.parse()?; + let ch_id: u64 = Self::resolve_channel(channel).parse()?; + let envelope = serde_json::to_string(request)?; + // The target mention is generated from the validated target ID exactly + // once. The JSON envelope is the control-plane payload; no presentation + // edit is used to route it. + let content = format!("<@{target_id}>\n{envelope}"); + let msg = ChannelId::new(ch_id) + .send_message( + &self.http, + CreateMessage::new() + .content(content) + .allowed_mentions( + CreateAllowedMentions::new() + .users([UserId::new(target_id)]) + .replied_user(false), + ), + ) + .await?; + Ok(MessageRef { + channel: channel.clone(), + message_id: msg.id.to_string(), + }) } async fn create_thread( @@ -265,9 +331,32 @@ pub struct Handler { pub reminder_store: ReminderStore, /// Track scheduled reminder IDs to prevent duplicate scheduling on reconnect. pub scheduled_ids: tokio::sync::Mutex>, + /// Recently accepted structured handoff event IDs; prevents replay within a process. + pub handoff_events: tokio::sync::Mutex>, +} + +fn accept_handoff_event_in( + events: &mut HashMap, + event_id: &str, +) -> bool { + const HANDOFF_REPLAY_WINDOW: std::time::Duration = std::time::Duration::from_secs(600); + events.retain(|_, seen_at| seen_at.elapsed() < HANDOFF_REPLAY_WINDOW); + if events.contains_key(event_id) { + return false; + } + events.insert(event_id.to_string(), tokio::time::Instant::now()); + true } impl Handler { + /// Accept each validated control-plane event at most once during its replay + /// window. This is intentionally process-local; the envelope expiry and + /// event ID remain part of the wire contract for a future shared dedupe store. + async fn accept_handoff_event(&self, event_id: &str) -> bool { + let mut events = self.handoff_events.lock().await; + accept_handoff_event_in(&mut events, event_id) + } + /// Check if the bot has participated in a Discord thread, and whether /// other bots have also posted in it. /// Returns `(involved, other_bot_present)`. @@ -485,6 +574,32 @@ impl EventHandler for Handler { .iter() .any(|r| self.allowed_role_ids.contains(&r.get()))); + // Control-plane messages are parsed before ordinary bot gating. A + // schema-looking message that fails validation is dropped rather than + // falling through as an executable prompt. + let handoff = parse_handoff_envelope(&msg.content); + if is_handoff_candidate(&msg.content) { + let valid = handoff.as_ref().is_some_and(|request| { + validate_handoff_envelope( + &msg.content, + request, + msg.author.id.get(), + bot_id.get(), + &self.trusted_bot_ids, + chrono::Utc::now().timestamp().max(0) as u64, + ) + }); + if !valid { + tracing::warn!(author = %msg.author.id, channel = %msg.channel_id, "invalid structured handoff rejected"); + return; + } + let request = handoff.as_ref().expect("validated handoff must be present"); + if !self.accept_handoff_event(&request.event_id).await { + tracing::info!(event_id = %request.event_id, "duplicate structured handoff ignored"); + return; + } + } + // Early-gating optimization for bot messages to avoid unnecessary // async/HTTP thread detection calls when ambient mode is inactive and // the bot would gate it out anyway. (#1197 regression safety) @@ -836,7 +951,10 @@ impl EventHandler for Handler { return; } - let prompt = resolve_mentions(&msg.content, bot_id, &self.allowed_role_ids); + let prompt = handoff + .as_ref() + .map(|request| request.payload.text.clone()) + .unwrap_or_else(|| resolve_mentions(&msg.content, bot_id, &self.allowed_role_ids)); // No text and no attachments → skip if prompt.is_empty() && msg.attachments.is_empty() { @@ -1389,6 +1507,11 @@ impl EventHandler for Handler { } async fn ready(&self, ctx: Context, ready: Ready) { + let adapter = self + .adapter + .get_or_init(|| Arc::new(DiscordAdapter::new(ctx.http.clone()))) + .clone(); + adapter.set_bot_identity(&ready.user.id.to_string()); info!(user = %ready.user.name, "discord bot connected"); // Build the shared command list once. @@ -3265,6 +3388,133 @@ fn truncate_to_utf16_budget(body: &str, prefix: &str, suffix: &str, limit: usize truncated } +/// Parse the JSON control envelope carried after the generated target mention. +/// Presentation messages are never parsed here because they do not use this schema. +fn parse_handoff_envelope(content: &str) -> Option { + let (mention, body) = content.split_once('\n')?; + if !mention.starts_with("<@") || !mention.ends_with('>') { + return None; + } + serde_json::from_str(body.trim()).ok() +} + +fn is_handoff_candidate(content: &str) -> bool { + content.contains(MULTIBOT_HANDOFF_SCHEMA) +} + +/// Validate every wire-level handoff invariant before it can bypass ordinary +/// bot-message gating. Invalid candidates fail closed and are never sent to ACP. +fn validate_handoff_envelope( + content: &str, + request: &HandoffRequest, + author_id: u64, + bot_id: u64, + trusted_bot_ids: &HashSet, + now: u64, +) -> bool { + let target_id = match request.target_bot_id.parse::() { + Ok(id) => id, + Err(_) => return false, + }; + let source_id = match request.source_bot_id.parse::() { + Ok(id) => id, + Err(_) => return false, + }; + let target_mention = format!("<@{target_id}>"); + let Some(first_line) = content.lines().next() else { + return false; + }; + + request.schema == MULTIBOT_HANDOFF_SCHEMA + && source_id == author_id + && source_id != bot_id + && trusted_bot_ids.contains(&author_id) + && target_id == bot_id + && first_line == target_mention + && content.matches(&target_mention).count() == 1 + && !content.contains("<@!") + && !content.contains("<@&") + && !content.contains("@everyone") + && !content.contains("@here") + && request.event_id.len() <= 64 + && uuid::Uuid::parse_str(&request.event_id).is_ok() + && request.expires_at > now + && request.expires_at <= now.saturating_add(3600) + && request.hop_count == 0 + && !request.origin_channel_id.is_empty() + && !request.payload.text.is_empty() + && request.payload.text.len() <= 100_000 +} + +#[cfg(test)] +mod handoff_tests { + use super::*; + use crate::adapter::{HandoffPayload, MULTIBOT_HANDOFF_SCHEMA}; + + fn request(target: u64) -> HandoffRequest { + HandoffRequest { + schema: MULTIBOT_HANDOFF_SCHEMA.into(), + source_bot_id: "10".into(), + target_bot_id: target.to_string(), + event_id: uuid::Uuid::nil().to_string(), + origin_channel_id: "thread".into(), + origin_thread_id: Some("thread".into()), + origin_message_id: None, + expires_at: 2_000, + hop_count: 0, + payload: HandoffPayload { text: "complete".into() }, + } + } + + #[test] + fn handoff_requires_exact_target_mention_and_validated_target() { + let req = request(20); + let content = format!("<@20>\n{}", serde_json::to_string(&req).unwrap()); + let mut trusted = HashSet::new(); + trusted.insert(10); + assert!(validate_handoff_envelope(&content, &req, 10, 20, &trusted, 1_000)); + assert!(!validate_handoff_envelope( + &content.replace("<@20>", "<@21>"), + &req, + 10, + 20, + &trusted, + 1_000 + )); + } + + #[test] + fn handoff_rejects_untrusted_expired_or_forwarded_requests() { + let mut req = request(20); + let content = format!("<@20>\n{}", serde_json::to_string(&req).unwrap()); + let mut trusted = HashSet::new(); + assert!(!validate_handoff_envelope(&content, &req, 10, 20, &trusted, 1_000)); + trusted.insert(10); + req.expires_at = 1_000; + let content = format!("<@20>\n{}", serde_json::to_string(&req).unwrap()); + assert!(!validate_handoff_envelope(&content, &req, 10, 20, &trusted, 1_000)); + req.expires_at = 2_000; + req.hop_count = 1; + let content = format!("<@20>\n{}", serde_json::to_string(&req).unwrap()); + assert!(!validate_handoff_envelope(&content, &req, 10, 20, &trusted, 1_000)); + } + + #[test] + fn handoff_event_ids_are_replay_protected() { + let mut events = HashMap::new(); + assert!(accept_handoff_event_in(&mut events, "event-1")); + assert!(!accept_handoff_event_in(&mut events, "event-1")); + assert!(accept_handoff_event_in(&mut events, "event-2")); + } + + #[test] + fn handoff_parser_rejects_non_schema_content() { + assert!(parse_handoff_envelope("<@20>\nnot json").is_none()); + assert!(!is_handoff_candidate("<@20>\nhello")); + assert!(is_handoff_candidate(MULTIBOT_HANDOFF_SCHEMA)); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/main.rs b/src/main.rs index ac54067e2..4b8bad7e3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1344,6 +1344,7 @@ async fn main() -> anyhow::Result<()> { ambient: ambient_dispatcher, reminder_store: reminder_store.clone(), scheduled_ids: tokio::sync::Mutex::new(std::collections::HashSet::new()), + handoff_events: tokio::sync::Mutex::new(std::collections::HashMap::new()), }; let intents = GatewayIntents::GUILD_MESSAGES