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.