From 4003300a8c151b7d099566810ccdfab803f70ee7 Mon Sep 17 00:00:00 2001 From: "firat.sertgoz" Date: Sun, 22 Feb 2026 22:07:57 +0400 Subject: [PATCH] fix: improve Telegram status delivery and reliability (#304) * fix: make Telegram status prompts reliable Approval and auth prompts could be missed when polling or reply-context sends failed, leaving users stuck in waiting states. This adds explicit status mapping and retries, keeps typing active through intermediate work while suppressing noisy tool telemetry, and adds regression tests plus CI coverage for the Telegram channel crate. * fix: normalize terminal status handling Terminal status strings from the agent loop can vary in casing and formatting, which could leak internal status lines to Telegram. This normalizes Done/Interrupted mapping and filters terminal status text consistently to keep chat UX clean while preserving actionable prompts. --------- Co-authored-by: Claude Opus 4.6 --- .github/workflows/test.yml | 2 + FEATURE_PARITY.md | 2 +- channels-src/telegram/Cargo.toml | 3 +- channels-src/telegram/src/lib.rs | 656 ++++++++++++++++++++++++------- src/channels/wasm/wrapper.rs | 631 +++++++++++++++++++++++++++-- wit/channel.wit | 12 + 6 files changed, 1147 insertions(+), 159 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 13fc8410..755fbf45 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -19,3 +19,5 @@ jobs: - uses: Swatinem/rust-cache@v2 - name: Run Tests run: cargo test --all-features -- --nocapture + - name: Run Telegram Channel Tests + run: cargo test --manifest-path channels-src/telegram/Cargo.toml -- --nocapture diff --git a/FEATURE_PARITY.md b/FEATURE_PARITY.md index d6e8fceb..ba7b5c24 100644 --- a/FEATURE_PARITY.md +++ b/FEATURE_PARITY.md @@ -120,7 +120,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O | Per-group tool policies | ✅ | ❌ | Allow/deny specific tools | | Thread isolation | ✅ | ✅ | Separate sessions per thread | | Per-channel media limits | ✅ | 🚧 | Caption support for media; no size limits | -| Typing indicators | ✅ | 🚧 | TUI shows status | +| Typing indicators | ✅ | 🚧 | TUI + Telegram typing/actionable status prompts; richer parity pending | | Per-channel ackReaction config | ✅ | ❌ | Customizable acknowledgement reactions | | Group session priming | ✅ | ❌ | Member roster injected for context | | Sender_id in trusted metadata | ✅ | ❌ | Exposed in system metadata | diff --git a/channels-src/telegram/Cargo.toml b/channels-src/telegram/Cargo.toml index 1964e327..83e0c8e0 100644 --- a/channels-src/telegram/Cargo.toml +++ b/channels-src/telegram/Cargo.toml @@ -17,7 +17,6 @@ serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" # Exclude from parent workspace (this is a standalone WASM component) -[workspace] [profile.release] # Optimize for size @@ -25,3 +24,5 @@ opt-level = "s" lto = true strip = true codegen-units = 1 + +[workspace] diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs index 5c2f91af..c1f8539a 100644 --- a/channels-src/telegram/src/lib.rs +++ b/channels-src/telegram/src/lib.rs @@ -244,6 +244,67 @@ struct TelegramConfig { struct TelegramChannel; +#[derive(Debug, Clone, PartialEq, Eq)] +enum TelegramStatusAction { + Typing, + Notify(String), +} + +const TELEGRAM_STATUS_MAX_CHARS: usize = 600; + +fn truncate_status_message(input: &str, max_chars: usize) -> String { + let mut iter = input.chars(); + let truncated: String = iter.by_ref().take(max_chars).collect(); + if iter.next().is_some() { + format!("{}...", truncated) + } else { + truncated + } +} + +fn status_message_for_user(update: &StatusUpdate) -> Option { + let message = update.message.trim(); + if message.is_empty() { + None + } else { + Some(truncate_status_message(message, TELEGRAM_STATUS_MAX_CHARS)) + } +} + +fn get_updates_url(offset: i64, timeout_secs: u32) -> String { + format!( + "https://api.telegram.org/bot{{TELEGRAM_BOT_TOKEN}}/getUpdates?offset={}&timeout={}&allowed_updates=[\"message\",\"edited_message\"]", + offset, timeout_secs + ) +} + +fn classify_status_update(update: &StatusUpdate) -> Option { + match update.status { + StatusType::Thinking => Some(TelegramStatusAction::Typing), + StatusType::Done | StatusType::Interrupted => None, + // Tool telemetry can be noisy in chat; keep it as typing-only UX. + StatusType::ToolStarted | StatusType::ToolCompleted | StatusType::ToolResult => None, + StatusType::Status => { + let msg = update.message.trim(); + if msg.eq_ignore_ascii_case("Done") + || msg.eq_ignore_ascii_case("Interrupted") + || msg.eq_ignore_ascii_case("Awaiting approval") + || msg.eq_ignore_ascii_case("Rejected") + { + None + } else { + status_message_for_user(update).map(TelegramStatusAction::Notify) + } + } + StatusType::ApprovalNeeded + | StatusType::JobStarted + | StatusType::AuthRequired + | StatusType::AuthCompleted => { + status_message_for_user(update).map(TelegramStatusAction::Notify) + } + } +} + impl Guest for TelegramChannel { fn on_start(config_json: String) -> Result { channel_host::log( @@ -422,20 +483,36 @@ impl Guest for TelegramChannel { &format!("Polling getUpdates with offset {}", offset), ); - // Build getUpdates URL with parameters - // - offset: Identifier of the first update to be returned - // - timeout: Long polling timeout in seconds (Telegram recommends 30+) - // - allowed_updates: Only get message updates - let url = format!( - "https://api.telegram.org/bot{{TELEGRAM_BOT_TOKEN}}/getUpdates?offset={}&timeout=30&allowed_updates=[\"message\",\"edited_message\"]", - offset - ); + let headers_json = serde_json::json!({}).to_string(); + let primary_url = get_updates_url(offset, 30); - let headers = serde_json::json!({}); + // 35s HTTP timeout outlives Telegram's 30s server-side long-poll. + // If the TCP connection drops, retry once immediately with a short poll + // so we don't wait a full extra tick (~30s) before delivering updates. + let result = match channel_host::http_request( + "GET", + &primary_url, + &headers_json, + None, + Some(35_000), + ) { + Ok(response) => Ok(response), + Err(primary_err) => { + channel_host::log( + channel_host::LogLevel::Warn, + &format!( + "getUpdates request failed ({}), retrying once immediately", + primary_err + ), + ); - // 35s HTTP timeout outlives Telegram's 30s server-side long-poll - let result = - channel_host::http_request("GET", &url, &headers.to_string(), None, Some(35_000)); + let retry_url = get_updates_url(offset, 3); + channel_host::http_request("GET", &retry_url, &headers_json, None, Some(8_000)) + .map_err(|retry_err| { + format!("primary error: {}; retry error: {}", primary_err, retry_err) + }) + } + }; match result { Ok(response) => { @@ -516,7 +593,7 @@ impl Guest for TelegramChannel { let result = send_message( metadata.chat_id, &response.content, - metadata.message_id, + Some(metadata.message_id), Some("Markdown"), ); @@ -539,7 +616,7 @@ impl Guest for TelegramChannel { let msg_id = send_message( metadata.chat_id, &response.content, - metadata.message_id, + Some(metadata.message_id), None, ) .map_err(|e| format!("Plain-text retry also failed: {}", e))?; @@ -558,10 +635,10 @@ impl Guest for TelegramChannel { } fn on_status(update: StatusUpdate) { - // Only send typing indicator for Thinking status - if !matches!(update.status, StatusType::Thinking) { - return; - } + let action = match classify_status_update(&update) { + Some(action) => action, + None => return, + }; // Parse chat_id from metadata let metadata: TelegramMessageMetadata = match serde_json::from_str(&update.metadata_json) { @@ -569,40 +646,68 @@ impl Guest for TelegramChannel { Err(_) => { channel_host::log( channel_host::LogLevel::Debug, - "on_status: no valid Telegram metadata, skipping typing indicator", + "on_status: no valid Telegram metadata, skipping status update", ); return; } }; - // POST /sendChatAction with action "typing" - let payload = serde_json::json!({ - "chat_id": metadata.chat_id, - "action": "typing" - }); + match action { + TelegramStatusAction::Typing => { + // POST /sendChatAction with action "typing" + let payload = serde_json::json!({ + "chat_id": metadata.chat_id, + "action": "typing" + }); - let payload_bytes = match serde_json::to_vec(&payload) { - Ok(b) => b, - Err(_) => return, - }; + let payload_bytes = match serde_json::to_vec(&payload) { + Ok(b) => b, + Err(_) => return, + }; - let headers = serde_json::json!({ - "Content-Type": "application/json" - }); + let headers = serde_json::json!({ + "Content-Type": "application/json" + }); - let result = channel_host::http_request( - "POST", - "https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendChatAction", - &headers.to_string(), - Some(&payload_bytes), - None, - ); + let result = channel_host::http_request( + "POST", + "https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendChatAction", + &headers.to_string(), + Some(&payload_bytes), + None, + ); - if let Err(e) = result { - channel_host::log( - channel_host::LogLevel::Debug, - &format!("sendChatAction failed: {}", e), - ); + if let Err(e) = result { + channel_host::log( + channel_host::LogLevel::Debug, + &format!("sendChatAction failed: {}", e), + ); + } + } + 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) + { + channel_host::log( + channel_host::LogLevel::Warn, + &format!( + "Failed to send status reply ({}), retrying without reply context", + first_err + ), + ); + + if let Err(retry_err) = send_message(metadata.chat_id, &prompt, None, None) { + channel_host::log( + channel_host::LogLevel::Debug, + &format!( + "Failed to send status message without reply context: {}", + retry_err + ), + ); + } + } + } } } @@ -643,15 +748,18 @@ impl std::fmt::Display for SendError { fn send_message( chat_id: i64, text: &str, - reply_to_message_id: i64, + reply_to_message_id: Option, parse_mode: Option<&str>, ) -> Result { let mut payload = serde_json::json!({ "chat_id": chat_id, "text": text, - "reply_to_message_id": reply_to_message_id, }); + if let Some(message_id) = reply_to_message_id { + payload["reply_to_message_id"] = serde_json::Value::Number(message_id.into()); + } + if let Some(mode) = parse_mode { payload["parse_mode"] = serde_json::Value::String(mode.to_string()); } @@ -831,40 +939,17 @@ fn register_webhook(tunnel_url: &str, webhook_secret: Option<&str>) -> Result<() /// Send a pairing code message to a chat. Used when an unknown user DMs the bot. fn send_pairing_reply(chat_id: i64, code: &str) -> Result<(), String> { - let payload = serde_json::json!({ - "chat_id": chat_id, - "text": format!( + send_message( + chat_id, + &format!( "To pair with this bot, run: `ironclaw pairing approve telegram {}`", code ), - "parse_mode": "Markdown", - }); - - let payload_bytes = - serde_json::to_vec(&payload).map_err(|e| format!("Failed to serialize payload: {}", e))?; - - let headers = serde_json::json!({ - "Content-Type": "application/json" - }); - - let result = channel_host::http_request( - "POST", - "https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage", - &headers.to_string(), - Some(&payload_bytes), None, - ); - - match result { - Ok(response) => { - if response.status != 200 { - let body_str = String::from_utf8_lossy(&response.body); - return Err(format!("HTTP {}: {}", response.status, body_str)); - } - Ok(()) - } - Err(e) => Err(format!("HTTP request failed: {}", e)), - } + Some("Markdown"), + ) + .map(|_| ()) + .map_err(|e| e.to_string()) } // ============================================================================ @@ -1027,33 +1112,17 @@ fn handle_message(message: TelegramMessage) { let metadata_json = serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); - // Clean the message text (strip bot mentions and commands) let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); - let cleaned_text = clean_message_text( + let content_to_emit = match content_to_emit_for_agent( &content, if bot_username.is_empty() { None } else { Some(bot_username.as_str()) }, - ); - - // Determine what to emit to the agent. - // - `/start` (no args): emit a welcome placeholder so the agent greets the user - // - Other bare `/commands` (e.g. /interrupt, /help): pass the raw command through - // so Submission::parse() can handle it - // - Commands with args (e.g. `/start hello`): cleaned_text already has the args - // - Plain text: pass through as-is - let trimmed_content = content.trim(); - let content_to_emit = if trimmed_content.eq_ignore_ascii_case("/start") { - "[User started the bot]".to_string() - } else if cleaned_text.is_empty() && trimmed_content.starts_with('/') { - // Bare control command like /interrupt, /stop, /help — pass through raw - trimmed_content.to_string() - } else if cleaned_text.is_empty() { - return; - } else { - cleaned_text + ) { + Some(value) => value, + None => return, }; // Emit the message to the agent @@ -1121,6 +1190,31 @@ fn clean_message_text(text: &str, bot_username: Option<&str>) -> String { result } +/// Decide which user content should be emitted to the agent loop. +/// +/// - `/start` emits a placeholder so the agent can greet the user +/// - bare slash commands are passed through for Submission parsing +/// - empty/mention-only messages are ignored +/// - otherwise cleaned text is emitted +fn content_to_emit_for_agent(content: &str, bot_username: Option<&str>) -> Option { + let cleaned_text = clean_message_text(content, bot_username); + let trimmed_content = content.trim(); + + if trimmed_content.eq_ignore_ascii_case("/start") { + return Some("[User started the bot]".to_string()); + } + + if cleaned_text.is_empty() && trimmed_content.starts_with('/') { + return Some(trimmed_content.to_string()); + } + + if cleaned_text.is_empty() { + return None; + } + + Some(cleaned_text) +} + // ============================================================================ // Utilities // ============================================================================ @@ -1181,62 +1275,126 @@ mod tests { // Commands with args: command prefix stripped, args returned assert_eq!(clean_message_text("/start hello", None), "hello"); assert_eq!(clean_message_text("/help me please", None), "me please"); - assert_eq!(clean_message_text("/model claude-opus-4-6", None), "claude-opus-4-6"); + assert_eq!( + clean_message_text("/model claude-opus-4-6", None), + "claude-opus-4-6" + ); } /// Tests for the content_to_emit logic in handle_message. - /// Since handle_message uses WASM host calls, we test the decision logic inline. + /// Since handle_message uses WASM host calls, test the extracted decision function. #[test] fn test_content_to_emit_logic() { - // Simulates the content_to_emit decision for various inputs. - // This mirrors the logic in handle_message after clean_message_text. - fn resolve_content(content: &str) -> Option { - let cleaned_text = clean_message_text(content, None); - let trimmed_content = content.trim(); - if trimmed_content.eq_ignore_ascii_case("/start") { - Some("[User started the bot]".to_string()) - } else if cleaned_text.is_empty() && trimmed_content.starts_with('/') { - Some(trimmed_content.to_string()) - } else if cleaned_text.is_empty() { - None // would return/skip in handle_message - } else { - Some(cleaned_text) - } - } - // /start → welcome placeholder - assert_eq!(resolve_content("/start"), Some("[User started the bot]".to_string())); - assert_eq!(resolve_content("/Start"), Some("[User started the bot]".to_string())); - assert_eq!(resolve_content(" /start "), Some("[User started the bot]".to_string())); + assert_eq!( + content_to_emit_for_agent("/start", None), + Some("[User started the bot]".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/Start", None), + Some("[User started the bot]".to_string()) + ); + assert_eq!( + content_to_emit_for_agent(" /start ", None), + Some("[User started the bot]".to_string()) + ); // /start with args → pass args through - assert_eq!(resolve_content("/start hello"), Some("hello".to_string())); + assert_eq!( + content_to_emit_for_agent("/start hello", None), + Some("hello".to_string()) + ); // Control commands → pass through raw so Submission::parse() can match - assert_eq!(resolve_content("/interrupt"), Some("/interrupt".to_string())); - assert_eq!(resolve_content("/stop"), Some("/stop".to_string())); - assert_eq!(resolve_content("/help"), Some("/help".to_string())); - assert_eq!(resolve_content("/undo"), Some("/undo".to_string())); - assert_eq!(resolve_content("/redo"), Some("/redo".to_string())); - assert_eq!(resolve_content("/ping"), Some("/ping".to_string())); - assert_eq!(resolve_content("/tools"), Some("/tools".to_string())); - assert_eq!(resolve_content("/compact"), Some("/compact".to_string())); - assert_eq!(resolve_content("/clear"), Some("/clear".to_string())); - assert_eq!(resolve_content("/version"), Some("/version".to_string())); + assert_eq!( + content_to_emit_for_agent("/interrupt", None), + Some("/interrupt".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/stop", None), + Some("/stop".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/help", None), + Some("/help".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/undo", None), + Some("/undo".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/redo", None), + Some("/redo".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/ping", None), + Some("/ping".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/tools", None), + Some("/tools".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/compact", None), + Some("/compact".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/clear", None), + Some("/clear".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/version", None), + Some("/version".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/approve", None), + Some("/approve".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/always", None), + Some("/always".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/deny", None), + Some("/deny".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/yes", None), + Some("/yes".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("/no", None), + Some("/no".to_string()) + ); // Commands with args → cleaned text (command stripped) - assert_eq!(resolve_content("/help me please"), Some("me please".to_string())); + assert_eq!( + content_to_emit_for_agent("/help me please", None), + Some("me please".to_string()) + ); // Plain text → pass through - assert_eq!(resolve_content("hello world"), Some("hello world".to_string())); - assert_eq!(resolve_content("just text"), Some("just text".to_string())); + assert_eq!( + content_to_emit_for_agent("hello world", None), + Some("hello world".to_string()) + ); + assert_eq!( + content_to_emit_for_agent("just text", None), + Some("just text".to_string()) + ); // Empty / whitespace → skip (None) - assert_eq!(resolve_content(""), None); - assert_eq!(resolve_content(" "), None); + assert_eq!(content_to_emit_for_agent("", None), None); + assert_eq!(content_to_emit_for_agent(" ", None), None); // Bare @mention without bot → skip - assert_eq!(resolve_content("@botname"), None); + assert_eq!(content_to_emit_for_agent("@botname", None), None); + + // With bot username configured: other mentions are preserved. + assert_eq!( + content_to_emit_for_agent("@alice hello", Some("MyBot")), + Some("@alice hello".to_string()) + ); } #[test] @@ -1317,4 +1475,236 @@ mod tests { assert_eq!(msg.text, None); assert_eq!(msg.caption.as_deref(), Some("What's in this image?")); } + + #[test] + fn test_get_updates_url_includes_offset_and_timeout() { + let url = get_updates_url(444_809_884, 30); + assert!(url.contains("offset=444809884")); + assert!(url.contains("timeout=30")); + assert!(url.contains("allowed_updates=[\"message\",\"edited_message\"]")); + } + + #[test] + fn test_classify_status_update_thinking() { + let update = StatusUpdate { + status: StatusType::Thinking, + message: "Thinking...".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Typing) + ); + } + + #[test] + fn test_classify_status_update_approval_needed() { + let update = StatusUpdate { + status: StatusType::ApprovalNeeded, + message: "Approval needed for tool 'http_request'".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Notify( + "Approval needed for tool 'http_request'".to_string() + )) + ); + } + + #[test] + fn test_classify_status_update_done_ignored() { + let update = StatusUpdate { + status: StatusType::Done, + message: "Done".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_auth_required() { + let update = StatusUpdate { + status: StatusType::AuthRequired, + message: "Authentication required for weather.".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Notify( + "Authentication required for weather.".to_string() + )) + ); + } + + #[test] + fn test_classify_status_update_tool_started_ignored() { + let update = StatusUpdate { + status: StatusType::ToolStarted, + message: "Tool started: http_request".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_tool_completed_ignored() { + let update = StatusUpdate { + status: StatusType::ToolCompleted, + message: "Tool completed: http_request (ok)".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_job_started_notify() { + let update = StatusUpdate { + status: StatusType::JobStarted, + message: "Job started: Daily sync".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Notify( + "Job started: Daily sync".to_string() + )) + ); + } + + #[test] + fn test_classify_status_update_auth_completed_notify() { + let update = StatusUpdate { + status: StatusType::AuthCompleted, + message: "Authentication completed for weather.".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Notify( + "Authentication completed for weather.".to_string() + )) + ); + } + + #[test] + fn test_classify_status_update_tool_result_ignored() { + let update = StatusUpdate { + status: StatusType::ToolResult, + message: "Tool result: http_request ...".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_awaiting_approval_ignored() { + let update = StatusUpdate { + status: StatusType::Status, + message: "Awaiting approval".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_interrupted_ignored() { + let update = StatusUpdate { + status: StatusType::Interrupted, + message: "Interrupted".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_status_done_ignored_case_insensitive() { + let update = StatusUpdate { + status: StatusType::Status, + message: "done".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_status_interrupted_ignored() { + let update = StatusUpdate { + status: StatusType::Status, + message: "interrupted".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_status_rejected_ignored() { + let update = StatusUpdate { + status: StatusType::Status, + message: "Rejected".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(classify_status_update(&update), None); + } + + #[test] + fn test_classify_status_update_status_notify() { + let update = StatusUpdate { + status: StatusType::Status, + message: "Context compaction started".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!( + classify_status_update(&update), + Some(TelegramStatusAction::Notify( + "Context compaction started".to_string() + )) + ); + } + + #[test] + fn test_status_message_for_user_ignores_blank() { + let update = StatusUpdate { + status: StatusType::AuthRequired, + message: " ".to_string(), + metadata_json: "{}".to_string(), + }; + + assert_eq!(status_message_for_user(&update), None); + } + + #[test] + fn test_truncate_status_message_appends_ellipsis() { + let input = "abcdefghijklmnopqrstuvwxyz"; + let output = truncate_status_message(input, 10); + assert_eq!(output, "abcdefghij..."); + } + + #[test] + fn test_status_message_for_user_truncates_long_input() { + let update = StatusUpdate { + status: StatusType::AuthRequired, + message: "x".repeat(700), + metadata_json: "{}".to_string(), + }; + + let msg = status_message_for_user(&update).expect("expected message"); + assert!(msg.len() <= TELEGRAM_STATUS_MAX_CHARS + 3); + assert!(msg.ends_with("...")); + } } diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 91bf655f..3b8e5759 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -1375,13 +1375,25 @@ impl WasmChannel { /// that repeats the call every 4 seconds (Telegram's typing indicator /// expires after ~5s). /// - /// On Done/Interrupted/Status: cancels the repeat task, fires on_status once. + /// On terminal or user-action-required states: cancels the repeat task, + /// then fires on_status once. + /// + /// On intermediate progress states (tool/auth/job/status updates), keeps + /// the typing repeater running and fires on_status once. /// On StreamChunk: no-op (too noisy). async fn handle_status_update( &self, status: StatusUpdate, metadata: &serde_json::Value, ) -> Result<(), ChannelError> { + fn is_terminal_text_status(msg: &str) -> bool { + let trimmed = msg.trim(); + trimmed.eq_ignore_ascii_case("done") + || trimmed.eq_ignore_ascii_case("interrupted") + || trimmed.eq_ignore_ascii_case("awaiting approval") + || trimmed.eq_ignore_ascii_case("rejected") + } + match &status { StatusUpdate::Thinking(_) => { // Cancel any existing typing task @@ -1508,8 +1520,8 @@ impl WasmChannel { let _ = self.call_on_status(&status, metadata).await; } } - _ => { - // Done, Interrupted, Status, ToolStarted, ToolCompleted: cancel and fire once + StatusUpdate::AuthRequired { .. } => { + // Waiting on user action: stop typing and fire once. self.cancel_typing_task().await; if let Err(e) = self.call_on_status(&status, metadata).await { @@ -1520,6 +1532,28 @@ impl WasmChannel { ); } } + StatusUpdate::Status(msg) if is_terminal_text_status(msg) => { + // Waiting on user or terminal states: stop typing and fire once. + self.cancel_typing_task().await; + + if let Err(e) = self.call_on_status(&status, metadata).await { + tracing::debug!( + channel = %self.name, + error = %e, + "on_status failed (best-effort)" + ); + } + } + _ => { + // Intermediate progress status: keep any existing typing task alive. + if let Err(e) = self.call_on_status(&status, metadata).await { + tracing::debug!( + channel = %self.name, + error = %e, + "on_status failed (best-effort)" + ); + } + } } Ok(()) @@ -2126,6 +2160,16 @@ fn convert_http_response(wit: wit_channel::OutgoingHttpResponse) -> HttpResponse } /// Convert a StatusUpdate + metadata into the WIT StatusUpdate type. +fn truncate_status_text(input: &str, max_chars: usize) -> String { + let mut iter = input.chars(); + let truncated: String = iter.by_ref().take(max_chars).collect(); + if iter.next().is_some() { + format!("{}...", truncated) + } else { + truncated + } +} + fn status_to_wit(status: &StatusUpdate, metadata: &serde_json::Value) -> wit_channel::StatusUpdate { let metadata_json = serde_json::to_string(metadata).unwrap_or_default(); @@ -2137,17 +2181,25 @@ fn status_to_wit(status: &StatusUpdate, metadata: &serde_json::Value) -> wit_cha }, StatusUpdate::ToolStarted { name } => wit_channel::StatusUpdate { status: wit_channel::StatusType::ToolStarted, - message: name.clone(), + message: format!("Tool started: {}", name), metadata_json, }, StatusUpdate::ToolCompleted { name, success } => wit_channel::StatusUpdate { status: wit_channel::StatusType::ToolCompleted, - message: format!("{}: {}", name, if *success { "ok" } else { "failed" }), + message: format!( + "Tool completed: {} ({})", + name, + if *success { "ok" } else { "failed" } + ), metadata_json, }, StatusUpdate::ToolResult { name, preview } => wit_channel::StatusUpdate { - status: wit_channel::StatusType::ToolCompleted, - message: format!("{}: {}", name, preview), + status: wit_channel::StatusType::ToolResult, + message: format!( + "Tool result: {}\n{}", + name, + truncate_status_text(preview, 280) + ), metadata_json, }, StatusUpdate::StreamChunk(chunk) => wit_channel::StatusUpdate { @@ -2156,11 +2208,16 @@ fn status_to_wit(status: &StatusUpdate, metadata: &serde_json::Value) -> wit_cha metadata_json, }, StatusUpdate::Status(msg) => { - // Map well-known status strings to WIT types - let status_type = match msg.as_str() { - "Done" => wit_channel::StatusType::Done, - "Interrupted" => wit_channel::StatusType::Interrupted, - _ => wit_channel::StatusType::Thinking, + // Map well-known status strings to WIT types (case-insensitive + // to stay consistent with is_terminal_text_status and the + // Telegram-side classify_status_update). + let trimmed = msg.trim(); + let status_type = if trimmed.eq_ignore_ascii_case("done") { + wit_channel::StatusType::Done + } else if trimmed.eq_ignore_ascii_case("interrupted") { + wit_channel::StatusType::Interrupted + } else { + wit_channel::StatusType::Status }; wit_channel::StatusUpdate { status: status_type, @@ -2169,34 +2226,62 @@ fn status_to_wit(status: &StatusUpdate, metadata: &serde_json::Value) -> wit_cha } } StatusUpdate::ApprovalNeeded { + request_id, tool_name, description, .. } => wit_channel::StatusUpdate { - status: wit_channel::StatusType::Thinking, - message: format!("Approval needed: {} - {}", tool_name, description), + status: wit_channel::StatusType::ApprovalNeeded, + message: format!( + "Approval needed for tool '{}'. {}\nRequest ID: {}\nReply with: yes (or /approve), no (or /deny), or always (or /always).", + tool_name, description, request_id + ), metadata_json, }, - StatusUpdate::JobStarted { job_id, title, .. } => wit_channel::StatusUpdate { - status: wit_channel::StatusType::Thinking, - message: format!("Job started: {} ({})", title, job_id), + StatusUpdate::JobStarted { + job_id, + title, + browse_url, + } => wit_channel::StatusUpdate { + status: wit_channel::StatusType::JobStarted, + message: format!("Job started: {} ({})\n{}", title, job_id, browse_url), metadata_json, }, - StatusUpdate::AuthRequired { extension_name, .. } => wit_channel::StatusUpdate { - status: wit_channel::StatusType::Thinking, - message: format!("Auth required: {}", extension_name), + StatusUpdate::AuthRequired { + extension_name, + instructions, + auth_url, + setup_url, + } => wit_channel::StatusUpdate { + status: wit_channel::StatusType::AuthRequired, + message: { + let mut lines = vec![format!("Authentication required for {}.", extension_name)]; + if let Some(text) = instructions + && !text.trim().is_empty() + { + lines.push(text.trim().to_string()); + } + if let Some(url) = auth_url { + lines.push(format!("Auth URL: {}", url)); + } + if let Some(url) = setup_url { + lines.push(format!("Setup URL: {}", url)); + } + lines.join("\n") + }, metadata_json, }, StatusUpdate::AuthCompleted { extension_name, success, - .. + message, } => wit_channel::StatusUpdate { - status: wit_channel::StatusType::Thinking, + status: wit_channel::StatusType::AuthCompleted, message: format!( - "Auth {}: {}", + "Authentication {} for {}. {}", if *success { "completed" } else { "failed" }, - extension_name + extension_name, + message ), metadata_json, }, @@ -2212,6 +2297,12 @@ fn clone_wit_status_update(update: &wit_channel::StatusUpdate) -> wit_channel::S wit_channel::StatusType::Interrupted => wit_channel::StatusType::Interrupted, wit_channel::StatusType::ToolStarted => wit_channel::StatusType::ToolStarted, wit_channel::StatusType::ToolCompleted => wit_channel::StatusType::ToolCompleted, + wit_channel::StatusType::ToolResult => wit_channel::StatusType::ToolResult, + wit_channel::StatusType::ApprovalNeeded => wit_channel::StatusType::ApprovalNeeded, + wit_channel::StatusType::Status => wit_channel::StatusType::Status, + wit_channel::StatusType::JobStarted => wit_channel::StatusType::JobStarted, + wit_channel::StatusType::AuthRequired => wit_channel::StatusType::AuthRequired, + wit_channel::StatusType::AuthCompleted => wit_channel::StatusType::AuthCompleted, }, message: update.message.clone(), metadata_json: update.metadata_json.clone(), @@ -2555,6 +2646,100 @@ mod tests { channel.shutdown().await.expect("Shutdown should succeed"); } + #[tokio::test] + async fn test_typing_task_persists_on_tool_started() { + let channel = create_test_channel(); + let _stream = channel.start().await.expect("Channel should start"); + + let metadata = serde_json::json!({"chat_id": 123}); + + // Start typing + let _ = channel + .send_status( + crate::channels::StatusUpdate::Thinking("Processing...".into()), + &metadata, + ) + .await; + assert!(channel.typing_task.read().await.is_some()); + + // Intermediate tool status should not cancel typing + let _ = channel + .send_status( + crate::channels::StatusUpdate::ToolStarted { + name: "http_request".into(), + }, + &metadata, + ) + .await; + + assert!(channel.typing_task.read().await.is_some()); + + channel.shutdown().await.expect("Shutdown should succeed"); + } + + #[tokio::test] + async fn test_typing_task_cancelled_on_approval_needed() { + let channel = create_test_channel(); + let _stream = channel.start().await.expect("Channel should start"); + + let metadata = serde_json::json!({"chat_id": 123}); + + // Start typing + let _ = channel + .send_status( + crate::channels::StatusUpdate::Thinking("Processing...".into()), + &metadata, + ) + .await; + assert!(channel.typing_task.read().await.is_some()); + + // Approval-needed should stop typing while waiting for user action + let _ = channel + .send_status( + crate::channels::StatusUpdate::ApprovalNeeded { + request_id: "req-1".into(), + tool_name: "http_request".into(), + description: "Fetch weather".into(), + parameters: serde_json::json!({"url": "https://wttr.in"}), + }, + &metadata, + ) + .await; + + assert!(channel.typing_task.read().await.is_none()); + + channel.shutdown().await.expect("Shutdown should succeed"); + } + + #[tokio::test] + async fn test_typing_task_cancelled_on_awaiting_approval_status() { + let channel = create_test_channel(); + let _stream = channel.start().await.expect("Channel should start"); + + let metadata = serde_json::json!({"chat_id": 123}); + + // Start typing + let _ = channel + .send_status( + crate::channels::StatusUpdate::Thinking("Processing...".into()), + &metadata, + ) + .await; + assert!(channel.typing_task.read().await.is_some()); + + // Legacy terminal status string should also cancel typing + let _ = channel + .send_status( + crate::channels::StatusUpdate::Status("Awaiting approval".into()), + &metadata, + ) + .await; + + assert!(channel.typing_task.read().await.is_none()); + + channel.shutdown().await.expect("Shutdown should succeed"); + } + #[tokio::test] async fn test_typing_task_replaced_on_new_thinking() { let channel = create_test_channel(); @@ -2678,6 +2863,27 @@ mod tests { assert!(matches!(wit.status, super::wit_channel::StatusType::Done)); } + #[test] + fn test_status_to_wit_done_case_insensitive() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + + // lowercase + let wit = status_to_wit( + &crate::channels::StatusUpdate::Status("done".into()), + &metadata, + ); + assert!(matches!(wit.status, super::wit_channel::StatusType::Done)); + + // with whitespace + let wit = status_to_wit( + &crate::channels::StatusUpdate::Status(" Done ".into()), + &metadata, + ); + assert!(matches!(wit.status, super::wit_channel::StatusType::Done)); + } + #[test] fn test_status_to_wit_interrupted() { use super::status_to_wit; @@ -2694,6 +2900,311 @@ mod tests { )); } + #[test] + fn test_status_to_wit_interrupted_case_insensitive() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + + // lowercase + let wit = status_to_wit( + &crate::channels::StatusUpdate::Status("interrupted".into()), + &metadata, + ); + assert!(matches!( + wit.status, + super::wit_channel::StatusType::Interrupted + )); + + // with whitespace + let wit = status_to_wit( + &crate::channels::StatusUpdate::Status(" Interrupted ".into()), + &metadata, + ); + assert!(matches!( + wit.status, + super::wit_channel::StatusType::Interrupted + )); + } + + #[test] + fn test_status_to_wit_generic_status() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::Status("Awaiting approval".into()), + &metadata, + ); + + assert!(matches!(wit.status, super::wit_channel::StatusType::Status)); + assert_eq!(wit.message, "Awaiting approval"); + } + + #[test] + fn test_status_to_wit_auth_required() { + use super::status_to_wit; + + let metadata = serde_json::json!({"chat_id": 42}); + let wit = status_to_wit( + &crate::channels::StatusUpdate::AuthRequired { + extension_name: "weather".to_string(), + instructions: Some("Paste your token".to_string()), + auth_url: Some("https://example.com/auth".to_string()), + setup_url: None, + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::AuthRequired + )); + assert!(wit.message.contains("Authentication required for weather")); + assert!(wit.message.contains("Paste your token")); + } + + #[test] + fn test_status_to_wit_tool_started() { + use super::status_to_wit; + + let metadata = serde_json::json!({"chat_id": 7}); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ToolStarted { + name: "http_request".to_string(), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ToolStarted + )); + assert_eq!(wit.message, "Tool started: http_request"); + } + + #[test] + fn test_status_to_wit_tool_completed_success() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ToolCompleted { + name: "http_request".to_string(), + success: true, + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ToolCompleted + )); + assert_eq!(wit.message, "Tool completed: http_request (ok)"); + } + + #[test] + fn test_status_to_wit_tool_completed_failure() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ToolCompleted { + name: "http_request".to_string(), + success: false, + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ToolCompleted + )); + assert_eq!(wit.message, "Tool completed: http_request (failed)"); + } + + #[test] + fn test_status_to_wit_tool_result() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ToolResult { + name: "http_request".to_string(), + preview: "{".to_string() + "\"temperature\": 22}", + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ToolResult + )); + assert!(wit.message.starts_with("Tool result: http_request\n")); + } + + #[test] + fn test_status_to_wit_tool_result_truncates_preview() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let long_preview = "x".repeat(400); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ToolResult { + name: "big_tool".to_string(), + preview: long_preview, + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ToolResult + )); + assert!(wit.message.ends_with("...")); + } + + #[test] + fn test_status_to_wit_job_started() { + use super::status_to_wit; + + let metadata = serde_json::json!({"chat_id": 1}); + let wit = status_to_wit( + &crate::channels::StatusUpdate::JobStarted { + job_id: "job-1".to_string(), + title: "Daily sync".to_string(), + browse_url: "https://example.com/jobs/job-1".to_string(), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::JobStarted + )); + assert!(wit.message.contains("Daily sync")); + assert!(wit.message.contains("https://example.com/jobs/job-1")); + } + + #[test] + fn test_status_to_wit_auth_completed_success() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::AuthCompleted { + extension_name: "weather".to_string(), + success: true, + message: "Token saved".to_string(), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::AuthCompleted + )); + assert!(wit.message.contains("Authentication completed")); + assert!(wit.message.contains("Token saved")); + } + + #[test] + fn test_status_to_wit_auth_completed_failure() { + use super::status_to_wit; + + let metadata = serde_json::json!(null); + let wit = status_to_wit( + &crate::channels::StatusUpdate::AuthCompleted { + extension_name: "weather".to_string(), + success: false, + message: "Invalid token".to_string(), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::AuthCompleted + )); + assert!(wit.message.contains("Authentication failed")); + assert!(wit.message.contains("Invalid token")); + } + + #[test] + fn test_status_to_wit_approval_needed() { + use super::status_to_wit; + + let metadata = serde_json::json!({"chat_id": 42}); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ApprovalNeeded { + request_id: "req-123".to_string(), + tool_name: "http_request".to_string(), + description: "Fetch weather data".to_string(), + parameters: serde_json::json!({"url": "https://api.weather.test"}), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ApprovalNeeded + )); + assert!(wit.message.contains("http_request")); + assert!(wit.message.contains("/approve")); + } + + #[test] + fn test_approval_prompt_roundtrip_submission_aliases() { + use super::status_to_wit; + use crate::agent::submission::{Submission, SubmissionParser}; + + let metadata = serde_json::json!({"chat_id": 42}); + let wit = status_to_wit( + &crate::channels::StatusUpdate::ApprovalNeeded { + request_id: "req-321".to_string(), + tool_name: "http_request".to_string(), + description: "Fetch weather data".to_string(), + parameters: serde_json::json!({"url": "https://api.weather.test"}), + }, + &metadata, + ); + + assert!(matches!( + wit.status, + super::wit_channel::StatusType::ApprovalNeeded + )); + assert!(wit.message.contains("/approve")); + assert!(wit.message.contains("/deny")); + assert!(wit.message.contains("/always")); + + let approve = SubmissionParser::parse("/approve"); + assert!(matches!( + approve, + Submission::ApprovalResponse { + approved: true, + always: false + } + )); + + let deny = SubmissionParser::parse("/deny"); + assert!(matches!( + deny, + Submission::ApprovalResponse { + approved: false, + always: false + } + )); + + let always = SubmissionParser::parse("/always"); + assert!(matches!( + always, + Submission::ApprovalResponse { + approved: true, + always: true + } + )); + } + #[test] fn test_clone_wit_status_update() { use super::{clone_wit_status_update, wit_channel}; @@ -2710,6 +3221,78 @@ mod tests { assert_eq!(cloned.metadata_json, "{\"a\":1}"); } + #[test] + fn test_clone_wit_status_update_approval_needed() { + use super::{clone_wit_status_update, wit_channel}; + + let original = wit_channel::StatusUpdate { + status: wit_channel::StatusType::ApprovalNeeded, + message: "approval needed".to_string(), + metadata_json: "{\"chat_id\":42}".to_string(), + }; + + let cloned = clone_wit_status_update(&original); + assert!(matches!( + cloned.status, + wit_channel::StatusType::ApprovalNeeded + )); + assert_eq!(cloned.message, "approval needed"); + assert_eq!(cloned.metadata_json, "{\"chat_id\":42}"); + } + + #[test] + fn test_clone_wit_status_update_auth_completed() { + use super::{clone_wit_status_update, wit_channel}; + + let original = wit_channel::StatusUpdate { + status: wit_channel::StatusType::AuthCompleted, + message: "auth complete".to_string(), + metadata_json: "{}".to_string(), + }; + + let cloned = clone_wit_status_update(&original); + assert!(matches!( + cloned.status, + wit_channel::StatusType::AuthCompleted + )); + assert_eq!(cloned.message, "auth complete"); + } + + #[test] + fn test_clone_wit_status_update_all_variants() { + use super::{clone_wit_status_update, wit_channel}; + + let variants = vec![ + wit_channel::StatusType::Thinking, + wit_channel::StatusType::Done, + wit_channel::StatusType::Interrupted, + wit_channel::StatusType::ToolStarted, + wit_channel::StatusType::ToolCompleted, + wit_channel::StatusType::ToolResult, + wit_channel::StatusType::ApprovalNeeded, + wit_channel::StatusType::Status, + wit_channel::StatusType::JobStarted, + wit_channel::StatusType::AuthRequired, + wit_channel::StatusType::AuthCompleted, + ]; + + for status in variants { + let original = wit_channel::StatusUpdate { + status, + message: "sample".to_string(), + metadata_json: "{}".to_string(), + }; + let cloned = clone_wit_status_update(&original); + + assert_eq!( + std::mem::discriminant(&cloned.status), + std::mem::discriminant(&original.status) + ); + assert_eq!(cloned.message, "sample"); + assert_eq!(cloned.metadata_json, "{}"); + } + } + #[test] fn test_redact_credentials_replaces_values() { use super::ChannelStoreData; diff --git a/wit/channel.wit b/wit/channel.wit index c716bc58..6333e3cd 100644 --- a/wit/channel.wit +++ b/wit/channel.wit @@ -261,6 +261,18 @@ interface channel { tool-started, /// A tool execution completed. tool-completed, + /// A tool execution produced a preview/result status. + tool-result, + /// A tool call is waiting for user approval. + approval-needed, + /// Generic status text that should be shown to the user. + status, + /// A background/sandbox job was started. + job-started, + /// An extension/tool requires user authentication. + auth-required, + /// Authentication flow completed. + auth-completed, } /// A status update from the agent.