From 0d7f7b817fa0cde002e1b0fe2190c09f866909c8 Mon Sep 17 00:00:00 2001 From: Coffee Date: Fri, 27 Mar 2026 19:04:15 +0800 Subject: [PATCH] Add configurable WeChat merge windows and clean debug logs --- channels-src/wechat/src/api.rs | 19 +- channels-src/wechat/src/lib.rs | 239 +++++++++++++++---- channels-src/wechat/src/state.rs | 1 + channels-src/wechat/src/types.rs | 9 +- channels-src/wechat/wechat.capabilities.json | 3 +- src/channels/wasm/attachment_hydration.rs | 18 +- src/llm/transcription/mod.rs | 6 +- 7 files changed, 224 insertions(+), 71 deletions(-) diff --git a/channels-src/wechat/src/api.rs b/channels-src/wechat/src/api.rs index 43df2fc5..000bc56a 100644 --- a/channels-src/wechat/src/api.rs +++ b/channels-src/wechat/src/api.rs @@ -51,6 +51,14 @@ fn summarize_body_preview(bytes: &[u8], limit: usize) -> String { pub fn get_updates( config: &WechatConfig, get_updates_buf: &str, +) -> Result { + get_updates_with_timeout(config, get_updates_buf, config.long_poll_timeout_ms) +} + +pub fn get_updates_with_timeout( + config: &WechatConfig, + get_updates_buf: &str, + timeout_ms: u32, ) -> Result { let body = serde_json::to_vec(&GetUpdatesRequest { get_updates_buf: get_updates_buf.to_string(), @@ -70,14 +78,9 @@ pub fn get_updates( config.long_poll_timeout_ms ), ); - let response = channel_host::http_request( - "POST", - &url, - &headers, - Some(&body), - Some(config.long_poll_timeout_ms), - ) - .map_err(|e| format!("getUpdates request failed: {e}"))?; + let response = + channel_host::http_request("POST", &url, &headers, Some(&body), Some(timeout_ms)) + .map_err(|e| format!("getUpdates request failed: {e}"))?; channel_host::log( channel_host::LogLevel::Info, diff --git a/channels-src/wechat/src/lib.rs b/channels-src/wechat/src/lib.rs index d95cfa2e..fbbbb552 100644 --- a/channels-src/wechat/src/lib.rs +++ b/channels-src/wechat/src/lib.rs @@ -1,5 +1,3 @@ -use std::collections::HashSet; - wit_bindgen::generate!({ world: "sandboxed-channel", path: "../../wit/channel.wit", @@ -39,17 +37,6 @@ enum WechatStatusAction { struct WechatChannel; -fn log_channel(level: channel_host::LogLevel, message: &str) { - #[cfg(not(test))] - channel_host::log(level, message); - - #[cfg(test)] - { - let _ = level; - let _ = message; - } -} - impl Guest for WechatChannel { fn on_start(config_json: String) -> Result { let config = serde_json::from_str::(&config_json) @@ -87,6 +74,7 @@ impl Guest for WechatChannel { let config = load_config(); let cursor = load_get_updates_buf(); + let mut current_cursor = cursor.clone(); let mut context_tokens = load_context_tokens(); let mut pending_inbound = match load_pending_inbound_bundles() { Ok(bundles) => bundles, @@ -98,10 +86,14 @@ impl Guest for WechatChannel { return; } }; - let carried_pending_keys: HashSet = pending_inbound.keys().cloned().collect(); let mut pending_inbound_changed = false; - match api::get_updates(&config, &cursor) { + for bundle in take_due_pending_bundles(&mut pending_inbound, channel_host::now_millis()) { + pending_inbound_changed = true; + emit_buffered_bundle(bundle); + } + + match api::get_updates(&config, ¤t_cursor) { Ok(response) => { if response.errcode == Some(-14) { channel_host::log( @@ -126,7 +118,8 @@ impl Guest for WechatChannel { } if let Some(next_cursor) = response.get_updates_buf.as_deref() { - if next_cursor != cursor { + if next_cursor != current_cursor { + current_cursor = next_cursor.to_string(); if let Err(error) = persist_get_updates_buf(next_cursor) { channel_host::log( channel_host::LogLevel::Warn, @@ -153,6 +146,8 @@ impl Guest for WechatChannel { &mut pending_inbound, bundle, &mut pending_inbound_changed, + channel_host::now_millis(), + u64::from(config.inbound_merge_window_ms), ); for emitted_bundle in emitted { emit_buffered_bundle(emitted_bundle); @@ -168,11 +163,20 @@ impl Guest for WechatChannel { } } - for key in carried_pending_keys { - if let Some(bundle) = pending_inbound.remove(&key) { - pending_inbound_changed = true; - emit_buffered_bundle(bundle); - } + collect_follow_up_bundles( + &config, + &mut current_cursor, + &mut context_tokens, + &mut context_tokens_changed, + &mut pending_inbound, + &mut pending_inbound_changed, + ); + + for bundle in + take_due_pending_bundles(&mut pending_inbound, channel_host::now_millis()) + { + pending_inbound_changed = true; + emit_buffered_bundle(bundle); } if context_tokens_changed { @@ -299,6 +303,7 @@ fn incoming_bundle_from_message( session_id: message.session_id, context_token: message.context_token, message_id: message.message_id, + flush_at_ms: 0, text, attachments, })) @@ -306,8 +311,10 @@ fn incoming_bundle_from_message( fn process_incoming_bundle( pending_inbound: &mut std::collections::HashMap, - bundle: PendingInboundBundle, + mut bundle: PendingInboundBundle, pending_inbound_changed: &mut bool, + now_ms: u64, + inbound_merge_window_ms: u64, ) -> Vec { let key = bundle.from_user_id.clone(); let bundle_has_text = !bundle.text.trim().is_empty(); @@ -321,39 +328,20 @@ fn process_incoming_bundle( pending.text = merge_text(&pending.text, &bundle.text); pending.attachments.extend(bundle.attachments); merge_bundle_metadata(&mut pending, &incoming_metadata); - log_channel( - channel_host::LogLevel::Info, - &format!( - "Merged buffered WeChat attachment message with follow-up text for {}", - pending.from_user_id - ), - ); return vec![pending]; } let incoming_metadata = bundle.clone(); pending.attachments.extend(bundle.attachments); merge_bundle_metadata(&mut pending, &incoming_metadata); + pending.flush_at_ms = next_flush_deadline(now_ms, inbound_merge_window_ms); pending_inbound.insert(key, pending); - log_channel( - channel_host::LogLevel::Info, - &format!( - "Buffered additional WeChat attachment for {} while waiting for follow-up text", - bundle.from_user_id - ), - ); return Vec::new(); } if bundle_has_attachments && !bundle_has_text { *pending_inbound_changed = true; - log_channel( - channel_host::LogLevel::Info, - &format!( - "Buffered WeChat attachment-only message for {} and will wait one poll cycle for follow-up text", - bundle.from_user_id - ), - ); + bundle.flush_at_ms = next_flush_deadline(now_ms, inbound_merge_window_ms); pending_inbound.insert(key, bundle); Vec::new() } else { @@ -361,6 +349,133 @@ fn process_incoming_bundle( } } +fn collect_follow_up_bundles( + config: &WechatConfig, + current_cursor: &mut String, + context_tokens: &mut std::collections::HashMap, + context_tokens_changed: &mut bool, + pending_inbound: &mut std::collections::HashMap, + pending_inbound_changed: &mut bool, +) { + while !pending_inbound.is_empty() { + let now_ms = channel_host::now_millis(); + let Some(timeout_ms) = next_follow_up_timeout_ms(pending_inbound, now_ms) else { + break; + }; + if timeout_ms == 0 { + break; + } + + let timeout_ms_u32 = timeout_ms.min(u64::from(u32::MAX)) as u32; + let response = match api::get_updates_with_timeout(config, current_cursor, timeout_ms_u32) { + Ok(response) => response, + Err(_) => break, + }; + + if response.errcode == Some(-14) { + channel_host::log( + channel_host::LogLevel::Error, + "WeChat getUpdates returned errcode=-14 during follow-up merge window; reconnect the channel", + ); + break; + } + + if response.ret.unwrap_or(0) != 0 { + let errmsg = response + .errmsg + .as_deref() + .unwrap_or("unknown WeChat polling error"); + channel_host::log( + channel_host::LogLevel::Warn, + &format!( + "WeChat getUpdates returned ret={} errmsg={errmsg} during follow-up merge window", + response.ret.unwrap_or(-1) + ), + ); + } + + if let Some(next_cursor) = response.get_updates_buf.as_deref() { + if next_cursor != current_cursor { + *current_cursor = next_cursor.to_string(); + if let Err(error) = persist_get_updates_buf(next_cursor) { + channel_host::log( + channel_host::LogLevel::Warn, + &format!("Failed to persist WeChat polling cursor: {error}"), + ); + } + } + } + + let mut saw_relevant_message = false; + for message in response.msgs { + if let Some(from_user_id) = message.from_user_id.as_deref() { + if let Some(context_token) = message.context_token.as_deref() { + let changed = context_tokens + .insert(from_user_id.to_string(), context_token.to_string()) + .as_deref() + != Some(context_token); + *context_tokens_changed |= changed; + } + } + match incoming_bundle_from_message(config, message) { + Ok(Some(bundle)) => { + let emitted = process_incoming_bundle( + pending_inbound, + bundle, + pending_inbound_changed, + channel_host::now_millis(), + u64::from(config.inbound_merge_window_ms), + ); + for emitted_bundle in emitted { + saw_relevant_message = true; + emit_buffered_bundle(emitted_bundle); + } + } + Ok(None) => {} + Err(error) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to map WeChat inbound message: {error}"), + ); + } + } + } + + if !saw_relevant_message && pending_inbound.is_empty() { + break; + } + } +} + +fn next_flush_deadline(now_ms: u64, inbound_merge_window_ms: u64) -> u64 { + now_ms.saturating_add(inbound_merge_window_ms) +} + +fn next_follow_up_timeout_ms( + pending_inbound: &std::collections::HashMap, + now_ms: u64, +) -> Option { + pending_inbound + .values() + .map(|bundle| bundle.flush_at_ms.saturating_sub(now_ms)) + .min() +} + +fn take_due_pending_bundles( + pending_inbound: &mut std::collections::HashMap, + now_ms: u64, +) -> Vec { + let due_keys = pending_inbound + .iter() + .filter_map(|(key, bundle)| (bundle.flush_at_ms <= now_ms).then_some(key.clone())) + .collect::>(); + + due_keys + .into_iter() + .filter_map(|key| pending_inbound.remove(&key)) + .collect() +} + fn emit_buffered_bundle(bundle: PendingInboundBundle) { let metadata = json!({ "from_user_id": bundle.from_user_id, @@ -402,7 +517,7 @@ fn merge_text(existing: &str, incoming: &str) -> String { (true, true) => String::new(), (true, false) => incoming.to_string(), (false, true) => existing.to_string(), - (false, false) => format!("{existing}\n{incoming}"), + (false, false) => format!("{existing}\n\n{incoming}"), } } @@ -616,7 +731,8 @@ mod tests { use super::{ classify_status_update, extract_text, merge_text, process_incoming_bundle, - PendingInboundBundle, StoredInboundAttachment, WechatStatusAction, + take_due_pending_bundles, PendingInboundBundle, StoredInboundAttachment, + WechatStatusAction, }; use crate::exports::near::agent::channel::{StatusType, StatusUpdate}; use crate::types::{MessageItem, VoiceItem, WechatMessage, MESSAGE_ITEM_VOICE}; @@ -628,6 +744,7 @@ mod tests { session_id: Some("session-1".to_string()), context_token: Some("ctx-1".to_string()), message_id: Some(1), + flush_at_ms: 0, text: text.to_string(), attachments: (0..image_count) .map(|index| StoredInboundAttachment { @@ -725,7 +842,7 @@ mod tests { #[test] fn test_merge_text_joins_non_empty_segments() { assert_eq!(merge_text("", "hello"), "hello"); - assert_eq!(merge_text("look", "what is this"), "look\nwhat is this"); + assert_eq!(merge_text("look", "what is this"), "look\n\nwhat is this"); assert_eq!(merge_text("look", ""), "look"); } @@ -760,16 +877,25 @@ mod tests { let mut pending = HashMap::new(); let mut changed = false; - let emitted = process_incoming_bundle(&mut pending, make_bundle("u1", "", 1), &mut changed); + let emitted = process_incoming_bundle( + &mut pending, + make_bundle("u1", "", 1), + &mut changed, + 100, + 5_000, + ); assert!(emitted.is_empty()); assert!(changed); assert_eq!(pending.len(), 1); + assert_eq!(pending["u1"].flush_at_ms, 5100); changed = false; let emitted = process_incoming_bundle( &mut pending, make_bundle("u1", "What is in this image?", 0), &mut changed, + 200, + 5_000, ); assert!(changed); assert!(pending.is_empty()); @@ -787,6 +913,8 @@ mod tests { &mut pending, make_bundle("u1", "Look at this image", 1), &mut changed, + 100, + 5_000, ); assert!(!changed); assert!(pending.is_empty()); @@ -794,4 +922,21 @@ mod tests { assert_eq!(emitted[0].text, "Look at this image"); assert_eq!(emitted[0].attachments.len(), 1); } + + #[test] + fn test_take_due_pending_bundles_emits_only_expired_entries() { + let mut pending = HashMap::new(); + let mut expired = make_bundle("u1", "", 1); + expired.flush_at_ms = 100; + let mut fresh = make_bundle("u2", "", 1); + fresh.flush_at_ms = 300; + pending.insert(expired.from_user_id.clone(), expired); + pending.insert(fresh.from_user_id.clone(), fresh); + + let due = take_due_pending_bundles(&mut pending, 200); + assert_eq!(due.len(), 1); + assert_eq!(due[0].from_user_id, "u1"); + assert_eq!(pending.len(), 1); + assert!(pending.contains_key("u2")); + } } diff --git a/channels-src/wechat/src/state.rs b/channels-src/wechat/src/state.rs index 503e2ed2..b2d172fe 100644 --- a/channels-src/wechat/src/state.rs +++ b/channels-src/wechat/src/state.rs @@ -64,6 +64,7 @@ pub struct PendingInboundBundle { pub session_id: Option, pub context_token: Option, pub message_id: Option, + pub flush_at_ms: u64, pub text: String, pub attachments: Vec, } diff --git a/channels-src/wechat/src/types.rs b/channels-src/wechat/src/types.rs index 675425c1..ae73e26d 100644 --- a/channels-src/wechat/src/types.rs +++ b/channels-src/wechat/src/types.rs @@ -12,6 +12,8 @@ pub struct WechatConfig { pub poll_interval_ms: u32, #[serde(default = "default_long_poll_timeout_ms")] pub long_poll_timeout_ms: u32, + #[serde(default = "default_inbound_merge_window_ms")] + pub inbound_merge_window_ms: u32, } fn default_base_url() -> String { @@ -31,7 +33,11 @@ fn default_poll_interval_ms() -> u32 { } fn default_long_poll_timeout_ms() -> u32 { - 15_000 + 35_000 +} + +fn default_inbound_merge_window_ms() -> u32 { + 5_000 } impl Default for WechatConfig { @@ -42,6 +48,7 @@ impl Default for WechatConfig { bot_type: default_bot_type(), poll_interval_ms: default_poll_interval_ms(), long_poll_timeout_ms: default_long_poll_timeout_ms(), + inbound_merge_window_ms: default_inbound_merge_window_ms(), } } } diff --git a/channels-src/wechat/wechat.capabilities.json b/channels-src/wechat/wechat.capabilities.json index ba148c83..05f3329b 100644 --- a/channels-src/wechat/wechat.capabilities.json +++ b/channels-src/wechat/wechat.capabilities.json @@ -45,6 +45,7 @@ "cdn_base_url": "https://novac2c.cdn.weixin.qq.com/c2c", "bot_type": "3", "poll_interval_ms": 30000, - "long_poll_timeout_ms": 15000 + "long_poll_timeout_ms": 35000, + "inbound_merge_window_ms": 5000 } } diff --git a/src/channels/wasm/attachment_hydration.rs b/src/channels/wasm/attachment_hydration.rs index b722ffc4..03a20a4c 100644 --- a/src/channels/wasm/attachment_hydration.rs +++ b/src/channels/wasm/attachment_hydration.rs @@ -47,15 +47,15 @@ pub(crate) async fn hydrate_attachment_for_channel( attachment.data = plaintext; if attachment.mime_type.starts_with("image/") { attachment.mime_type = detect_image_mime(&attachment.data).to_string(); - } else if is_wechat_silk_attachment(attachment) - && let Err(error) = maybe_transcode_wechat_silk_attachment(attachment) - { - tracing::warn!( - channel = %channel_name, - attachment_id = %attachment.id, - error = %error, - "Failed to transcode WeChat SILK attachment; preserving raw SILK" - ); + } else if is_wechat_silk_attachment(attachment) { + if let Err(error) = maybe_transcode_wechat_silk_attachment(attachment) { + tracing::warn!( + channel = %channel_name, + attachment_id = %attachment.id, + error = %error, + "Failed to transcode WeChat SILK attachment; preserving raw SILK" + ); + } } } Err(error) => { diff --git a/src/llm/transcription/mod.rs b/src/llm/transcription/mod.rs index ab2e43f9..08098232 100644 --- a/src/llm/transcription/mod.rs +++ b/src/llm/transcription/mod.rs @@ -119,6 +119,7 @@ impl TranscriptionMiddleware { Some(f) => f, None => { tracing::warn!( + attachment_id = %attachment.id, mime = %attachment.mime_type, "Skipping audio attachment with unsupported format" ); @@ -128,11 +129,6 @@ impl TranscriptionMiddleware { match self.provider.transcribe(&attachment.data, format).await { Ok(text) => { - tracing::info!( - attachment_id = %attachment.id, - text_len = text.len(), - "Transcribed audio attachment" - ); transcriptions.push((i, text)); } Err(e) => {