From 0e75ae15644a9cc9647859b3e45adeda9fdc8ce0 Mon Sep 17 00:00:00 2001 From: XU Ke Date: Tue, 13 Oct 2026 21:56:20 +0800 Subject: [PATCH] fix(weixin): buffer immediately on expired context and report honest SendOutcome - Remove invalid tokenless send fallback for Weixin iLink backend - Add SendOutcome enum (Sent, Buffered, Dropped) for ChatChannelBackend - Weixin backend buffers immediately when context is marked expired without wasting network roundtrips - Enforce pending buffer bounded capacity (50) and age expiration (10m), returning Dropped on overflow - Prioritize inbound user command dispatch over replaying buffered backlog - Cap replayed messages to 3 and abort replay immediately on ret = -2 - Update event_subscriber, scheduler, and command_dispatcher to log honest delivery status and advance debounce only on delivered sends - Add extensive unit tests covering Weixin buffering lifecycle, drop semantics, and subscriber debounce handling --- src-tauri/src/chat_channel/backends/lark.rs | 8 +- .../src/chat_channel/backends/telegram.rs | 20 +- src-tauri/src/chat_channel/backends/weixin.rs | 421 ++++++++++++++---- .../src/chat_channel/command_dispatcher.rs | 2 +- .../src/chat_channel/event_subscriber.rs | 272 ++++++++++- src-tauri/src/chat_channel/manager.rs | 6 +- src-tauri/src/chat_channel/scheduler.rs | 2 +- .../chat_channel/session_event_subscriber.rs | 10 +- src-tauri/src/chat_channel/traits.rs | 10 +- src-tauri/src/chat_channel/types.rs | 49 +- src-tauri/src/commands/conversations.rs | 12 +- 11 files changed, 679 insertions(+), 133 deletions(-) diff --git a/src-tauri/src/chat_channel/backends/lark.rs b/src-tauri/src/chat_channel/backends/lark.rs index 1cb748be46..4a7a858f6b 100644 --- a/src-tauri/src/chat_channel/backends/lark.rs +++ b/src-tauri/src/chat_channel/backends/lark.rs @@ -219,7 +219,7 @@ impl LarkBackend { &self, msg_type: &str, content: &str, - ) -> Result { + ) -> Result { let token = self.get_tenant_access_token().await?; let resp = self @@ -251,7 +251,7 @@ impl LarkBackend { } let message_id = result.data.and_then(|d| d.message_id).unwrap_or_default(); - Ok(SentMessageId(message_id)) + Ok(SendOutcome::Sent(SentMessageId(message_id))) } async fn start_ws_receiver( @@ -603,7 +603,7 @@ impl ChatChannelBackend for LarkBackend { *self.status.lock().await } - async fn send_message(&self, text: &str) -> Result { + async fn send_message(&self, text: &str) -> Result { let content = serde_json::json!({ "text": text }).to_string(); self.send_lark_message("text", &content).await } @@ -611,7 +611,7 @@ impl ChatChannelBackend for LarkBackend { async fn send_rich_message( &self, message: &RichMessage, - ) -> Result { + ) -> Result { let card = build_lark_card(message); let content = serde_json::to_string(&card) .map_err(|e| ChatChannelError::SendFailed(e.to_string()))?; diff --git a/src-tauri/src/chat_channel/backends/telegram.rs b/src-tauri/src/chat_channel/backends/telegram.rs index 0149d7f3f2..3e252a299a 100644 --- a/src-tauri/src/chat_channel/backends/telegram.rs +++ b/src-tauri/src/chat_channel/backends/telegram.rs @@ -91,7 +91,7 @@ impl TelegramBackend { text: &str, parse_mode: Option<&str>, target: Option<&ChannelMessageTarget>, - ) -> Result { + ) -> Result { self.send_text_with_reply_markup(text, parse_mode, target, None) .await } @@ -102,7 +102,7 @@ impl TelegramBackend { parse_mode: Option<&str>, target: Option<&ChannelMessageTarget>, reply_markup: Option, - ) -> Result { + ) -> Result { let body = telegram_send_message_body(&self.chat_id, text, parse_mode, target, reply_markup)?; @@ -137,14 +137,14 @@ impl TelegramBackend { .map(|id| id.to_string()) .unwrap_or_default(); - Ok(SentMessageId(message_id)) + Ok(SendOutcome::Sent(SentMessageId(message_id))) } async fn send_interactive_message_with_target( &self, message: &InteractiveMessage, target: Option<&ChannelMessageTarget>, - ) -> Result { + ) -> Result { let reply_markup = telegram_inline_keyboard(message); let markdown_text = format_telegram_markdown(&message.base); let result = self @@ -177,7 +177,7 @@ impl TelegramBackend { &self, message: &RichMessage, target: Option<&ChannelMessageTarget>, - ) -> Result { + ) -> Result { let markdown_text = format_telegram_markdown(message); let result = self .send_text(&markdown_text, Some("MarkdownV2"), target) @@ -454,14 +454,14 @@ impl ChatChannelBackend for TelegramBackend { *self.status.lock().await } - async fn send_message(&self, text: &str) -> Result { + async fn send_message(&self, text: &str) -> Result { self.send_text(text, None, None).await } async fn send_rich_message( &self, message: &RichMessage, - ) -> Result { + ) -> Result { self.send_rich_message_with_target(message, None).await } @@ -469,7 +469,7 @@ impl ChatChannelBackend for TelegramBackend { &self, message: &RichMessage, target: &ChannelMessageTarget, - ) -> Result { + ) -> Result { self.send_rich_message_with_target(message, Some(target)) .await } @@ -477,7 +477,7 @@ impl ChatChannelBackend for TelegramBackend { async fn send_interactive_message( &self, message: &InteractiveMessage, - ) -> Result { + ) -> Result { self.send_interactive_message_with_target(message, None) .await } @@ -486,7 +486,7 @@ impl ChatChannelBackend for TelegramBackend { &self, message: &InteractiveMessage, target: &ChannelMessageTarget, - ) -> Result { + ) -> Result { self.send_interactive_message_with_target(message, Some(target)) .await } diff --git a/src-tauri/src/chat_channel/backends/weixin.rs b/src-tauri/src/chat_channel/backends/weixin.rs index 3e4e5a8c2e..52bccb1acf 100644 --- a/src-tauri/src/chat_channel/backends/weixin.rs +++ b/src-tauri/src/chat_channel/backends/weixin.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use async_trait::async_trait; use base64::{engine::general_purpose::STANDARD as B64, Engine}; @@ -14,8 +14,19 @@ use crate::chat_channel::types::*; const ILINK_BASE_URL: &str = "https://ilinkai.weixin.qq.com"; const ILINK_CHANNEL_VERSION: &str = "1.0.2"; +/// iLink reports a reply context it will not accept — typically a +/// `context_token` that expired after inactivity, or per-token quota exhaustion, +/// sometimes with `errmsg` "rate limited" or empty — as `ret = -2`. +const CONTEXT_REJECTED_RET: i64 = -2; /// Maximum number of messages buffered while context_token is expired. const MAX_PENDING_MESSAGES: usize = 50; +/// Buffered messages older than this are dropped instead of replayed. A push +/// nudge is only useful while it is fresh, and replaying a backlog would flood +/// the chat with hours-old notifications as soon as the user messages the bot. +const MAX_PENDING_AGE: Duration = Duration::from_secs(10 * 60); +/// Upper bound on messages replayed on a fresh context_token, protecting the +/// ~10-message quota per token from being consumed by backlogged notifications. +const MAX_REPLAY_MESSAGES: usize = 3; /// Shared HTTP client for QR code auth requests (avoids re-creating TLS state). fn qr_client() -> reqwest::Client { @@ -57,6 +68,8 @@ pub struct WeixinQrcodeStatusPublic { pub status: String, } +/// One `sendmessage` call. `context_token` may be empty: iLink accepts a +/// tokenless send as a degraded fallback (see `WeixinBackend::send_text`). struct SendRequest<'a> { client: &'a reqwest::Client, base_url: &'a str, @@ -65,8 +78,6 @@ struct SendRequest<'a> { to_user_id: &'a str, context_token: &'a str, text: &'a str, - reply_context: &'a Mutex>, - pending_messages: &'a Mutex>, } // ── QR code auth functions (called before backend exists) ── @@ -244,6 +255,26 @@ struct WeixinReplyContext { expired: bool, } +/// A message iLink would not accept yet, retried on the next context refresh. +#[derive(Clone, Debug)] +struct PendingMessage { + text: String, + queued_at: Instant, +} + +/// Queue `msg` for the next `context_token` refresh, or drop it when the +/// buffer is already full. +async fn queue_pending(buf: &Mutex>, msg: PendingMessage) -> SendOutcome { + let mut buf = buf.lock().await; + if buf.len() < MAX_PENDING_MESSAGES { + buf.push(msg); + SendOutcome::Buffered + } else { + tracing::warn!("[Weixin] pending buffer full, dropping message"); + SendOutcome::Dropped + } +} + pub struct WeixinBackend { bot_token: String, base_url: String, @@ -252,8 +283,10 @@ pub struct WeixinBackend { channel_id: i32, shutdown_tx: Arc>>>, reply_context: Arc>>, - /// Messages that failed due to expired context_token, resend on next refresh. - pending_messages: Arc>>, + /// Messages iLink rejected because the reply context was stale. They are + /// retried on the next context refresh, minus any that aged out while + /// waiting (see `MAX_PENDING_AGE`). + pending_messages: Arc>>, /// Stable X-WECHAT-UIN value for this backend instance. wechat_uin: String, } @@ -301,12 +334,17 @@ impl WeixinBackend { } /// Build the JSON body for the iLink sendmessage API. - fn build_send_body(to_user_id: &str, context_token: &str, text: &str) -> serde_json::Value { + fn build_send_body( + to_user_id: &str, + context_token: &str, + client_id: &str, + text: &str, + ) -> serde_json::Value { serde_json::json!({ "msg": { "from_user_id": "", "to_user_id": to_user_id, - "client_id": format!("codeg-{}", uuid::Uuid::new_v4()), + "client_id": client_id, "message_type": 2, "message_state": 2, "context_token": context_token, @@ -319,10 +357,14 @@ impl WeixinBackend { }) } - /// Send a message via the iLink API and handle the response. - /// Returns `Ok(true)` if sent, `Ok(false)` if buffered due to expired context. - async fn do_send(req: SendRequest<'_>) -> Result { - let body = Self::build_send_body(req.to_user_id, req.context_token, req.text); + /// One `sendmessage` call. + /// + /// Returns `Ok(Some(client_id))` when iLink accepted the message, or + /// `Ok(None)` when it rejected the reply context (see + /// `CONTEXT_REJECTED_RET`). + async fn do_send(req: SendRequest<'_>) -> Result, ChatChannelError> { + let client_id = format!("codeg-{}", uuid::Uuid::new_v4()); + let body = Self::build_send_body(req.to_user_id, req.context_token, &client_id, req.text); let url = format!("{}/ilink/bot/sendmessage", req.base_url); let resp = req @@ -343,7 +385,8 @@ impl WeixinBackend { ))); } - // Check for ret errors in response (e.g. -2 = context expired) + // A stale reply context or quota limit comes back as a non-zero `ret` + // even though the HTTP call succeeded, so inspect the body before calling it delivered. if let Ok(resp_json) = serde_json::from_str::(&resp_text) { if let Some(ret) = resp_json.get("ret").and_then(|v| v.as_i64()) { if ret != 0 { @@ -353,17 +396,8 @@ impl WeixinBackend { .unwrap_or("unknown"); tracing::info!("[Weixin] sendmessage ret={ret}, errmsg={errmsg}"); - if ret == -2 { - // Context token expired — mark stale and buffer - if let Some(ref mut c) = *req.reply_context.lock().await { - c.expired = true; - } - let mut buf = req.pending_messages.lock().await; - if buf.len() < MAX_PENDING_MESSAGES { - buf.push(req.text.to_string()); - } - tracing::info!("[Weixin] context_token expired (ret=-2), buffered message"); - return Ok(false); + if ret == CONTEXT_REJECTED_RET { + return Ok(None); } return Err(ChatChannelError::SendFailed(format!("ret={ret}: {errmsg}"))); @@ -371,10 +405,10 @@ impl WeixinBackend { } } - Ok(true) + Ok(Some(client_id)) } - async fn send_text(&self, text: &str) -> Result { + async fn send_text(&self, text: &str) -> Result { // Extract context data under lock, then release let (to_user_id, context_token, expired) = { let guard = self.reply_context.lock().await; @@ -391,19 +425,21 @@ impl WeixinBackend { ) }; - // If context is expired, buffer the message for resend on next refresh + // If context is already expired, buffer immediately without network calls. + // This avoids blocking other channels behind network timeouts while context is stale. if expired { tracing::info!( - "[Weixin] context expired, buffering message (len={})", + "[Weixin] context expired, buffering message immediately (len={})", text.len() ); - let mut buf = self.pending_messages.lock().await; - if buf.len() < MAX_PENDING_MESSAGES { - buf.push(text.to_string()); - } else { - tracing::info!("[Weixin] pending buffer full, dropping message"); - } - return Ok(SentMessageId(String::new())); + return Ok(queue_pending( + &self.pending_messages, + PendingMessage { + text: text.to_string(), + queued_at: Instant::now(), + }, + ) + .await); } tracing::info!( @@ -412,7 +448,7 @@ impl WeixinBackend { text.len() ); - Self::do_send(SendRequest { + let delivered = Self::do_send(SendRequest { client: &self.client, base_url: &self.base_url, bot_token: &self.bot_token, @@ -420,12 +456,31 @@ impl WeixinBackend { to_user_id: &to_user_id, context_token: &context_token, text, - reply_context: &self.reply_context, - pending_messages: &self.pending_messages, }) .await?; - Ok(SentMessageId(String::new())) + match delivered { + Some(client_id) => Ok(SendOutcome::Sent(SentMessageId(client_id))), + None => { + // Reply context rejected (ret = -2). Remember it is stale so later + // sends buffer immediately, and queue this message. + if let Some(ref mut c) = *self.reply_context.lock().await { + c.expired = true; + } + tracing::info!( + "[Weixin] context_token rejected (ret={CONTEXT_REJECTED_RET}), buffering message (len={})", + text.len() + ); + Ok(queue_pending( + &self.pending_messages, + PendingMessage { + text: text.to_string(), + queued_at: Instant::now(), + }, + ) + .await) + } + } } } @@ -643,56 +698,21 @@ impl ChatChannelBackend for WeixinBackend { // Store reply context for outbound messages // Single lock scope to avoid TOCTOU - if !from_user_id.is_empty() && !context_token.is_empty() { - let was_expired = { - let mut guard = reply_context.lock().await; - let was = - guard.as_ref().map(|c| c.expired).unwrap_or(false); - *guard = Some(WeixinReplyContext { - to_user_id: from_user_id.to_string(), - context_token: context_token.to_string(), - expired: false, - }); - was - }; - - // Resend buffered messages with fresh context - if was_expired { - let buffered: Vec = - pending_messages.lock().await.drain(..).collect(); - if !buffered.is_empty() { - tracing::info!( - "[Weixin] context refreshed, resending {} buffered message(s)", - buffered.len() - ); - for pending_text in &buffered { - let ok = WeixinBackend::do_send(SendRequest { - client: &client, - base_url: &base_url, - bot_token: &bot_token, - wechat_uin: &wechat_uin, - to_user_id: from_user_id, - context_token, - text: pending_text, - reply_context: &reply_context, - pending_messages: &pending_messages, - }) - .await; - if let Err(e) = ok { - tracing::error!("[Weixin] resend error: {e}"); - // Re-buffer remaining on hard error - let mut buf = pending_messages.lock().await; - if buf.len() < MAX_PENDING_MESSAGES { - buf.push(pending_text.clone()); - } - } - // If do_send returned Ok(false), it - // already re-buffered internally. - } - } - } - } + let was_expired = if !from_user_id.is_empty() && !context_token.is_empty() { + let mut guard = reply_context.lock().await; + let was = + guard.as_ref().map(|c| c.expired).unwrap_or(false); + *guard = Some(WeixinReplyContext { + to_user_id: from_user_id.to_string(), + context_token: context_token.to_string(), + expired: false, + }); + was + } else { + false + }; + // 1. Dispatch incoming user command FIRST before replaying buffered messages tracing::debug!("[Weixin] dispatching: {text}"); let send_result = command_tx .send(IncomingCommand { @@ -707,6 +727,82 @@ impl ChatChannelBackend for WeixinBackend { if let Err(e) = send_result { tracing::error!("[Weixin] command_tx.send failed: {e}"); } + + // 2. Retry what was queued while context was stale, bounded to MAX_REPLAY_MESSAGES + if was_expired && !from_user_id.is_empty() && !context_token.is_empty() { + let buffered: Vec = + pending_messages.lock().await.drain(..).collect(); + let now = Instant::now(); + let (stale, fresh): (Vec<_>, Vec<_>) = + buffered.into_iter().partition(|m| { + now.duration_since(m.queued_at) + >= MAX_PENDING_AGE + }); + if !stale.is_empty() { + tracing::info!( + "[Weixin] dropping {} stale buffered message(s) older than {:?}", + stale.len(), + MAX_PENDING_AGE + ); + } + if !fresh.is_empty() { + // Give command_dispatcher a brief head start to answer the user's message + tokio::time::sleep(Duration::from_millis(300)).await; + + let to_replay: Vec = + if fresh.len() > MAX_REPLAY_MESSAGES { + let skip = fresh.len() - MAX_REPLAY_MESSAGES; + tracing::info!( + "[Weixin] bounding replay: replaying latest {} of {} buffered messages", + MAX_REPLAY_MESSAGES, + fresh.len() + ); + fresh.into_iter().skip(skip).collect() + } else { + fresh + }; + + tracing::info!( + "[Weixin] context refreshed, resending {} buffered message(s)", + to_replay.len() + ); + + for pending in to_replay { + let result = WeixinBackend::do_send(SendRequest { + client: &client, + base_url: &base_url, + bot_token: &bot_token, + wechat_uin: &wechat_uin, + to_user_id: from_user_id, + context_token, + text: &pending.text, + }) + .await; + match result { + Ok(Some(_)) => {} + Ok(None) => { + // Context rejected (quota or token expired). + // Re-mark expired so subsequent sends buffer, re-queue, and stop replaying. + if let Some(ref mut c) = *reply_context.lock().await { + c.expired = true; + } + tracing::info!( + "[Weixin] context_token rejected during replay (ret={CONTEXT_REJECTED_RET}), re-queueing and stopping replay" + ); + let _ = queue_pending(&pending_messages, pending).await; + break; + } + Err(e) => { + tracing::error!( + "[Weixin] resend error: {e}" + ); + let _ = queue_pending(&pending_messages, pending).await; + break; + } + } + } + } + } } } } else { @@ -742,14 +838,14 @@ impl ChatChannelBackend for WeixinBackend { *self.status.lock().await } - async fn send_message(&self, text: &str) -> Result { + async fn send_message(&self, text: &str) -> Result { self.send_text(text).await } async fn send_rich_message( &self, message: &RichMessage, - ) -> Result { + ) -> Result { let plain_text = message.to_plain_text(); self.send_text(&plain_text).await } @@ -800,3 +896,150 @@ impl ChatChannelBackend for WeixinBackend { Ok(()) } } + +#[cfg(test)] +impl WeixinBackend { + pub(crate) async fn set_test_reply_context( + &self, + to_user_id: &str, + context_token: &str, + expired: bool, + ) { + *self.reply_context.lock().await = Some(WeixinReplyContext { + to_user_id: to_user_id.to_string(), + context_token: context_token.to_string(), + expired, + }); + } + + pub(crate) async fn is_context_expired(&self) -> bool { + self.reply_context + .lock() + .await + .as_ref() + .map(|c| c.expired) + .unwrap_or(false) + } + + pub(crate) async fn pending_count(&self) -> usize { + self.pending_messages.lock().await.len() + } + + pub(crate) async fn push_test_pending(&self, text: &str, age: Duration) { + let queued_at = Instant::now().checked_sub(age).unwrap_or_else(Instant::now); + self.pending_messages.lock().await.push(PendingMessage { + text: text.to_string(), + queued_at, + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + async fn spawn_mock_server( + status_code: u16, + body: &'static str, + ) -> (String, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { + if let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 4096]; + let _ = socket.read(&mut buf).await; + let resp = format!( + "HTTP/1.1 {status_code} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(resp.as_bytes()).await; + let _ = socket.flush().await; + } + }); + (format!("http://{addr}"), handle) + } + + #[tokio::test] + async fn test_send_success() { + let (base_url, _server) = spawn_mock_server(200, r#"{"ret": 0, "errmsg": "ok"}"#).await; + let backend = WeixinBackend::new(1, "test_token".into(), base_url); + backend + .set_test_reply_context("wx_user_1", "valid_token", false) + .await; + + let outcome = backend.send_message("hello world").await.unwrap(); + assert!(outcome.is_sent(), "expected sent outcome, got {outcome:?}"); + assert_eq!(backend.pending_count().await, 0); + assert!(!backend.is_context_expired().await); + } + + #[tokio::test] + async fn test_send_rejected_context_buffers_and_marks_expired() { + let (base_url, _server) = + spawn_mock_server(200, r#"{"ret": -2, "errmsg": "token expired"}"#).await; + let backend = WeixinBackend::new(1, "test_token".into(), base_url); + backend + .set_test_reply_context("wx_user_1", "stale_token", false) + .await; + + let outcome = backend.send_message("test nudge").await.unwrap(); + assert_eq!(outcome, SendOutcome::Buffered); + assert!(backend.is_context_expired().await, "should mark context as expired"); + assert_eq!(backend.pending_count().await, 1, "should buffer rejected message"); + } + + #[tokio::test] + async fn test_send_when_expired_buffers_immediately_without_network() { + // Point to closed port: if any network call is made, it would fail or timeout + let backend = WeixinBackend::new(1, "test_token".into(), "http://127.0.0.1:1".into()); + backend + .set_test_reply_context("wx_user_1", "old_token", true) + .await; + + let outcome = backend.send_message("offline nudge").await.unwrap(); + assert_eq!(outcome, SendOutcome::Buffered); + assert_eq!(backend.pending_count().await, 1); + } + + #[tokio::test] + async fn test_send_buffer_full_reports_dropped() { + let backend = WeixinBackend::new(1, "test_token".into(), "http://127.0.0.1:1".into()); + backend + .set_test_reply_context("wx_user_1", "old_token", true) + .await; + + // Fill buffer to max capacity + for i in 0..MAX_PENDING_MESSAGES { + backend.push_test_pending(&format!("msg {i}"), Duration::from_secs(10)).await; + } + assert_eq!(backend.pending_count().await, MAX_PENDING_MESSAGES); + + let outcome = backend.send_message("overflow message").await.unwrap(); + assert_eq!(outcome, SendOutcome::Dropped); + assert_eq!(backend.pending_count().await, MAX_PENDING_MESSAGES); + } + + #[tokio::test] + async fn test_stale_buffered_messages_are_dropped_by_age() { + let now = Instant::now(); + let old_msg = PendingMessage { + text: "stale 15m".into(), + queued_at: now.checked_sub(Duration::from_secs(15 * 60)).unwrap(), + }; + let fresh_msg = PendingMessage { + text: "fresh 30s".into(), + queued_at: now.checked_sub(Duration::from_secs(30)).unwrap(), + }; + let messages = vec![old_msg, fresh_msg]; + let (stale, fresh): (Vec<_>, Vec<_>) = messages + .into_iter() + .partition(|m| now.duration_since(m.queued_at) >= MAX_PENDING_AGE); + + assert_eq!(stale.len(), 1); + assert_eq!(stale[0].text, "stale 15m"); + assert_eq!(fresh.len(), 1); + assert_eq!(fresh[0].text, "fresh 30s"); + } +} diff --git a/src-tauri/src/chat_channel/command_dispatcher.rs b/src-tauri/src/chat_channel/command_dispatcher.rs index 9da9b2215d..b8bda84a3c 100644 --- a/src-tauri/src/chat_channel/command_dispatcher.rs +++ b/src-tauri/src/chat_channel/command_dispatcher.rs @@ -178,7 +178,7 @@ async fn send_dispatch_message( } }; let (status, error_detail) = match &send_result { - Ok(_) => ("sent", None), + Ok(outcome) => (outcome.status_str(), None), Err(e) => { tracing::error!( "[ChatChannel] failed to send response for {:?} to channel {}: {e}", diff --git a/src-tauri/src/chat_channel/event_subscriber.rs b/src-tauri/src/chat_channel/event_subscriber.rs index ceb5890973..af8917255d 100644 --- a/src-tauri/src/chat_channel/event_subscriber.rs +++ b/src-tauri/src/chat_channel/event_subscriber.rs @@ -385,13 +385,13 @@ async fn process_envelope( // Send let send_result = manager.send_to_channel(ch.id, &msg).await; let (status, error_detail) = match &send_result { - Ok(_) => { - // Only update the debounce timestamp on success, and only for - // debounced event types. - if debounced { + Ok(outcome) => { + // Only advance debounce timestamp on actual delivery, not when + // buffered locally or dropped. + if outcome.is_sent() && debounced { last_push.insert(key, now); } - ("sent", None) + (outcome.status_str(), None) } Err(e) => ("failed", Some(e.to_string())), }; @@ -465,7 +465,7 @@ mod permission_push_tests { use crate::chat_channel::session_bridge::ActiveSession; use crate::chat_channel::traits::ChatChannelBackend; use crate::chat_channel::types::{ - ChannelConnectionStatus, ChannelType, IncomingCommand, SentMessageId, + ChannelConnectionStatus, ChannelType, IncomingCommand, SendOutcome, SentMessageId, }; use crate::db::test_helpers; use crate::models::agent::AgentType; @@ -497,16 +497,111 @@ mod permission_push_tests { async fn status(&self) -> ChannelConnectionStatus { ChannelConnectionStatus::Connected } - async fn send_message(&self, text: &str) -> Result { + async fn send_message(&self, text: &str) -> Result { self.rec.msgs.lock().await.push(text.to_string()); - Ok(SentMessageId("1".into())) + Ok(SendOutcome::Sent(SentMessageId("1".into()))) } async fn send_rich_message( &self, message: &RichMessage, - ) -> Result { + ) -> Result { self.rec.msgs.lock().await.push(message.to_plain_text()); - Ok(SentMessageId("1".into())) + Ok(SendOutcome::Sent(SentMessageId("1".into()))) + } + async fn test_connection(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + } + + /// Backend that queues a message instead of delivering it, the way the + /// Weixin iLink backend does while its `context_token` is stale. + struct BufferingBackend; + + #[async_trait] + impl ChatChannelBackend for BufferingBackend { + fn channel_type(&self) -> ChannelType { + ChannelType::Telegram + } + async fn start(&self, _tx: mpsc::Sender) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn stop(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn status(&self) -> ChannelConnectionStatus { + ChannelConnectionStatus::Connected + } + async fn send_message(&self, _text: &str) -> Result { + Ok(SendOutcome::Buffered) + } + async fn send_rich_message( + &self, + _message: &RichMessage, + ) -> Result { + Ok(SendOutcome::Buffered) + } + async fn test_connection(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + } + + /// Backend that drops messages when pending queue is full. + struct DroppingBackend; + + #[async_trait] + impl ChatChannelBackend for DroppingBackend { + fn channel_type(&self) -> ChannelType { + ChannelType::Telegram + } + async fn start(&self, _tx: mpsc::Sender) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn stop(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn status(&self) -> ChannelConnectionStatus { + ChannelConnectionStatus::Connected + } + async fn send_message(&self, _text: &str) -> Result { + Ok(SendOutcome::Dropped) + } + async fn send_rich_message( + &self, + _message: &RichMessage, + ) -> Result { + Ok(SendOutcome::Dropped) + } + async fn test_connection(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + } + + /// Backend that simulates a provider returning an empty message id on success + /// (e.g. Telegram or Lark). + struct EmptyIdBackend; + + #[async_trait] + impl ChatChannelBackend for EmptyIdBackend { + fn channel_type(&self) -> ChannelType { + ChannelType::Telegram + } + async fn start(&self, _tx: mpsc::Sender) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn stop(&self) -> Result<(), ChatChannelError> { + Ok(()) + } + async fn status(&self) -> ChannelConnectionStatus { + ChannelConnectionStatus::Connected + } + async fn send_message(&self, _text: &str) -> Result { + Ok(SendOutcome::Sent(SentMessageId(String::new()))) + } + async fn send_rich_message( + &self, + _message: &RichMessage, + ) -> Result { + Ok(SendOutcome::Sent(SentMessageId(String::new()))) } async fn test_connection(&self) -> Result<(), ChatChannelError> { Ok(()) @@ -700,6 +795,163 @@ mod permission_push_tests { assert!(msgs[0].contains("Bash: npm test"), "got {:?}", msgs[0]); } + /// A push the backend could only queue must be recorded as `buffered` + /// rather than `sent`, so a stalled channel is visible in the message log + /// instead of looking healthy. + #[tokio::test] + async fn buffered_push_is_logged_as_buffered() { + let db = test_helpers::fresh_in_memory_db().await; + // `chat_channel_message_log` carries a foreign key to `chat_channel`, + // so the channel has to exist in the database as well as in the + // in-memory manager. + let channel_id = crate::db::service::chat_channel_service::create( + &db.conn, + "Stalled test".to_string(), + "telegram".to_string(), + "{}".to_string(), + true, + false, + None, + ) + .await + .expect("seed chat channel") + .id; + + let chat = ChatChannelManager::new(); + chat.add_channel( + channel_id, + "stalled".into(), + ChannelType::Telegram, + Box::new(BufferingBackend), + ) + .await + .unwrap(); + let bridge = Arc::new(Mutex::new(SessionBridge::new())); + let config = config_all_on(channel_id); + let mut last_push = HashMap::new(); + + process_envelope( + &permission_envelope("desktop-conn"), + &bridge, + &chat, + &db.conn, + &config, + &mut last_push, + &test_client(), + ) + .await; + + let logs = crate::db::service::chat_channel_message_log_service::list_by_channel( + &db.conn, channel_id, 10, 0, + ) + .await + .expect("list message logs"); + assert_eq!(logs.len(), 1, "expected one log row, got {logs:?}"); + assert_eq!(logs[0].status, "buffered"); + assert!(logs[0].error_detail.is_none()); + } + + #[tokio::test] + async fn dropped_push_is_logged_as_dropped() { + let db = test_helpers::fresh_in_memory_db().await; + let channel_id = crate::db::service::chat_channel_service::create( + &db.conn, + "Dropped test".to_string(), + "telegram".to_string(), + "{}".to_string(), + true, + false, + None, + ) + .await + .expect("seed chat channel") + .id; + + let chat = ChatChannelManager::new(); + chat.add_channel( + channel_id, + "dropped".into(), + ChannelType::Telegram, + Box::new(DroppingBackend), + ) + .await + .unwrap(); + let bridge = Arc::new(Mutex::new(SessionBridge::new())); + let config = config_all_on(channel_id); + let mut last_push = HashMap::new(); + + process_envelope( + &permission_envelope("desktop-conn"), + &bridge, + &chat, + &db.conn, + &config, + &mut last_push, + &test_client(), + ) + .await; + + let logs = crate::db::service::chat_channel_message_log_service::list_by_channel( + &db.conn, channel_id, 10, 0, + ) + .await + .expect("list message logs"); + assert_eq!(logs.len(), 1, "expected one log row, got {logs:?}"); + assert_eq!(logs[0].status, "dropped"); + assert!(logs[0].error_detail.is_none()); + } + + #[tokio::test] + async fn empty_provider_id_is_logged_as_sent_and_advances_debounce() { + let db = test_helpers::fresh_in_memory_db().await; + let channel_id = crate::db::service::chat_channel_service::create( + &db.conn, + "Empty id test".to_string(), + "telegram".to_string(), + "{}".to_string(), + true, + false, + None, + ) + .await + .expect("seed chat channel") + .id; + + let chat = ChatChannelManager::new(); + chat.add_channel( + channel_id, + "empty_id".into(), + ChannelType::Telegram, + Box::new(EmptyIdBackend), + ) + .await + .unwrap(); + let bridge = Arc::new(Mutex::new(SessionBridge::new())); + let config = config_all_on(channel_id); + let mut last_push = HashMap::new(); + + process_envelope( + &turn_complete_envelope("desktop-conn"), + &bridge, + &chat, + &db.conn, + &config, + &mut last_push, + &test_client(), + ) + .await; + + let logs = crate::db::service::chat_channel_message_log_service::list_by_channel( + &db.conn, channel_id, 10, 0, + ) + .await + .expect("list message logs"); + assert_eq!(logs.len(), 1, "expected one log row, got {logs:?}"); + assert_eq!(logs[0].status, "sent", "empty id from provider must not be confused with buffered"); + assert!(logs[0].error_detail.is_none()); + assert!(!last_push.is_empty(), "debounce map must be updated on real delivery"); + } + /// A permission request from a chat-channel-bridged connection is suppressed /// here — the session relay owns the interactive flow for that channel. #[tokio::test] diff --git a/src-tauri/src/chat_channel/manager.rs b/src-tauri/src/chat_channel/manager.rs index 6c49b49af5..926a04fc51 100644 --- a/src-tauri/src/chat_channel/manager.rs +++ b/src-tauri/src/chat_channel/manager.rs @@ -133,7 +133,7 @@ impl ChatChannelManager { &self, channel_id: i32, message: &RichMessage, - ) -> Result { + ) -> Result { let backend = { let channels = self.inner.channels.lock().await; channels @@ -149,7 +149,7 @@ impl ChatChannelManager { &self, target: &ChannelMessageTarget, message: &RichMessage, - ) -> Result { + ) -> Result { let backend = { let channels = self.inner.channels.lock().await; channels @@ -165,7 +165,7 @@ impl ChatChannelManager { &self, target: &ChannelMessageTarget, message: &InteractiveMessage, - ) -> Result { + ) -> Result { let backend = { let channels = self.inner.channels.lock().await; channels diff --git a/src-tauri/src/chat_channel/scheduler.rs b/src-tauri/src/chat_channel/scheduler.rs index 779439d929..b42ab365d8 100644 --- a/src-tauri/src/chat_channel/scheduler.rs +++ b/src-tauri/src/chat_channel/scheduler.rs @@ -81,7 +81,7 @@ pub fn spawn_daily_report_scheduler( let send_result = manager.send_to_channel(ch.id, &message).await; let (status, error_detail) = match &send_result { - Ok(_) => ("sent", None), + Ok(outcome) => (outcome.status_str(), None), Err(e) => ("failed", Some(e.to_string())), }; diff --git a/src-tauri/src/chat_channel/session_event_subscriber.rs b/src-tauri/src/chat_channel/session_event_subscriber.rs index 6fed9ec465..5f6609a202 100644 --- a/src-tauri/src/chat_channel/session_event_subscriber.rs +++ b/src-tauri/src/chat_channel/session_event_subscriber.rs @@ -1256,7 +1256,7 @@ mod async_relay_dedup_tests { use crate::chat_channel::session_bridge::{ActiveSession, SessionBridge}; use crate::chat_channel::traits::ChatChannelBackend; use crate::chat_channel::types::{ - ChannelConnectionStatus, ChannelType, IncomingCommand, RichMessage, SentMessageId, + ChannelConnectionStatus, ChannelType, IncomingCommand, RichMessage, SendOutcome, SentMessageId, }; use crate::db::test_helpers; use crate::models::agent::AgentType; @@ -1294,16 +1294,16 @@ mod async_relay_dedup_tests { async fn status(&self) -> ChannelConnectionStatus { ChannelConnectionStatus::Connected } - async fn send_message(&self, text: &str) -> Result { + async fn send_message(&self, text: &str) -> Result { self.rec.msgs.lock().await.push(text.to_string()); - Ok(SentMessageId("1".into())) + Ok(SendOutcome::Sent(SentMessageId("1".into()))) } async fn send_rich_message( &self, message: &RichMessage, - ) -> Result { + ) -> Result { self.rec.msgs.lock().await.push(message.body.clone()); - Ok(SentMessageId("1".into())) + Ok(SendOutcome::Sent(SentMessageId("1".into()))) } async fn test_connection(&self) -> Result<(), ChatChannelError> { Ok(()) diff --git a/src-tauri/src/chat_channel/traits.rs b/src-tauri/src/chat_channel/traits.rs index 670b25efa1..fb14c05b73 100644 --- a/src-tauri/src/chat_channel/traits.rs +++ b/src-tauri/src/chat_channel/traits.rs @@ -22,13 +22,13 @@ pub trait ChatChannelBackend: Send + Sync + 'static { async fn status(&self) -> ChannelConnectionStatus; /// Send a plain text message. - async fn send_message(&self, text: &str) -> Result; + async fn send_message(&self, text: &str) -> Result; /// Send a rich/structured message (Telegram Markdown / Lark Card). async fn send_rich_message( &self, message: &RichMessage, - ) -> Result; + ) -> Result; /// Send a rich message to a provider-specific thread/topic target. /// Backends without thread semantics keep the existing channel-level behavior. @@ -36,7 +36,7 @@ pub trait ChatChannelBackend: Send + Sync + 'static { &self, message: &RichMessage, _target: &ChannelMessageTarget, - ) -> Result { + ) -> Result { self.send_rich_message(message).await } @@ -66,7 +66,7 @@ pub trait ChatChannelBackend: Send + Sync + 'static { async fn send_interactive_message( &self, message: &InteractiveMessage, - ) -> Result { + ) -> Result { self.send_rich_message(&message.to_rich_fallback()).await } @@ -75,7 +75,7 @@ pub trait ChatChannelBackend: Send + Sync + 'static { &self, message: &InteractiveMessage, _target: &ChannelMessageTarget, - ) -> Result { + ) -> Result { self.send_interactive_message(message).await } diff --git a/src-tauri/src/chat_channel/types.rs b/src-tauri/src/chat_channel/types.rs index 6c1d342b39..c18e01ad70 100644 --- a/src-tauri/src/chat_channel/types.rs +++ b/src-tauri/src/chat_channel/types.rs @@ -47,9 +47,56 @@ pub enum ChannelConnectionStatus { Error, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct SentMessageId(pub String); +/// Outcome of attempting to send a message via a chat channel backend. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SendOutcome { + /// Delivered to the upstream provider (with provider message id or client id). + Sent(SentMessageId), + /// Queued locally for delivery once context refreshes (e.g. Weixin iLink). + Buffered, + /// Dropped because the local pending buffer was full or message aged out. + Dropped, +} + +impl SendOutcome { + pub fn is_sent(&self) -> bool { + matches!(self, SendOutcome::Sent(_)) + } + + pub fn is_buffered(&self) -> bool { + matches!(self, SendOutcome::Buffered) + } + + pub fn is_dropped(&self) -> bool { + matches!(self, SendOutcome::Dropped) + } + + pub fn status_str(&self) -> &'static str { + match self { + SendOutcome::Sent(_) => "sent", + SendOutcome::Buffered => "buffered", + SendOutcome::Dropped => "dropped", + } + } + + pub fn sent_message_id(&self) -> Option<&SentMessageId> { + match self { + SendOutcome::Sent(id) => Some(id), + _ => None, + } + } +} + +impl From for SendOutcome { + fn from(id: SentMessageId) -> Self { + SendOutcome::Sent(id) + } +} + pub struct IncomingCommand { pub channel_id: i32, pub sender_id: String, diff --git a/src-tauri/src/commands/conversations.rs b/src-tauri/src/commands/conversations.rs index bf02d6ad59..e6f71e69b1 100644 --- a/src-tauri/src/commands/conversations.rs +++ b/src-tauri/src/commands/conversations.rs @@ -5278,20 +5278,24 @@ mod tests { &self, _text: &str, ) -> Result< - crate::chat_channel::types::SentMessageId, + crate::chat_channel::types::SendOutcome, crate::chat_channel::error::ChatChannelError, > { - Ok(crate::chat_channel::types::SentMessageId("sent".into())) + Ok(crate::chat_channel::types::SendOutcome::Sent( + crate::chat_channel::types::SentMessageId("sent".into()), + )) } async fn send_rich_message( &self, _message: &crate::chat_channel::types::RichMessage, ) -> Result< - crate::chat_channel::types::SentMessageId, + crate::chat_channel::types::SendOutcome, crate::chat_channel::error::ChatChannelError, > { - Ok(crate::chat_channel::types::SentMessageId("sent".into())) + Ok(crate::chat_channel::types::SendOutcome::Sent( + crate::chat_channel::types::SentMessageId("sent".into()), + )) } async fn edit_thread_title(