From eb6f563d84e40e15945efdacee38edafefbf1a59 Mon Sep 17 00:00:00 2001 From: Ron Liu Date: Tue, 16 Jun 2026 10:03:05 +1000 Subject: [PATCH] feat(telegram): transcribe incoming voice messages (#3679) Co-authored-by: Ron --- .../providers/telegram/channel_ops.rs | 2 +- .../providers/telegram/channel_recv.rs | 580 +++++++++++++++++- .../providers/telegram/channel_tests.rs | 269 ++++++++ .../providers/telegram/channel_types.rs | 13 + 4 files changed, 841 insertions(+), 23 deletions(-) diff --git a/src/openhuman/channels/providers/telegram/channel_ops.rs b/src/openhuman/channels/providers/telegram/channel_ops.rs index cf57c68fe..a9978986f 100644 --- a/src/openhuman/channels/providers/telegram/channel_ops.rs +++ b/src/openhuman/channels/providers/telegram/channel_ops.rs @@ -423,7 +423,7 @@ Ensure only one `openhuman` process is using this bot token." continue; } - let Some(msg) = self.parse_update_message(update) else { + let Some(msg) = self.parse_update_message_or_voice(update).await else { self.handle_unauthorized_message(update).await; continue; }; diff --git a/src/openhuman/channels/providers/telegram/channel_recv.rs b/src/openhuman/channels/providers/telegram/channel_recv.rs index 576907f62..db058ba42 100644 --- a/src/openhuman/channels/providers/telegram/channel_recv.rs +++ b/src/openhuman/channels/providers/telegram/channel_recv.rs @@ -1,10 +1,23 @@ //! Telegram channel — inbound message/reaction parsing, allowlist checks, mention filtering, //! unauthorized-message handling, and typing-action helpers. -use super::channel_types::{TelegramChannel, TelegramReactionEvent, APPROVAL_PROMPT_DEBOUNCE_SECS}; +use super::channel_types::{ + TelegramChannel, TelegramReactionEvent, TelegramVoiceAttachment, APPROVAL_PROMPT_DEBOUNCE_SECS, + TELEGRAM_MAX_VOICE_FILE_BYTES, +}; use crate::openhuman::channels::traits::{Channel, ChannelMessage, SendMessage}; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; use std::time::Instant; +#[derive(Debug, Clone)] +pub(crate) struct TelegramIncomingMessageContext { + pub(crate) sender_identity: String, + pub(crate) reply_target: String, + pub(crate) chat_id: String, + pub(crate) message_id: i64, + pub(crate) mention_text: Option, +} + impl TelegramChannel { pub(crate) fn typing_body_for_recipient(recipient: &str) -> serde_json::Value { let (chat_id, thread_id) = Self::parse_reply_target(recipient); @@ -230,9 +243,12 @@ impl TelegramChannel { return; }; - let Some(text) = message.get("text").and_then(serde_json::Value::as_str) else { + if !Self::is_supported_unauthorized_message(message) { + tracing::debug!("[telegram][approval] ignoring unsupported unauthorized update"); return; - }; + } + + let text = message.get("text").and_then(serde_json::Value::as_str); let username_opt = message .get("from") @@ -292,7 +308,7 @@ impl TelegramChannel { return; } - if let Some(code) = Self::extract_bind_code(text) { + if let Some(code) = text.and_then(Self::extract_bind_code) { if let Some(pairing) = self.pairing.as_ref() { match pairing.try_pair(code).await { Ok(Some(_token)) => { @@ -417,6 +433,14 @@ impl TelegramChannel { } } + pub(crate) fn is_supported_unauthorized_message(message: &serde_json::Value) -> bool { + message + .get("text") + .and_then(serde_json::Value::as_str) + .is_some() + || message.get("voice").is_some() + } + pub(crate) fn parse_update_message( &self, update: &serde_json::Value, @@ -426,7 +450,219 @@ impl TelegramChannel { .or_else(|| update.get("edited_message"))?; let text = message.get("text").and_then(serde_json::Value::as_str)?; + let ctx = self.parse_incoming_message_context(message, Some(text))?; + let content = match ctx.mention_text.clone() { + Some(content) => content, + None if self.mention_only && Self::is_group_message(message) => return None, + None => text.to_string(), + }; + (!content.trim().is_empty()).then(|| self.channel_message_from_context(ctx, content)) + } + + pub(crate) async fn parse_update_message_or_voice( + &self, + update: &serde_json::Value, + ) -> Option { + let update_id = update.get("update_id").and_then(serde_json::Value::as_i64); + let message = update + .get("message") + .or_else(|| update.get("edited_message")); + let chat_id = message + .and_then(|message| message.get("chat")) + .and_then(|chat| chat.get("id")) + .and_then(serde_json::Value::as_i64); + let message_id = message + .and_then(|message| message.get("message_id")) + .and_then(serde_json::Value::as_i64); + let has_text = message + .and_then(|message| message.get("text")) + .and_then(serde_json::Value::as_str) + .is_some(); + let has_voice = message.and_then(|message| message.get("voice")).is_some(); + + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + has_text, + has_voice, + "[telegram:voice] parse update dispatch" + ); + + if let Some(msg) = self.parse_update_message(update) { + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + "[telegram:voice] selected text parser" + ); + return Some(msg); + } + + if has_voice { + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + "[telegram:voice] selected voice parser" + ); + } else { + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + "[telegram:voice] update has no supported text or voice payload" + ); + } + + let msg = self.parse_update_voice_message(update).await; + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + parsed = msg.is_some(), + "[telegram:voice] parse update result" + ); + msg + } + + pub(crate) fn parse_update_voice_attachment( + update: &serde_json::Value, + ) -> Option { + let voice = update + .get("message") + .or_else(|| update.get("edited_message"))? + .get("voice")?; + + let file_id = voice + .get("file_id") + .and_then(serde_json::Value::as_str)? + .trim(); + if file_id.is_empty() { + return None; + } + + Some(TelegramVoiceAttachment { + file_id: file_id.to_string(), + file_unique_id: voice + .get("file_unique_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + file_size: voice.get("file_size").and_then(serde_json::Value::as_u64), + mime_type: voice + .get("mime_type") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + }) + } + + pub(crate) async fn parse_update_voice_message( + &self, + update: &serde_json::Value, + ) -> Option { + let update_id = update.get("update_id").and_then(serde_json::Value::as_i64); + let message = update + .get("message") + .or_else(|| update.get("edited_message"))?; + let voice = Self::parse_update_voice_attachment(update)?; + let caption = message.get("caption").and_then(serde_json::Value::as_str); + let chat_id = message + .get("chat") + .and_then(|chat| chat.get("id")) + .and_then(serde_json::Value::as_i64); + let message_id = message + .get("message_id") + .and_then(serde_json::Value::as_i64); + + tracing::debug!( + update_id = ?update_id, + chat_id = ?chat_id, + message_id = ?message_id, + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + has_caption = caption.is_some_and(|caption| !caption.trim().is_empty()), + "[telegram:voice] parse voice message entry" + ); + + let ctx = self.parse_incoming_message_context(message, caption)?; + tracing::debug!( + update_id = ?update_id, + chat_id = %ctx.chat_id, + message_id = ctx.message_id, + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + has_mention_text = ctx.mention_text.is_some(), + "[telegram:voice] voice message accepted for transcription" + ); + + match self.transcribe_telegram_voice(&voice).await { + Ok(transcript) => { + let transcript = transcript.trim(); + if transcript.is_empty() { + tracing::warn!( + chat_id = %ctx.chat_id, + message_id = ctx.message_id, + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + "[telegram:voice] inbound voice transcription returned empty text" + ); + self.send_voice_transcription_failure(&ctx).await; + return None; + } + + let mention_only_group = self.mention_only && Self::is_group_message(message); + let content = + Self::voice_message_content(transcript, caption, &ctx, mention_only_group); + + tracing::debug!( + update_id = ?update_id, + chat_id = %ctx.chat_id, + message_id = ctx.message_id, + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + content_chars = content.chars().count(), + "[telegram:voice] voice transcript ready for channel dispatch" + ); + + Some(self.channel_message_from_context(ctx, content)) + } + Err(error) => { + let error = self.redact_bot_token(error.to_string()); + tracing::warn!( + chat_id = %ctx.chat_id, + message_id = ctx.message_id, + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + %error, + "[telegram:voice] inbound voice transcription failed" + ); + self.send_voice_transcription_failure(&ctx).await; + None + } + } + } + + pub(crate) fn voice_message_content( + transcript: &str, + caption: Option<&str>, + ctx: &TelegramIncomingMessageContext, + mention_only_group: bool, + ) -> String { + let caption_prefix = if mention_only_group { + ctx.mention_text.as_deref() + } else { + caption + } + .map(str::trim) + .filter(|caption| !caption.is_empty()); + + match caption_prefix { + Some(prefix) => format!("{prefix}\n\n{transcript}"), + None => transcript.to_string(), + } + } + + pub(crate) fn parse_incoming_message_context( + &self, + message: &serde_json::Value, + mention_source: Option<&str>, + ) -> Option { let username = message .get("from") .and_then(|from| from.get("username")) @@ -455,23 +691,27 @@ impl TelegramChannel { tracing::debug!( username = %username, sender_id = sender_id.as_deref().unwrap_or("none"), - message_len = text.len(), + message_len = mention_source.map(str::len).unwrap_or_default(), "[telegram] dropped message: sender not in allowed_users (unauthorized handler may reply)" ); return None; } let is_group = Self::is_group_message(message); - if self.mention_only && is_group { + let mention_text = if self.mention_only && is_group { + let mention_source = mention_source?; let bot_username = self.bot_username.lock(); if let Some(ref bot_username) = *bot_username { - if !Self::contains_bot_mention(text, bot_username) { + if !Self::contains_bot_mention(mention_source, bot_username) { return None; } + Self::normalize_incoming_content(mention_source, bot_username) } else { return None; } - } + } else { + None + }; let chat_id = message .get("chat") @@ -506,7 +746,6 @@ impl TelegramChannel { // Telegram "reply" targeting should point to the inbound message itself so the // assistant response is visibly attached in chat. We still retain the inbound // parent reference in logs for reply-context diagnostics. - let outbound_reply_to_message_id = Some(message_id.to_string()); tracing::debug!( chat_id, message_id, @@ -514,26 +753,323 @@ impl TelegramChannel { "Telegram inbound message parsed for reply mapping" ); - let content = if self.mention_only && is_group { - let bot_username = self.bot_username.lock(); - let bot_username = bot_username.as_ref()?; - Self::normalize_incoming_content(text, bot_username)? - } else { - text.to_string() - }; - - Some(ChannelMessage { - id: format!("telegram_{chat_id}_{message_id}"), - sender: sender_identity, + Some(TelegramIncomingMessageContext { + sender_identity, reply_target, + chat_id, + message_id, + mention_text, + }) + } + + pub(crate) fn channel_message_from_context( + &self, + ctx: TelegramIncomingMessageContext, + content: String, + ) -> ChannelMessage { + ChannelMessage { + id: format!("telegram_{}_{}", ctx.chat_id, ctx.message_id), + sender: ctx.sender_identity, + reply_target: ctx.reply_target, content, channel: "telegram".to_string(), timestamp: std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(), - thread_ts: outbound_reply_to_message_id, - }) + thread_ts: Some(ctx.message_id.to_string()), + } + } + + pub(crate) async fn send_voice_transcription_failure( + &self, + ctx: &TelegramIncomingMessageContext, + ) { + let _ = self + .send( + &SendMessage::new( + "Voice transcription failed. Please try again or send text.", + &ctx.reply_target, + ) + .in_thread(Some(ctx.message_id.to_string())), + ) + .await; + } + + pub(crate) async fn transcribe_telegram_voice( + &self, + voice: &TelegramVoiceAttachment, + ) -> anyhow::Result { + tracing::debug!( + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + declared_file_size = ?voice.file_size, + mime_type = voice.mime_type.as_deref().unwrap_or("unknown"), + "[telegram:voice] transcribe entry" + ); + + if let Some(file_size) = voice.file_size { + if file_size > TELEGRAM_MAX_VOICE_FILE_BYTES { + anyhow::bail!( + "Telegram voice file too large: {file_size} bytes (max {TELEGRAM_MAX_VOICE_FILE_BYTES})" + ); + } + } + + let (audio_bytes, file_name, api_file_size) = self + .download_telegram_voice_file(&voice.file_id, voice.file_unique_id.as_deref()) + .await?; + tracing::debug!( + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + downloaded_bytes = audio_bytes.len(), + api_file_size = ?api_file_size, + file_name = %file_name, + "[telegram:voice] download completed before transcription" + ); + + if let Some(file_size) = api_file_size { + if file_size > TELEGRAM_MAX_VOICE_FILE_BYTES { + anyhow::bail!( + "Telegram getFile reported voice file too large: {file_size} bytes (max {TELEGRAM_MAX_VOICE_FILE_BYTES})" + ); + } + } + if u64::try_from(audio_bytes.len()).unwrap_or(u64::MAX) > TELEGRAM_MAX_VOICE_FILE_BYTES { + anyhow::bail!( + "downloaded Telegram voice file too large: {} bytes (max {TELEGRAM_MAX_VOICE_FILE_BYTES})", + audio_bytes.len() + ); + } + + let audio_base64 = BASE64.encode(&audio_bytes); + let config = crate::openhuman::config::rpc::load_config_with_timeout() + .await + .map_err(anyhow::Error::msg)?; + let provider_name = crate::openhuman::voice::effective_stt_provider(&config); + let model = crate::openhuman::voice::DEFAULT_WHISPER_MODEL.to_string(); + let provider = + crate::openhuman::voice::create_stt_provider(&provider_name, &model, &config)?; + let mime_type = voice.mime_type.as_deref().unwrap_or("audio/ogg"); + + tracing::debug!( + provider = provider.name(), + mime_type = %mime_type, + file_name = %file_name, + bytes = audio_bytes.len(), + "[telegram:voice] calling STT provider for inbound voice" + ); + + let outcome = provider + .transcribe( + &config, + &audio_base64, + Some(mime_type), + Some(&file_name), + None, + ) + .await + .map_err(anyhow::Error::msg)?; + + let text = outcome.value.text; + tracing::debug!( + file_unique_id = voice.file_unique_id.as_deref().unwrap_or("unknown"), + transcript_chars = text.chars().count(), + "[telegram:voice] transcription completed" + ); + + Ok(text) + } + + pub(crate) fn redact_bot_token(&self, value: impl AsRef) -> String { + if self.bot_token.is_empty() { + return value.as_ref().to_string(); + } + value.as_ref().replace(&self.bot_token, "") + } + + pub(crate) async fn download_telegram_voice_file( + &self, + file_id: &str, + file_unique_id: Option<&str>, + ) -> anyhow::Result<(Vec, String, Option)> { + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_id_present = !file_id.trim().is_empty(), + "[telegram:voice:download] requesting Telegram getFile" + ); + + let resp = self + .http_client() + .post(self.api_url("getFile")) + .json(&serde_json::json!({ "file_id": file_id })) + .send() + .await + .map_err(|e| anyhow::anyhow!("{}", self.redact_bot_token(e.to_string())))?; + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + status = %status, + response_bytes = body.len(), + "[telegram:voice:download] Telegram getFile response received" + ); + + if !status.is_success() { + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + status = %status, + error = %self.redact_bot_token(&body), + "[telegram:voice:download] Telegram getFile failed" + ); + anyhow::bail!( + "Telegram getFile failed ({status}): {}", + self.redact_bot_token(body) + ); + } + + let payload: serde_json::Value = serde_json::from_str(&body) + .map_err(|e| anyhow::anyhow!("Telegram getFile returned invalid JSON: {e}"))?; + if !payload + .get("ok") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + { + let description = payload + .get("description") + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown Telegram getFile error"); + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + description, + "[telegram:voice:download] Telegram getFile returned ok=false" + ); + anyhow::bail!("Telegram getFile returned ok=false: {description}"); + } + + let result = payload + .get("result") + .ok_or_else(|| anyhow::anyhow!("Telegram getFile response missing result"))?; + let file_path = result + .get("file_path") + .and_then(serde_json::Value::as_str) + .filter(|path| !path.trim().is_empty()) + .ok_or_else(|| anyhow::anyhow!("Telegram getFile response missing file_path"))?; + let file_size = result.get("file_size").and_then(serde_json::Value::as_u64); + if let Some(file_size) = file_size { + if file_size > TELEGRAM_MAX_VOICE_FILE_BYTES { + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_size, + max_bytes = TELEGRAM_MAX_VOICE_FILE_BYTES, + "[telegram:voice:download] Telegram getFile file_size exceeds cap" + ); + anyhow::bail!( + "Telegram getFile reported voice file too large: {file_size} bytes (max {TELEGRAM_MAX_VOICE_FILE_BYTES})" + ); + } + } + let file_name = file_path + .rsplit('/') + .next() + .filter(|name| !name.trim().is_empty()) + .unwrap_or("voice.ogg") + .to_string(); + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_name = %file_name, + file_size = ?file_size, + "[telegram:voice:download] Telegram getFile result parsed" + ); + + let download_url = format!("{}/file/bot{}/{}", self.api_base, self.bot_token, file_path); + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_name = %file_name, + "[telegram:voice:download] starting Telegram voice file download" + ); + let mut file_resp = self + .http_client() + .get(download_url) + .send() + .await + .map_err(|e| anyhow::anyhow!("{}", self.redact_bot_token(e.to_string())))?; + if let Some(content_length) = file_resp.content_length() { + if content_length > TELEGRAM_MAX_VOICE_FILE_BYTES { + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + content_length, + max_bytes = TELEGRAM_MAX_VOICE_FILE_BYTES, + "[telegram:voice:download] voice download Content-Length exceeds cap" + ); + anyhow::bail!( + "Telegram voice download too large: {content_length} bytes (max {TELEGRAM_MAX_VOICE_FILE_BYTES})" + ); + } + } + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + content_length = ?file_resp.content_length(), + "[telegram:voice:download] voice download headers received" + ); + + let status = file_resp.status(); + if !status.is_success() { + let body = file_resp.text().await.unwrap_or_default(); + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + status = %status, + error = %self.redact_bot_token(&body), + "[telegram:voice:download] Telegram voice download failed" + ); + anyhow::bail!( + "Telegram voice download failed ({status}): {}", + self.redact_bot_token(body) + ); + } + + let mut bytes = Vec::new(); + while let Some(chunk) = file_resp + .chunk() + .await + .map_err(|e| anyhow::anyhow!("{}", self.redact_bot_token(e.to_string())))? + { + Self::append_telegram_voice_download_chunk( + &mut bytes, + &chunk, + TELEGRAM_MAX_VOICE_FILE_BYTES, + )?; + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_name = %file_name, + chunk_bytes = chunk.len(), + downloaded_bytes = bytes.len(), + "[telegram:voice:download] received voice file chunk" + ); + } + tracing::debug!( + file_unique_id = file_unique_id.unwrap_or("unknown"), + file_name = %file_name, + downloaded_bytes = bytes.len(), + "[telegram:voice:download] completed Telegram voice file download" + ); + + Ok((bytes, file_name, file_size)) + } + + pub(crate) fn append_telegram_voice_download_chunk( + bytes: &mut Vec, + chunk: &[u8], + max_bytes: u64, + ) -> anyhow::Result<()> { + let next_len = bytes + .len() + .checked_add(chunk.len()) + .ok_or_else(|| anyhow::anyhow!("Telegram voice download size overflow"))?; + if u64::try_from(next_len).unwrap_or(u64::MAX) > max_bytes { + anyhow::bail!("Telegram voice download too large: {next_len} bytes (max {max_bytes})"); + } + + bytes.extend_from_slice(chunk); + Ok(()) } pub(crate) fn parse_update_reaction( diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index ce796cc36..5ed6394b3 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/src/openhuman/channels/providers/telegram/channel_tests.rs @@ -449,6 +449,275 @@ fn parse_update_message_sets_thread_ts_to_current_message_id_for_outbound_reply( assert_eq!(msg.reply_target, "12345"); } +#[test] +fn parse_update_voice_attachment_extracts_telegram_voice_metadata() { + let update = serde_json::json!({ + "update_id": 6, + "message": { + "message_id": 101, + "voice": { + "file_id": "AwACAgUAAxkBAAIB", + "file_unique_id": "AgADabc", + "duration": 3, + "mime_type": "audio/ogg", + "file_size": 4096 + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": 12345 + } + } + }); + + let voice = TelegramChannel::parse_update_voice_attachment(&update) + .expect("voice attachment should parse"); + + assert_eq!(voice.file_id, "AwACAgUAAxkBAAIB"); + assert_eq!(voice.file_unique_id.as_deref(), Some("AgADabc")); + assert_eq!(voice.mime_type.as_deref(), Some("audio/ogg")); + assert_eq!(voice.file_size, Some(4096)); +} + +#[test] +fn inbound_voice_context_preserves_reply_mapping_for_transcript_dispatch() { + let ch = TelegramChannel::new("token".into(), vec!["*".into()], false); + let update = serde_json::json!({ + "update_id": 7, + "message": { + "message_id": 202, + "voice": { + "file_id": "voice-file-id" + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": -100200300 + }, + "message_thread_id": 42, + "reply_to_message": { + "message_id": 199 + } + } + }); + + let ctx = ch + .parse_incoming_message_context(&update["message"], None) + .expect("authorized voice message context should parse"); + let msg = ch.channel_message_from_context(ctx, "transcribed voice text".to_string()); + + assert_eq!(msg.sender, "alice"); + assert_eq!(msg.reply_target, "-100200300:42"); + assert_eq!(msg.thread_ts.as_deref(), Some("202")); + assert_eq!(msg.id, "telegram_-100200300_202"); + assert_eq!(msg.content, "transcribed voice text"); +} + +#[test] +fn inbound_voice_context_mention_only_group_requires_caption_mention() { + let ch = TelegramChannel::new("token".into(), vec!["*".into()], true); + *ch.bot_username.lock() = Some("mybot".to_string()); + let update = serde_json::json!({ + "update_id": 8, + "message": { + "message_id": 303, + "caption": "@mybot summarize this", + "voice": { + "file_id": "voice-file-id" + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": -100200300, + "type": "supergroup" + } + } + }); + + let ctx = ch + .parse_incoming_message_context(&update["message"], Some("@mybot summarize this")) + .expect("caption mention should allow voice in mention-only group"); + + assert_eq!(ctx.mention_text.as_deref(), Some("summarize this")); + + let no_caption_update = serde_json::json!({ + "update_id": 9, + "message": { + "message_id": 304, + "voice": { + "file_id": "voice-file-id" + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": -100200300, + "type": "supergroup" + } + } + }); + + assert!(ch + .parse_incoming_message_context(&no_caption_update["message"], None) + .is_none()); +} + +#[test] +fn inbound_voice_content_preserves_caption_outside_mention_only_groups() { + let ch = TelegramChannel::new("token".into(), vec!["*".into()], false); + let update = serde_json::json!({ + "message": { + "message_id": 305, + "caption": "summarize this", + "voice": { + "file_id": "voice-file-id" + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": 12345, + "type": "private" + } + } + }); + + let ctx = ch + .parse_incoming_message_context(&update["message"], Some("summarize this")) + .expect("authorized voice message context should parse"); + let content = TelegramChannel::voice_message_content( + "transcribed voice", + Some("summarize this"), + &ctx, + false, + ); + + assert_eq!(content, "summarize this\n\ntranscribed voice"); +} + +#[test] +fn inbound_voice_content_uses_normalized_caption_in_mention_only_groups() { + let ch = TelegramChannel::new("token".into(), vec!["*".into()], true); + *ch.bot_username.lock() = Some("mybot".to_string()); + let update = serde_json::json!({ + "message": { + "message_id": 306, + "caption": "@mybot summarize this", + "voice": { + "file_id": "voice-file-id" + }, + "from": { + "id": 555, + "username": "alice" + }, + "chat": { + "id": -100200300, + "type": "supergroup" + } + } + }); + + let ctx = ch + .parse_incoming_message_context(&update["message"], Some("@mybot summarize this")) + .expect("caption mention should allow voice in mention-only group"); + let content = TelegramChannel::voice_message_content( + "transcribed voice", + Some("@mybot summarize this"), + &ctx, + true, + ); + + assert_eq!(content, "summarize this\n\ntranscribed voice"); +} + +#[test] +fn unauthorized_approval_only_supports_text_or_voice_messages() { + let text = serde_json::json!({ "text": "hello" }); + let voice = serde_json::json!({ "voice": { "file_id": "voice-file-id" } }); + let sticker = serde_json::json!({ "sticker": { "file_id": "sticker-file-id" } }); + let photo = serde_json::json!({ "photo": [{ "file_id": "photo-file-id" }] }); + + assert!(TelegramChannel::is_supported_unauthorized_message(&text)); + assert!(TelegramChannel::is_supported_unauthorized_message(&voice)); + assert!(!TelegramChannel::is_supported_unauthorized_message( + &sticker + )); + assert!(!TelegramChannel::is_supported_unauthorized_message(&photo)); +} + +#[test] +fn telegram_error_redaction_hides_bot_token() { + let ch = TelegramChannel::new("123456:ABCSECRET".into(), vec!["*".into()], false); + let redacted = ch.redact_bot_token( + "request failed for https://api.telegram.org/bot123456:ABCSECRET/getFile", + ); + + assert!(!redacted.contains("123456:ABCSECRET")); + assert!(redacted.contains("")); +} + +#[tokio::test] +async fn download_telegram_voice_file_uses_get_file_path_and_downloads_bytes() { + use wiremock::matchers::{body_json, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + let mut ch = TelegramChannel::new("token".into(), vec!["*".into()], false); + ch.api_base = server.uri(); + + Mock::given(method("POST")) + .and(path("/bottoken/getFile")) + .and(body_json(serde_json::json!({ "file_id": "voice-file-id" }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "ok": true, + "result": { + "file_path": "voice/file_1.ogg", + "file_size": 4 + } + }))) + .mount(&server) + .await; + + Mock::given(method("GET")) + .and(path("/file/bottoken/voice/file_1.ogg")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(vec![1, 2, 3, 4])) + .mount(&server) + .await; + + let (bytes, file_name, file_size) = ch + .download_telegram_voice_file("voice-file-id", Some("voice-unique-id")) + .await + .expect("mocked Telegram voice download should succeed"); + + assert_eq!(bytes, vec![1, 2, 3, 4]); + assert_eq!(file_name, "file_1.ogg"); + assert_eq!(file_size, Some(4)); +} + +#[test] +fn append_telegram_voice_download_chunk_enforces_size_cap() { + let mut bytes = vec![1, 2, 3]; + + TelegramChannel::append_telegram_voice_download_chunk(&mut bytes, &[4], 4) + .expect("chunk within cap should append"); + assert_eq!(bytes, vec![1, 2, 3, 4]); + + let error = TelegramChannel::append_telegram_voice_download_chunk(&mut bytes, &[5], 4) + .expect_err("chunk beyond cap should fail"); + assert!(error + .to_string() + .contains("Telegram voice download too large")); + assert_eq!(bytes, vec![1, 2, 3, 4]); +} + #[test] fn parse_update_reaction_extracts_actor_target_and_emoji() { let ch = TelegramChannel::new("token".into(), vec!["*".into()], false); diff --git a/src/openhuman/channels/providers/telegram/channel_types.rs b/src/openhuman/channels/providers/telegram/channel_types.rs index 815edf203..f539b362d 100644 --- a/src/openhuman/channels/providers/telegram/channel_types.rs +++ b/src/openhuman/channels/providers/telegram/channel_types.rs @@ -9,6 +9,11 @@ use std::time::Instant; pub(crate) const TELEGRAM_RECENT_UPDATE_CACHE_SIZE: usize = 4096; +/// Telegram Bot API caps downloaded files at 20MB for `getFile`. +/// Keep our inbound voice path inside the same bound before base64 encoding +/// and dispatching to STT. +pub(crate) const TELEGRAM_MAX_VOICE_FILE_BYTES: u64 = 20 * 1024 * 1024; + /// De-bounce window for approval prompts: suppress duplicate prompts sent to the /// same chat+sender within this duration (prevents restart-race and rapid-fire spam). pub(crate) const APPROVAL_PROMPT_DEBOUNCE_SECS: u64 = 60; @@ -33,6 +38,14 @@ pub(crate) struct TelegramReactionEvent { pub(crate) emoji: String, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct TelegramVoiceAttachment { + pub(crate) file_id: String, + pub(crate) file_unique_id: Option, + pub(crate) file_size: Option, + pub(crate) mime_type: Option, +} + /// Telegram channel — long-polls the Bot API for updates pub struct TelegramChannel { pub(crate) bot_token: String,