From 9c73eabc4ea16b1b265cc9f111361694bebb092d Mon Sep 17 00:00:00 2001 From: Spacebot Date: Mon, 9 Mar 2026 13:20:09 -0700 Subject: [PATCH 1/4] fix(telegram): prevent panic when splitting messages at UTF-8 char boundaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The split_message function could panic when hard-cutting a message at Telegram's 4096-byte limit if the cut point landed inside a multi-byte UTF-8 character (e.g., em dashes '—'). Changed unwrap_or(max_len) to unwrap_or_else() which walks backward from max_len to find the last valid char boundary using is_char_boundary(). Also added handling for edge case where split_at could be 0. --- src/messaging/telegram.rs | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/messaging/telegram.rs b/src/messaging/telegram.rs index dc6f762d1..8f88f9599 100644 --- a/src/messaging/telegram.rs +++ b/src/messaging/telegram.rs @@ -1016,10 +1016,26 @@ fn split_message(text: &str, max_len: usize) -> Vec { break; } + // Find split point: prefer newline, then space, then hard-cut. + // When hard-cutting, ensure we don't split mid-character (UTF-8 safety). let split_at = remaining[..max_len] .rfind('\n') .or_else(|| remaining[..max_len].rfind(' ')) - .unwrap_or(max_len); + .unwrap_or_else(|| { + // Hard-cut: find the last valid char boundary before max_len + let mut pos = max_len; + while pos > 0 && !remaining.is_char_boundary(pos) { + pos -= 1; + } + pos + }); + + // Avoid empty chunks if split_at is 0 (e.g., first char is multi-byte) + let split_at = if split_at == 0 { + remaining.char_indices().nth(1).map(|(i, _)| i).unwrap_or(remaining.len()) + } else { + split_at + }; chunks.push(remaining[..split_at].to_string()); remaining = remaining[split_at..].trim_start(); From 27b108425a212e7eecb370f60a369d3cfddf3114 Mon Sep 17 00:00:00 2001 From: Spacebot Date: Mon, 9 Mar 2026 17:23:36 -0700 Subject: [PATCH 2/4] feat(telegram): add native streaming support via sendMessageDraft API - Add sendMessageDraft method for Telegram Bot API 9.5+ native streaming - Implement draft-based updates for smooth animated text in private chats - Add configuration for enabling/disabling native streaming - Fall back to edit-based streaming for groups/channels - Add draft_id and is_private tracking to ActiveStream struct --- src/config/load.rs | 2 + src/config/permissions.rs | 25 +++--- src/config/toml_schema.rs | 8 ++ src/config/types.rs | 7 ++ src/messaging/telegram.rs | 162 +++++++++++++++++++++++++++++++++++++- 5 files changed, 189 insertions(+), 15 deletions(-) diff --git a/src/config/load.rs b/src/config/load.rs index f39b3ecc9..383824dc7 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -1874,6 +1874,7 @@ impl Config { enabled: instance.enabled && token.is_some(), token: token.unwrap_or_default(), dm_allowed_users: instance.dm_allowed_users, + native_streaming: instance.native_streaming, } }) .collect::>(); @@ -1891,6 +1892,7 @@ impl Config { token: token.unwrap_or_default(), instances, dm_allowed_users: t.dm_allowed_users, + native_streaming: t.native_streaming, }) }), email: toml.messaging.email.and_then(|email| { diff --git a/src/config/permissions.rs b/src/config/permissions.rs index 8844cf614..d98ce6ea5 100644 --- a/src/config/permissions.rs +++ b/src/config/permissions.rs @@ -19,14 +19,6 @@ pub struct DiscordPermissions { impl DiscordPermissions { /// Build from the current config's discord settings and bindings. - pub fn from_config(discord: &DiscordConfig, bindings: &[Binding]) -> Self { - Self::from_bindings_for_adapter( - discord.dm_allowed_users.clone(), - discord.allow_bot_messages, - bindings, - None, - ) - } /// Build permissions for a named Discord adapter instance. pub fn from_instance_config(instance: &DiscordInstanceConfig, bindings: &[Binding]) -> Self { @@ -121,9 +113,7 @@ pub struct SlackPermissions { impl SlackPermissions { /// Build from the current config's slack settings and bindings. - pub fn from_config(slack: &SlackConfig, bindings: &[Binding]) -> Self { Self::from_bindings_for_adapter(slack.dm_allowed_users.clone(), bindings, None) - } /// Build permissions for a named Slack adapter instance. pub fn from_instance_config(instance: &SlackInstanceConfig, bindings: &[Binding]) -> Self { @@ -201,18 +191,26 @@ pub struct TelegramPermissions { pub chat_filter: Option>, /// User IDs allowed in private chats. pub dm_allowed_users: Vec, + /// Whether to use native streaming via sendMessageDraft API + pub native_streaming: bool, } impl TelegramPermissions { /// Build from the current config's telegram settings and bindings. pub fn from_config(telegram: &TelegramConfig, bindings: &[Binding]) -> Self { - Self::from_bindings_for_adapter(telegram.dm_allowed_users.clone(), bindings, None) + Self::from_bindings_for_adapter( + telegram.dm_allowed_users.clone(), + telegram.native_streaming, + bindings, + None, + ) } /// Build permissions for a named Telegram adapter instance. pub fn from_instance_config(instance: &TelegramInstanceConfig, bindings: &[Binding]) -> Self { Self::from_bindings_for_adapter( instance.dm_allowed_users.clone(), + instance.native_streaming, bindings, Some(instance.name.as_str()), ) @@ -220,6 +218,7 @@ impl TelegramPermissions { fn from_bindings_for_adapter( seed_dm_allowed_users: Vec, + native_streaming: bool, bindings: &[Binding], adapter_selector: Option<&str>, ) -> Self { @@ -261,6 +260,7 @@ impl TelegramPermissions { Self { chat_filter, dm_allowed_users, + native_streaming, } } } @@ -278,9 +278,6 @@ pub struct TwitchPermissions { impl TwitchPermissions { /// Build from the current config's twitch settings and bindings. - pub fn from_config(_twitch: &TwitchConfig, bindings: &[Binding]) -> Self { - Self::from_bindings_for_adapter(bindings, None) - } /// Build permissions for a named Twitch adapter instance. pub fn from_instance_config(instance: &TwitchInstanceConfig, bindings: &[Binding]) -> Self { diff --git a/src/config/toml_schema.rs b/src/config/toml_schema.rs index d105a7384..db28e1556 100644 --- a/src/config/toml_schema.rs +++ b/src/config/toml_schema.rs @@ -561,6 +561,8 @@ pub(super) struct TomlTelegramConfig { pub(super) instances: Vec, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default = "default_native_streaming")] + pub(super) native_streaming: bool, } #[derive(Deserialize)] @@ -571,6 +573,8 @@ pub(super) struct TomlTelegramInstanceConfig { pub(super) token: Option, #[serde(default)] pub(super) dm_allowed_users: Vec, + #[serde(default = "default_native_streaming")] + pub(super) native_streaming: bool, } #[derive(Deserialize)] @@ -721,6 +725,10 @@ pub(super) fn default_email_max_attachment_bytes() -> usize { 10 * 1024 * 1024 } +pub(super) fn default_native_streaming() -> bool { + true +} + #[derive(Deserialize)] pub(super) struct TomlBinding { pub(super) agent_id: String, diff --git a/src/config/types.rs b/src/config/types.rs index 88fd09497..4e2bb4faa 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -2057,6 +2057,9 @@ impl SystemSecrets for SlackConfig { #[derive(Clone)] pub struct TelegramConfig { + /// Whether to use native streaming via sendMessageDraft API (Bot API 9.5+) + /// Only works for private chats. Defaults to true. + pub native_streaming: bool, pub enabled: bool, pub token: String, /// Additional named Telegram bot instances for this platform. @@ -2072,6 +2075,8 @@ pub struct TelegramInstanceConfig { pub token: String, /// User IDs allowed to DM this bot instance. pub dm_allowed_users: Vec, + /// Whether to use native streaming via sendMessageDraft API + pub native_streaming: bool, } impl std::fmt::Debug for TelegramInstanceConfig { @@ -2081,6 +2086,7 @@ impl std::fmt::Debug for TelegramInstanceConfig { .field("enabled", &self.enabled) .field("token", &"[REDACTED]") .field("dm_allowed_users", &self.dm_allowed_users) + .field("native_streaming", &self.native_streaming) .finish() } } @@ -2092,6 +2098,7 @@ impl std::fmt::Debug for TelegramConfig { .field("token", &"[REDACTED]") .field("instances", &self.instances) .field("dm_allowed_users", &self.dm_allowed_users) + .field("native_streaming", &self.native_streaming) .finish() } } diff --git a/src/messaging/telegram.rs b/src/messaging/telegram.rs index 8f88f9599..f4cfb8a8a 100644 --- a/src/messaging/telegram.rs +++ b/src/messaging/telegram.rs @@ -15,6 +15,7 @@ use teloxide::types::{ ParseMode, ReactionType, ReplyParameters, UpdateKind, UserId, }; use teloxide::{ApiError, Bot, RequestError}; +use reqwest::Client; use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, LazyLock}; @@ -38,6 +39,10 @@ pub struct TelegramAdapter { typing_tasks: Arc>>>, /// Shutdown signal for the polling loop. shutdown_tx: Arc>>>, + /// HTTP client for raw Telegram API calls (sendMessageDraft). + http_client: Client, + /// Bot token for raw API calls. + token: String, } /// Tracks an in-progress streaming message edit. @@ -45,6 +50,10 @@ struct ActiveStream { chat_id: ChatId, message_id: MessageId, last_edit: Instant, + /// Draft ID for native streaming (sendMessageDraft API). + draft_id: Option, + /// Whether this is a private chat (draft API only works for private chats). + is_private: bool, } /// Telegram's per-message character limit. @@ -56,6 +65,9 @@ const FORMATTED_SPLIT_LENGTH: usize = MAX_MESSAGE_LENGTH / 2; /// Minimum interval between streaming edits to avoid rate limits. const STREAM_EDIT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(1000); +/// Minimum interval between draft updates (100-200ms recommended by Telegram). +const DRAFT_UPDATE_INTERVAL: std::time::Duration = std::time::Duration::from_millis(150); + impl TelegramAdapter { pub fn new( runtime_key: impl Into, @@ -74,7 +86,94 @@ impl TelegramAdapter { active_messages: Arc::new(RwLock::new(HashMap::new())), typing_tasks: Arc::new(RwLock::new(HashMap::new())), shutdown_tx: Arc::new(RwLock::new(None)), + http_client: Client::new(), + token, + } + } + + /// Send a message draft using Telegram's native streaming API (Bot API 9.5+). + /// + /// This method uses the `sendMessageDraft` endpoint which provides smooth + /// animated text updates in the Telegram client. Only works for private chats. + /// + /// # Arguments + /// * `chat_id` - The target chat ID (must be a private chat) + /// * `draft_id` - Unique identifier for this draft (same ID = animated updates) + /// * `text` - The text content to display + /// * `parse_mode` - Optional parse mode (HTML, Markdown, MarkdownV2) + /// + /// # Returns + /// `true` on success, `false` on failure + pub async fn send_message_draft( + &self, + chat_id: i64, + draft_id: i32, + text: &str, + parse_mode: Option<&str>, + ) -> bool { + let url = format!( + "https://api.telegram.org/bot{}/sendMessageDraft", + self.token + ); + + let mut body = serde_json::json!({ + "chat_id": chat_id, + "draft_id": draft_id, + "text": text, + }); + + if let Some(mode) = parse_mode { + body["parse_mode"] = serde_json::Value::String(mode.to_string()); + } + + match self.http_client + .post(&url) + .json(&body) + .send() + .await + { + Ok(response) => { + if response.status().is_success() { + true + } else { + tracing::debug!( + status = %response.status(), + "sendMessageDraft returned non-success status" + ); + false + } + } + Err(error) => { + tracing::debug!(%error, "sendMessageDraft request failed"); + false + } + } + } + + /// Generate a unique draft ID for a streaming session. + fn generate_draft_id(&self) -> i32 { + // Use timestamp in milliseconds XORed with a random component + // Ensure it's non-zero (required by Telegram API) + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i32; + // Ensure non-zero by ORing with 1 + timestamp | 1 + } + + /// Check if native streaming should be used for this message. + /// Returns true if native_streaming is enabled and the chat is private. + fn should_use_native_streaming(&self, message: &InboundMessage) -> bool { + let permissions = self.permissions.load(); + if !permissions.native_streaming { + return false; } + // Check if this is a private chat + message.metadata + .get("telegram_chat_type") + .and_then(|v| v.as_str()) + .is_some_and(|t| t == "private") } fn extract_chat_id(&self, message: &InboundMessage) -> anyhow::Result { @@ -428,6 +527,32 @@ impl Messaging for TelegramAdapter { OutboundResponse::StreamStart => { self.stop_typing(&message.conversation_id).await; + let is_private = self.should_use_native_streaming(&message); + let chat_id_value = chat_id.0; + + if is_private { + // Use native sendMessageDraft API for private chats + let draft_id = self.generate_draft_id(); + + // Send initial draft + if self.send_message_draft(chat_id_value, draft_id, "...", None).await { + self.active_messages.write().await.insert( + message.conversation_id.clone(), + ActiveStream { + chat_id, + message_id: MessageId(0), // Not used for draft streaming + last_edit: Instant::now(), + draft_id: Some(draft_id), + is_private: true, + }, + ); + return Ok(()); + } + // Fall through to edit-based streaming if draft fails + tracing::debug!("sendMessageDraft failed, falling back to edit-based streaming"); + } + + // Fallback: traditional edit-based streaming let placeholder = self .bot .send_message(chat_id, "...") @@ -441,13 +566,22 @@ impl Messaging for TelegramAdapter { chat_id, message_id: placeholder.id, last_edit: Instant::now(), + draft_id: None, + is_private: false, }, ); } OutboundResponse::StreamChunk(text) => { let mut active = self.active_messages.write().await; if let Some(stream) = active.get_mut(&message.conversation_id) { - if stream.last_edit.elapsed() < STREAM_EDIT_INTERVAL { + // Use faster interval for draft streaming + let min_interval = if stream.is_private && stream.draft_id.is_some() { + DRAFT_UPDATE_INTERVAL + } else { + STREAM_EDIT_INTERVAL + }; + + if stream.last_edit.elapsed() < min_interval { return Ok(()); } @@ -458,6 +592,20 @@ impl Messaging for TelegramAdapter { text }; + // Use native draft streaming if available + if stream.is_private { + if let Some(draft_id) = stream.draft_id { + let html = markdown_to_telegram_html(&display_text); + // Try HTML first, fall back to plain text + if !self.send_message_draft(stream.chat_id.0, draft_id, &html, Some("HTML")).await { + self.send_message_draft(stream.chat_id.0, draft_id, &display_text, None).await; + } + stream.last_edit = Instant::now(); + return Ok(()); + } + } + + // Fallback: traditional edit-based streaming let html = markdown_to_telegram_html(&display_text); if let Err(html_error) = self .bot @@ -480,6 +628,18 @@ impl Messaging for TelegramAdapter { } } OutboundResponse::StreamEnd => { + // For draft streaming, we need to send the final message + // The draft will be automatically discarded when we send a real message + let active = self.active_messages.read().await; + if let Some(stream) = active.get(&message.conversation_id) { + if stream.is_private && stream.draft_id.is_some() { + // Draft streaming: the final message will replace the draft + // No explicit cleanup needed - just remove from active + tracing::trace!("Draft streaming ended for conversation {}", message.conversation_id); + } + } + drop(active); + self.active_messages .write() .await From 937f0e461aff385e07d9465ca4eddf900e88f31d Mon Sep 17 00:00:00 2001 From: Spacebot Date: Mon, 9 Mar 2026 21:00:31 -0700 Subject: [PATCH 3/4] fix(browser): prevent panic when truncating node names at UTF-8 char boundaries Hardcoded byte slice at index 200 panics when the cutoff lands inside a multi-byte character. Use floor_char_boundary to find the nearest valid char boundary at or before 200 bytes instead. --- src/tools/browser.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/tools/browser.rs b/src/tools/browser.rs index d0590846c..ebaea395f 100644 --- a/src/tools/browser.rs +++ b/src/tools/browser.rs @@ -530,7 +530,8 @@ fn render_snapshot_node(node: &SnapshotNode, depth: usize, output: &mut String) output.push_str(" \""); // Truncate very long names for context efficiency. let display_name = if node.name.len() > 200 { - format!("{}...", &node.name[..200]) + let truncated = node.name.floor_char_boundary(200); + format!("{}...", &node.name[..truncated]) } else { node.name.clone() }; From e0106a64762d17b0cba8bb5db3dc757bd3ca25b9 Mon Sep 17 00:00:00 2001 From: Spacebot Date: Mon, 9 Mar 2026 23:20:09 -0700 Subject: [PATCH 4/4] fix(permissions): restore missing from_config constructors for Discord, Slack, Twitch Accidentally stripped in the native streaming PR (27b1084). The function signatures for DiscordPermissions::from_config, SlackPermissions::from_config, and TwitchPermissions::from_config were removed leaving orphaned function bodies that caused compile errors. --- src/config/permissions.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/config/permissions.rs b/src/config/permissions.rs index d98ce6ea5..38bcc5e96 100644 --- a/src/config/permissions.rs +++ b/src/config/permissions.rs @@ -19,6 +19,14 @@ pub struct DiscordPermissions { impl DiscordPermissions { /// Build from the current config's discord settings and bindings. + pub fn from_config(discord: &DiscordConfig, bindings: &[Binding]) -> Self { + Self::from_bindings_for_adapter( + discord.dm_allowed_users.clone(), + discord.allow_bot_messages, + bindings, + None, + ) + } /// Build permissions for a named Discord adapter instance. pub fn from_instance_config(instance: &DiscordInstanceConfig, bindings: &[Binding]) -> Self { @@ -113,7 +121,9 @@ pub struct SlackPermissions { impl SlackPermissions { /// Build from the current config's slack settings and bindings. + pub fn from_config(slack: &SlackConfig, bindings: &[Binding]) -> Self { Self::from_bindings_for_adapter(slack.dm_allowed_users.clone(), bindings, None) + } /// Build permissions for a named Slack adapter instance. pub fn from_instance_config(instance: &SlackInstanceConfig, bindings: &[Binding]) -> Self { @@ -278,6 +288,9 @@ pub struct TwitchPermissions { impl TwitchPermissions { /// Build from the current config's twitch settings and bindings. + pub fn from_config(_twitch: &TwitchConfig, bindings: &[Binding]) -> Self { + Self::from_bindings_for_adapter(bindings, None) + } /// Build permissions for a named Twitch adapter instance. pub fn from_instance_config(instance: &TwitchInstanceConfig, bindings: &[Binding]) -> Self {