diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs index d8718ebb..936197bc 100644 --- a/channels-src/telegram/src/lib.rs +++ b/channels-src/telegram/src/lib.rs @@ -100,6 +100,15 @@ struct TelegramMessage { /// Sticker. sticker: Option, + + /// Forum topic ID. Present when the message is sent inside a forum topic. + /// https://core.telegram.org/bots/api#message + #[serde(default)] + message_thread_id: Option, + + /// True when this message is sent inside a forum topic. + #[serde(default)] + is_topic_message: Option, } /// Telegram PhotoSize object. @@ -198,6 +207,10 @@ struct TelegramChat { /// Title for groups/channels. title: Option, + /// True when the supergroup has topics (forum mode) enabled. + #[serde(default)] + is_forum: Option, + /// Username for private chats. username: Option, } @@ -290,6 +303,10 @@ struct TelegramMessageMetadata { /// Whether this is a private (DM) chat. is_private: bool, + + /// Forum topic thread ID (for routing replies back to the correct topic). + #[serde(default, skip_serializing_if = "Option::is_none")] + message_thread_id: Option, } /// Channel configuration injected by host. @@ -680,7 +697,7 @@ impl Guest for TelegramChannel { let metadata: TelegramMessageMetadata = serde_json::from_str(&response.metadata_json) .map_err(|e| format!("Failed to parse metadata: {}", e))?; - send_response(metadata.chat_id, &response, Some(metadata.message_id)) + send_response(metadata.chat_id, &response, Some(metadata.message_id), metadata.message_thread_id) } fn on_broadcast(user_id: String, response: AgentResponse) -> Result<(), String> { @@ -688,7 +705,7 @@ impl Guest for TelegramChannel { .parse() .map_err(|e| format!("Invalid chat_id '{}': {}", user_id, e))?; - send_response(chat_id, &response, None) + send_response(chat_id, &response, None, None) } fn on_status(update: StatusUpdate) { @@ -712,11 +729,17 @@ impl Guest for TelegramChannel { match action { TelegramStatusAction::Typing => { // POST /sendChatAction with action "typing" - let payload = serde_json::json!({ + let mut payload = serde_json::json!({ "chat_id": metadata.chat_id, "action": "typing" }); + // sendChatAction requires message_thread_id even for the General + // topic (id=1), unlike sendMessage which rejects it. + if let Some(thread_id) = metadata.message_thread_id { + payload["message_thread_id"] = serde_json::Value::Number(thread_id.into()); + } + let payload_bytes = match serde_json::to_vec(&payload) { Ok(b) => b, Err(_) => return, @@ -744,7 +767,7 @@ impl Guest for TelegramChannel { TelegramStatusAction::Notify(prompt) => { // Send user-visible status updates for actionable events. if let Err(first_err) = - send_message(metadata.chat_id, &prompt, Some(metadata.message_id), None) + send_message(metadata.chat_id, &prompt, Some(metadata.message_id), None, metadata.message_thread_id) { channel_host::log( channel_host::LogLevel::Warn, @@ -754,7 +777,7 @@ impl Guest for TelegramChannel { ), ); - if let Err(retry_err) = send_message(metadata.chat_id, &prompt, None, None) { + if let Err(retry_err) = send_message(metadata.chat_id, &prompt, None, None, metadata.message_thread_id) { channel_host::log( channel_host::LogLevel::Debug, &format!( @@ -797,6 +820,15 @@ impl std::fmt::Display for SendError { } } +/// Normalize `message_thread_id` for outbound API calls. +/// +/// Telegram rejects `sendMessage` (and other send methods) when +/// `message_thread_id = 1` (the "General" topic). Return `None` in that +/// case so the field is omitted from the payload. +fn normalize_thread_id(thread_id: Option) -> Option { + thread_id.filter(|&id| id != 1) +} + /// Send a message via the Telegram Bot API. /// /// Returns the sent message_id on success. When `parse_mode` is set and @@ -807,7 +839,10 @@ fn send_message( text: &str, reply_to_message_id: Option, parse_mode: Option<&str>, + message_thread_id: Option, ) -> Result { + let message_thread_id = normalize_thread_id(message_thread_id); + let mut payload = serde_json::json!({ "chat_id": chat_id, "text": text, @@ -821,6 +856,10 @@ fn send_message( payload["parse_mode"] = serde_json::Value::String(mode.to_string()); } + if let Some(thread_id) = message_thread_id { + payload["message_thread_id"] = serde_json::Value::Number(thread_id.into()); + } + let payload_bytes = serde_json::to_vec(&payload) .map_err(|e| SendError::Other(format!("Failed to serialize payload: {}", e)))?; @@ -1036,7 +1075,10 @@ fn send_photo( mime_type: &str, data: &[u8], reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { + let message_thread_id = normalize_thread_id(message_thread_id); + if data.len() > MAX_PHOTO_SIZE { channel_host::log( channel_host::LogLevel::Info, @@ -1046,7 +1088,7 @@ fn send_photo( data.len() ), ); - return send_document(chat_id, filename, mime_type, data, reply_to_message_id); + return send_document(chat_id, filename, mime_type, data, reply_to_message_id, message_thread_id); } let boundary = format!("ironclaw-{}", channel_host::now_millis()); @@ -1056,6 +1098,9 @@ fn send_photo( if let Some(msg_id) = reply_to_message_id { write_multipart_field(&mut body, &boundary, "reply_to_message_id", &msg_id.to_string()); } + if let Some(thread_id) = message_thread_id { + write_multipart_field(&mut body, &boundary, "message_thread_id", &thread_id.to_string()); + } write_multipart_file(&mut body, &boundary, "photo", filename, mime_type, data); body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes()); @@ -1097,7 +1142,10 @@ fn send_document( mime_type: &str, data: &[u8], reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { + let message_thread_id = normalize_thread_id(message_thread_id); + let boundary = format!("ironclaw-{}", channel_host::now_millis()); let mut body = Vec::new(); @@ -1105,6 +1153,9 @@ fn send_document( if let Some(msg_id) = reply_to_message_id { write_multipart_field(&mut body, &boundary, "reply_to_message_id", &msg_id.to_string()); } + if let Some(thread_id) = message_thread_id { + write_multipart_field(&mut body, &boundary, "message_thread_id", &thread_id.to_string()); + } write_multipart_file(&mut body, &boundary, "document", filename, mime_type, data); body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes()); @@ -1154,10 +1205,11 @@ fn send_response( chat_id: i64, response: &AgentResponse, reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { // Send attachments first (photos/documents) for attachment in &response.attachments { - send_attachment(chat_id, attachment, reply_to_message_id)?; + send_attachment(chat_id, attachment, reply_to_message_id, message_thread_id)?; } // Skip text if empty and we already sent attachments @@ -1166,10 +1218,10 @@ fn send_response( } // Try Markdown, fall back to plain text on parse errors - match send_message(chat_id, &response.content, reply_to_message_id, Some("Markdown")) { + match send_message(chat_id, &response.content, reply_to_message_id, Some("Markdown"), message_thread_id) { Ok(_) => Ok(()), Err(SendError::ParseEntities(_)) => { - send_message(chat_id, &response.content, reply_to_message_id, None) + send_message(chat_id, &response.content, reply_to_message_id, None, message_thread_id) .map(|_| ()) .map_err(|e| format!("Plain-text retry also failed: {}", e)) } @@ -1182,6 +1234,7 @@ fn send_attachment( chat_id: i64, attachment: &Attachment, reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { if PHOTO_MIME_TYPES.contains(&attachment.mime_type.as_str()) { send_photo( @@ -1190,6 +1243,7 @@ fn send_attachment( &attachment.mime_type, &attachment.data, reply_to_message_id, + message_thread_id, ) } else { send_document( @@ -1198,6 +1252,7 @@ fn send_attachment( &attachment.mime_type, &attachment.data, reply_to_message_id, + message_thread_id, ) } } @@ -1357,6 +1412,7 @@ fn send_pairing_reply(chat_id: i64, code: &str) -> Result<(), String> { ), None, Some("Markdown"), + None, // Pairing happens in DMs, not forum topics ) .map(|_| ()) .map_err(|e| e.to_string()) @@ -1774,6 +1830,8 @@ fn handle_message(message: TelegramMessage) { } } + let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); + // For group chats, only respond if bot was mentioned or respond_to_all is enabled if !is_private { let respond_to_all = channel_host::workspace_read(RESPOND_TO_ALL_GROUP_PATH) @@ -1783,7 +1841,6 @@ fn handle_message(message: TelegramMessage) { if !respond_to_all { let has_command = content.starts_with('/'); - let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); let has_bot_mention = if bot_username.is_empty() { content.contains('@') } else { @@ -1814,11 +1871,23 @@ fn handle_message(message: TelegramMessage) { message_id: message.message_id, user_id: from.id, is_private, + message_thread_id: message.message_thread_id, }; let metadata_json = serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); - let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); + // Compute thread_id for forum topics: "chat_id:topic_id" to prevent + // collisions across different groups (topic IDs are only unique per chat). + // Only use message_thread_id when the chat is a forum — non-forum groups + // also carry message_thread_id for reply threads, which are not topics. + let thread_id = if message.chat.is_forum == Some(true) { + message.message_thread_id.map(|topic_id| { + format!("{}:{}", message.chat.id, topic_id) + }) + } else { + None + }; + let content_to_emit = match content_to_emit_for_agent( &content, if bot_username.is_empty() { @@ -1838,7 +1907,7 @@ fn handle_message(message: TelegramMessage) { user_id: from.id.to_string(), user_name: Some(user_name), content: content_to_emit, - thread_id: None, // Telegram doesn't have threads in the same way + thread_id, metadata_json, attachments, }); @@ -2657,4 +2726,100 @@ mod tests { // Verify the constant is 20 MB, matching the Slack channel limit assert_eq!(MAX_DOWNLOAD_SIZE_BYTES, 20 * 1024 * 1024); } + + // === Forum Topics (thread_id) tests === + + #[test] + fn test_parse_forum_message_with_thread_id() { + let json = r#"{ + "message_id": 100, + "message_thread_id": 42, + "is_topic_message": true, + "from": {"id": 1, "is_bot": false, "first_name": "A"}, + "chat": {"id": -1001234567890, "type": "supergroup", "is_forum": true}, + "text": "Hello from a topic" + }"#; + let msg: TelegramMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message_thread_id, Some(42)); + assert_eq!(msg.is_topic_message, Some(true)); + assert_eq!(msg.chat.is_forum, Some(true)); + } + + #[test] + fn test_parse_non_forum_message_backward_compat() { + let json = r#"{ + "message_id": 1, + "from": {"id": 1, "is_bot": false, "first_name": "A"}, + "chat": {"id": 1, "type": "private"}, + "text": "Hello" + }"#; + let msg: TelegramMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message_thread_id, None); + assert_eq!(msg.is_topic_message, None); + assert_eq!(msg.chat.is_forum, None); + } + + #[test] + fn test_metadata_with_message_thread_id() { + let metadata = TelegramMessageMetadata { + chat_id: -1001234567890, + message_id: 100, + user_id: 42, + is_private: false, + message_thread_id: Some(7), + }; + let json = serde_json::to_string(&metadata).unwrap(); + let parsed: TelegramMessageMetadata = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.message_thread_id, Some(7)); + } + + #[test] + fn test_metadata_backward_compat_no_thread_id() { + // Old metadata JSON without message_thread_id should deserialize with None + let json = r#"{"chat_id":123,"message_id":1,"user_id":42,"is_private":true}"#; + let metadata: TelegramMessageMetadata = serde_json::from_str(json).unwrap(); + assert_eq!(metadata.message_thread_id, None); + } + + #[test] + fn test_metadata_thread_id_not_serialized_when_none() { + let metadata = TelegramMessageMetadata { + chat_id: 123, + message_id: 1, + user_id: 42, + is_private: true, + message_thread_id: None, + }; + let json = serde_json::to_string(&metadata).unwrap(); + assert!(!json.contains("message_thread_id")); + } + + #[test] + fn test_thread_id_composition() { + // Verify "chat_id:topic_id" format for forum topics + let chat_id: i64 = -1001234567890; + let topic_id: i64 = 42; + let thread_id = format!("{}:{}", chat_id, topic_id); + assert_eq!(thread_id, "-1001234567890:42"); + } + + #[test] + fn test_normalize_thread_id_general_topic() { + // General topic (id=1) must be omitted — Telegram rejects sendMessage + // with message_thread_id=1. + assert_eq!(normalize_thread_id(Some(1)), None); + } + + #[test] + fn test_normalize_thread_id_regular_topic() { + // Non-General topics pass through unchanged + assert_eq!(normalize_thread_id(Some(42)), Some(42)); + assert_eq!(normalize_thread_id(Some(123)), Some(123)); + } + + #[test] + fn test_normalize_thread_id_none() { + // None stays None + assert_eq!(normalize_thread_id(None), None); + } }