From 5705366efbc634748ff450626f235254a5df551c Mon Sep 17 00:00:00 2001 From: Coffee Date: Thu, 26 Mar 2026 18:59:26 +0800 Subject: [PATCH] Add WeChat image messaging and QR login polish --- Cargo.lock | 1 + Cargo.toml | 1 + FEATURE_PARITY.md | 2 +- channels-src/wechat/Cargo.lock | 160 ++++++++ channels-src/wechat/Cargo.toml | 4 + channels-src/wechat/build.sh | 9 +- channels-src/wechat/src/api.rs | 95 ++++- channels-src/wechat/src/auth.rs | 2 +- channels-src/wechat/src/lib.rs | 350 +++++++++++++++-- channels-src/wechat/src/media.rs | 355 ++++++++++++++++++ channels-src/wechat/src/state.rs | 98 ++++- channels-src/wechat/src/types.rs | 52 +++ channels-src/wechat/wechat.capabilities.json | 4 +- .../2026-03-25-wechat-integration-design.md | 4 +- src/agent/dispatcher.rs | 1 - src/agent/thread_ops.rs | 162 +++++--- src/channels/wasm/attachment_hydration.rs | 313 +++++++++++++++ src/channels/wasm/host.rs | 3 + src/channels/wasm/mod.rs | 1 + src/channels/wasm/router.rs | 9 + src/channels/wasm/wrapper.rs | 223 +++++++---- src/extensions/manager.rs | 4 + 22 files changed, 1655 insertions(+), 198 deletions(-) create mode 100644 channels-src/wechat/src/media.rs create mode 100644 src/channels/wasm/attachment_hydration.rs diff --git a/Cargo.lock b/Cargo.lock index a813ef2b..f7ab18c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3392,6 +3392,7 @@ dependencies = [ name = "ironclaw" version = "0.19.0" dependencies = [ + "aes", "aes-gcm", "aho-corasick", "anyhow", diff --git a/Cargo.toml b/Cargo.toml index 99992a40..00882f15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -135,6 +135,7 @@ wasmtime-wasi = "28" # WASI support for component model wasmparser = "0.220" # WASM binary parsing for validation # Cryptography for secrets management +aes = "0.8" aes-gcm = "0.10" hkdf = "0.12" hmac = "0.12" diff --git a/FEATURE_PARITY.md b/FEATURE_PARITY.md index 0e63b4c0..b8619525 100644 --- a/FEATURE_PARITY.md +++ b/FEATURE_PARITY.md @@ -77,7 +77,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O | Linq | ✅ | ❌ | P3 | Real iMessage via API, no Mac required | | Feishu/Lark | ✅ | 🚧 | P3 | WASM channel with Event Subscription v2.0; Bitable/Docx tools planned | | LINE | ✅ | ❌ | P3 | | -| WeChat (iLink bot) | ✅ | ❌ | P2 | Extension-first channel (`channels-src/wechat`), single-account DM flow with QR login and typing; multi-account/media follow-up | +| WeChat (iLink bot) | ✅ | 🚧 | P2 | Extension-first channel (`channels-src/wechat`), single-account DM flow with QR login, typing, and image send/receive; multi-account and broader media parity follow-up | | WebChat | ✅ | ✅ | - | Web gateway chat | | Matrix | ✅ | ❌ | P3 | E2EE support | | Mattermost | ✅ | ❌ | P3 | Emoji reactions, interactive buttons, model picker | diff --git a/channels-src/wechat/Cargo.lock b/channels-src/wechat/Cargo.lock index a8084278..655c86f5 100644 --- a/channels-src/wechat/Cargo.lock +++ b/channels-src/wechat/Cargo.lock @@ -2,6 +2,17 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + [[package]] name = "ahash" version = "0.8.12" @@ -32,18 +43,87 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", +] + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "equivalent" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + [[package]] name = "hashbrown" version = "0.14.5" @@ -83,6 +163,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "itoa" version = "1.0.18" @@ -95,12 +184,28 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67" +[[package]] +name = "libc" +version = "0.2.183" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5b646652bf6661599e1da8901b3b9522896f01e736bad5f723fe7a3a27f899d" + [[package]] name = "log" version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "md-5" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.0" @@ -113,6 +218,15 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -141,6 +255,36 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "rand" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + [[package]] name = "semver" version = "1.0.27" @@ -216,6 +360,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "typenum" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -234,6 +384,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasm-encoder" version = "0.220.1" @@ -277,7 +433,11 @@ dependencies = [ name = "wechat-channel" version = "0.1.0" dependencies = [ + "aes", "base64", + "cipher", + "md-5", + "rand", "serde", "serde_json", "wit-bindgen", diff --git a/channels-src/wechat/Cargo.toml b/channels-src/wechat/Cargo.toml index cf40f496..e4384e2b 100644 --- a/channels-src/wechat/Cargo.toml +++ b/channels-src/wechat/Cargo.toml @@ -13,6 +13,10 @@ wit-bindgen = "0.36" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" base64 = "0.22" +aes = "0.8" +cipher = "0.4" +md-5 = "0.10" +rand = "0.8" [profile.release] opt-level = "s" diff --git a/channels-src/wechat/build.sh b/channels-src/wechat/build.sh index 4013354d..c3188103 100755 --- a/channels-src/wechat/build.sh +++ b/channels-src/wechat/build.sh @@ -10,8 +10,13 @@ cargo build --release --target wasm32-wasip2 WASM_PATH="target/wasm32-wasip2/release/wechat_channel.wasm" if [ -f "$WASM_PATH" ]; then - wasm-tools component new "$WASM_PATH" -o wechat.wasm 2>/dev/null || cp "$WASM_PATH" wechat.wasm - wasm-tools strip wechat.wasm -o wechat.wasm + if command -v wasm-tools >/dev/null 2>&1; then + wasm-tools component new "$WASM_PATH" -o wechat.wasm 2>/dev/null || cp "$WASM_PATH" wechat.wasm + wasm-tools strip wechat.wasm -o wechat.wasm + else + cp "$WASM_PATH" wechat.wasm + echo "wasm-tools not found; copied raw wasm output without component conversion/strip" + fi echo "Built: wechat.wasm ($(du -h wechat.wasm | cut -f1))" echo "" diff --git a/channels-src/wechat/src/api.rs b/channels-src/wechat/src/api.rs index 19bc6fd1..77ca2a66 100644 --- a/channels-src/wechat/src/api.rs +++ b/channels-src/wechat/src/api.rs @@ -3,11 +3,12 @@ use base64::Engine as _; use crate::near::agent::channel_host; use crate::types::{ BaseInfo, GetConfigRequest, GetConfigResponse, GetUpdatesRequest, GetUpdatesResponse, - MessageItem, OutboundWechatMessage, SendMessageRequest, SendTypingRequest, SendTypingResponse, - TextItem, WechatConfig, MESSAGE_ITEM_TEXT, MESSAGE_STATE_FINISH, MESSAGE_TYPE_BOT, + GetUploadUrlRequest, GetUploadUrlResponse, MessageItem, OutboundWechatMessage, + SendMessageRequest, SendTypingRequest, SendTypingResponse, TextItem, WechatConfig, + MESSAGE_ITEM_TEXT, MESSAGE_STATE_FINISH, MESSAGE_TYPE_BOT, }; -fn base_info() -> BaseInfo { +pub fn base_info() -> BaseInfo { BaseInfo { channel_version: env!("CARGO_PKG_VERSION").to_string(), } @@ -37,6 +38,16 @@ fn request_headers(body: &[u8]) -> String { .to_string() } +fn summarize_body_preview(bytes: &[u8], limit: usize) -> String { + let preview = String::from_utf8_lossy(&bytes[..bytes.len().min(limit)]); + let normalized = preview.replace(['\n', '\r'], " "); + if bytes.len() > limit { + format!("{normalized}...") + } else { + normalized + } +} + pub fn get_updates( config: &WechatConfig, get_updates_buf: &str, @@ -51,6 +62,14 @@ pub fn get_updates( "{}ilink/bot/getupdates", ensure_trailing_slash(&config.base_url) ); + channel_host::log( + channel_host::LogLevel::Info, + &format!( + "WeChat getUpdates request: cursor_len={} timeout_ms={}", + get_updates_buf.len(), + config.long_poll_timeout_ms + ), + ); let response = channel_host::http_request( "POST", &url, @@ -60,13 +79,42 @@ pub fn get_updates( ) .map_err(|e| format!("getUpdates request failed: {e}"))?; + channel_host::log( + channel_host::LogLevel::Info, + &format!( + "WeChat getUpdates response: status={} bytes={} has_image_marker={} has_aeskey_marker={} preview={}", + response.status, + response.body.len(), + response + .body + .windows(b"image_item".len()) + .any(|window| window == b"image_item"), + response + .body + .windows(b"aeskey".len()) + .any(|window| window == b"aeskey"), + summarize_body_preview(&response.body, 160) + ), + ); + if response.status != 200 { let body = String::from_utf8_lossy(&response.body); return Err(format!("getUpdates returned {}: {}", response.status, body)); } - serde_json::from_slice(&response.body) - .map_err(|e| format!("Failed to parse getUpdates response: {e}")) + let parsed: GetUpdatesResponse = serde_json::from_slice(&response.body) + .map_err(|e| format!("Failed to parse getUpdates response: {e}"))?; + channel_host::log( + channel_host::LogLevel::Info, + &format!( + "WeChat getUpdates parsed: ret={:?} errcode={:?} msg_count={} next_cursor_len={}", + parsed.ret, + parsed.errcode, + parsed.msgs.len(), + parsed.get_updates_buf.as_deref().unwrap_or_default().len() + ), + ); + Ok(parsed) } pub fn send_text_message( @@ -87,13 +135,21 @@ pub fn send_text_message( text_item: Some(TextItem { text: text.to_string(), }), + image_item: None, }], context_token: context_token.map(str::to_string), }, base_info: base_info(), }; - let body = serde_json::to_vec(&message) + send_message_request(config, &message) +} + +pub fn send_message_request( + config: &WechatConfig, + message: &SendMessageRequest, +) -> Result<(), String> { + let body = serde_json::to_vec(message) .map_err(|e| format!("Failed to encode sendMessage request: {e}"))?; let headers = request_headers(&body); let url = format!( @@ -115,6 +171,33 @@ pub fn send_text_message( Ok(()) } +pub fn get_upload_url( + config: &WechatConfig, + request: &GetUploadUrlRequest, +) -> Result { + let body = serde_json::to_vec(request) + .map_err(|e| format!("Failed to encode getUploadUrl request: {e}"))?; + let headers = request_headers(&body); + let url = format!( + "{}ilink/bot/getuploadurl", + ensure_trailing_slash(&config.base_url) + ); + + let response = channel_host::http_request("POST", &url, &headers, Some(&body), Some(15_000)) + .map_err(|e| format!("getUploadUrl request failed: {e}"))?; + + if response.status != 200 { + let body = String::from_utf8_lossy(&response.body); + return Err(format!( + "getUploadUrl returned {}: {}", + response.status, body + )); + } + + serde_json::from_slice(&response.body) + .map_err(|e| format!("Failed to parse getUploadUrl response: {e}")) +} + pub fn get_config( config: &WechatConfig, ilink_user_id: &str, diff --git a/channels-src/wechat/src/auth.rs b/channels-src/wechat/src/auth.rs index 2535f4b3..c1edf5ba 100644 --- a/channels-src/wechat/src/auth.rs +++ b/channels-src/wechat/src/auth.rs @@ -3,4 +3,4 @@ pub const CONFIG_PATH: &str = "config.json"; pub const GET_UPDATES_BUF_PATH: &str = "state/get_updates_buf.json"; pub const CONTEXT_TOKENS_PATH: &str = "state/context_tokens.json"; pub const TYPING_TICKETS_PATH: &str = "state/typing_tickets.json"; -pub const SESSION_EXPIRED_PATH: &str = "state/session_expired"; +pub const PENDING_INBOUND_PATH: &str = "state/pending_inbound.json"; diff --git a/channels-src/wechat/src/lib.rs b/channels-src/wechat/src/lib.rs index 780b8d23..a3a238f0 100644 --- a/channels-src/wechat/src/lib.rs +++ b/channels-src/wechat/src/lib.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + wit_bindgen::generate!({ world: "sandboxed-channel", path: "../../wit/channel.wit", @@ -5,6 +7,7 @@ wit_bindgen::generate!({ mod api; mod auth; +mod media; mod state; mod types; @@ -16,9 +19,10 @@ use serde_json::json; use crate::auth::TOKEN_SECRET_NAME; use crate::state::{ - clear_session_expired, load_config, load_context_tokens, load_get_updates_buf, - load_typing_tickets, mark_session_expired, persist_config, persist_context_tokens, - persist_get_updates_buf, persist_typing_tickets, session_expired, TypingTicketEntry, + load_config, load_context_tokens, load_get_updates_buf, load_pending_inbound_bundles, + load_typing_tickets, persist_config, persist_context_tokens, persist_get_updates_buf, + persist_pending_inbound_bundles, persist_typing_tickets, PendingInboundBundle, + StoredInboundAttachment, TypingTicketEntry, }; use crate::types::{ OutboundMetadata, WechatConfig, WechatMessage, MESSAGE_ITEM_TEXT, MESSAGE_TYPE_USER, @@ -35,12 +39,22 @@ 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) .map_err(|e| format!("Failed to parse WeChat config: {e}"))?; persist_config(&config)?; - clear_session_expired(); Ok(ChannelConfig { display_name: "WeChat".to_string(), @@ -63,14 +77,6 @@ impl Guest for WechatChannel { } fn on_poll() { - if session_expired() { - channel_host::log( - channel_host::LogLevel::Warn, - "WeChat session is marked expired; reconnect the channel to resume polling", - ); - return; - } - if !channel_host::secret_exists(TOKEN_SECRET_NAME) { channel_host::log( channel_host::LogLevel::Warn, @@ -82,14 +88,25 @@ impl Guest for WechatChannel { let config = load_config(); let cursor = load_get_updates_buf(); let mut context_tokens = load_context_tokens(); + let mut pending_inbound = match load_pending_inbound_bundles() { + Ok(bundles) => bundles, + Err(error) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to load WeChat pending inbound bundles: {error}"), + ); + return; + } + }; + let carried_pending_keys: HashSet = pending_inbound.keys().cloned().collect(); + let mut pending_inbound_changed = false; match api::get_updates(&config, &cursor) { Ok(response) => { if response.errcode == Some(-14) { - mark_session_expired(); channel_host::log( channel_host::LogLevel::Error, - "WeChat session expired; reconnect the channel", + "WeChat getUpdates returned errcode=-14; reconnect the channel", ); return; } @@ -130,7 +147,39 @@ impl Guest for WechatChannel { context_tokens_changed |= changed; } } - emit_incoming_message(message); + match incoming_bundle_from_message(&config, message) { + Ok(Some(bundle)) => { + let emitted = process_incoming_bundle( + &mut pending_inbound, + bundle, + &mut pending_inbound_changed, + ); + for emitted_bundle in emitted { + emit_buffered_bundle(emitted_bundle); + } + } + Ok(None) => {} + Err(error) => { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to map WeChat inbound message: {error}"), + ); + } + } + } + + for key in carried_pending_keys { + if let Some(bundle) = pending_inbound.remove(&key) { + pending_inbound_changed = true; + log_channel( + channel_host::LogLevel::Info, + &format!( + "Flushing buffered WeChat image-only message for {} after waiting one poll cycle", + bundle.from_user_id + ), + ); + emit_buffered_bundle(bundle); + } } if context_tokens_changed { @@ -141,6 +190,15 @@ impl Guest for WechatChannel { ); } } + + if pending_inbound_changed { + if let Err(error) = persist_pending_inbound_bundles(&pending_inbound) { + channel_host::log( + channel_host::LogLevel::Warn, + &format!("Failed to persist WeChat pending inbound bundles: {error}"), + ); + } + } } Err(error) => { channel_host::log( @@ -173,12 +231,7 @@ impl Guest for WechatChannel { ); } - api::send_text_message( - &config, - &metadata.from_user_id, - response.content.trim(), - context_token.as_deref(), - ) + send_response(&config, &metadata, &response, context_token.as_deref()) } fn on_status(update: StatusUpdate) { @@ -225,38 +278,186 @@ impl Guest for WechatChannel { fn on_shutdown() {} } -fn emit_incoming_message(message: WechatMessage) { +fn incoming_bundle_from_message( + config: &WechatConfig, + message: WechatMessage, +) -> Result, String> { if message.message_type != Some(MESSAGE_TYPE_USER) { - return; + return Ok(None); } - let Some(from_user_id) = message.from_user_id.as_deref() else { - return; + let from_user_id = match message.from_user_id.as_deref() { + Some(user_id) => user_id, + None => return Ok(None), }; let text = extract_text(&message); - if text.trim().is_empty() { - return; + let attachments = media::extract_image_attachments(config, &message)? + .into_iter() + .map(StoredInboundAttachment::from) + .collect::>(); + if text.trim().is_empty() && attachments.is_empty() { + return Ok(None); } + Ok(Some(PendingInboundBundle { + from_user_id: from_user_id.to_string(), + to_user_id: message.to_user_id, + session_id: message.session_id, + context_token: message.context_token, + message_id: message.message_id, + text, + attachments, + })) +} + +fn process_incoming_bundle( + pending_inbound: &mut std::collections::HashMap, + bundle: PendingInboundBundle, + pending_inbound_changed: &mut bool, +) -> Vec { + let key = bundle.from_user_id.clone(); + let bundle_has_text = !bundle.text.trim().is_empty(); + let bundle_has_attachments = !bundle.attachments.is_empty(); + + if let Some(mut pending) = pending_inbound.remove(&key) { + *pending_inbound_changed = true; + + if bundle_has_text { + let incoming_metadata = bundle.clone(); + 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_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 image-only message for {} and will wait one poll cycle for follow-up text", + bundle.from_user_id + ), + ); + pending_inbound.insert(key, bundle); + Vec::new() + } else { + vec![bundle] + } +} + +fn emit_buffered_bundle(bundle: PendingInboundBundle) { let metadata = json!({ - "from_user_id": from_user_id, - "to_user_id": message.to_user_id, - "message_id": message.message_id, - "session_id": message.session_id, - "context_token": message.context_token, + "from_user_id": bundle.from_user_id, + "to_user_id": bundle.to_user_id, + "message_id": bundle.message_id, + "session_id": bundle.session_id, + "context_token": bundle.context_token, }); channel_host::emit_message(&EmittedMessage { - user_id: from_user_id.to_string(), + user_id: bundle.from_user_id.clone(), user_name: None, - content: text, - thread_id: Some(format!("wechat:{from_user_id}")), + content: bundle.text, + thread_id: Some(format!("wechat:{}", bundle.from_user_id)), metadata_json: metadata.to_string(), - attachments: Vec::new(), + attachments: bundle.attachments.into_iter().map(Into::into).collect(), }); } +fn merge_bundle_metadata(target: &mut PendingInboundBundle, incoming: &PendingInboundBundle) { + if incoming.to_user_id.is_some() { + target.to_user_id = incoming.to_user_id.clone(); + } + if incoming.session_id.is_some() { + target.session_id = incoming.session_id.clone(); + } + if incoming.context_token.is_some() { + target.context_token = incoming.context_token.clone(); + } + if incoming.message_id.is_some() { + target.message_id = incoming.message_id; + } +} + +fn merge_text(existing: &str, incoming: &str) -> String { + let existing = existing.trim(); + let incoming = incoming.trim(); + match (existing.is_empty(), incoming.is_empty()) { + (true, true) => String::new(), + (true, false) => incoming.to_string(), + (false, true) => existing.to_string(), + (false, false) => format!("{existing}\n{incoming}"), + } +} + +fn send_response( + config: &WechatConfig, + metadata: &OutboundMetadata, + response: &AgentResponse, + context_token: Option<&str>, +) -> Result<(), String> { + let mut remaining_text = response.content.trim().to_string(); + let mut sent_attachment = false; + + for attachment in &response.attachments { + if !attachment.mime_type.starts_with("image/") { + return Err(format!( + "WeChat currently supports image attachments only, got {} ({})", + attachment.filename, attachment.mime_type + )); + } + + let caption = if sent_attachment { + "" + } else { + remaining_text.as_str() + }; + media::send_image_attachment( + config, + &metadata.from_user_id, + attachment, + context_token, + caption, + )?; + sent_attachment = true; + remaining_text.clear(); + } + + if !remaining_text.is_empty() || !sent_attachment { + api::send_text_message( + config, + &metadata.from_user_id, + &remaining_text, + context_token, + )?; + } + + Ok(()) +} + fn extract_text(message: &WechatMessage) -> String { message .item_list @@ -413,9 +614,37 @@ export!(WechatChannel); #[cfg(test)] mod tests { - use super::{classify_status_update, WechatStatusAction}; + use std::collections::HashMap; + + use super::{ + classify_status_update, merge_text, process_incoming_bundle, PendingInboundBundle, + StoredInboundAttachment, WechatStatusAction, + }; use crate::exports::near::agent::channel::{StatusType, StatusUpdate}; + fn make_bundle(user_id: &str, text: &str, image_count: usize) -> PendingInboundBundle { + PendingInboundBundle { + from_user_id: user_id.to_string(), + to_user_id: Some("bot".to_string()), + session_id: Some("session-1".to_string()), + context_token: Some("ctx-1".to_string()), + message_id: Some(1), + text: text.to_string(), + attachments: (0..image_count) + .map(|index| StoredInboundAttachment { + id: format!("att-{index}"), + mime_type: "image/jpeg".to_string(), + filename: Some(format!("photo-{index}.jpg")), + size_bytes: Some(128), + source_url: Some("https://example.com/image.jpg".to_string()), + storage_key: None, + extracted_text: None, + extras_json: "{}".to_string(), + }) + .collect(), + } + } + #[test] fn test_classify_status_update_thinking_starts_typing() { let update = StatusUpdate { @@ -493,4 +722,51 @@ mod tests { assert_eq!(classify_status_update(&update), None); } + + #[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", ""), "look"); + } + + #[test] + fn test_process_incoming_bundle_merges_buffered_image_with_follow_up_text() { + let mut pending = HashMap::new(); + let mut changed = false; + + let emitted = process_incoming_bundle(&mut pending, make_bundle("u1", "", 1), &mut changed); + assert!(emitted.is_empty()); + assert!(changed); + assert_eq!(pending.len(), 1); + + changed = false; + let emitted = process_incoming_bundle( + &mut pending, + make_bundle("u1", "What is in this image?", 0), + &mut changed, + ); + assert!(changed); + assert!(pending.is_empty()); + assert_eq!(emitted.len(), 1); + assert_eq!(emitted[0].text, "What is in this image?"); + assert_eq!(emitted[0].attachments.len(), 1); + } + + #[test] + fn test_process_incoming_bundle_emits_text_and_images_together_without_buffering() { + let mut pending = HashMap::new(); + let mut changed = false; + + let emitted = process_incoming_bundle( + &mut pending, + make_bundle("u1", "Look at this image", 1), + &mut changed, + ); + assert!(!changed); + assert!(pending.is_empty()); + assert_eq!(emitted.len(), 1); + assert_eq!(emitted[0].text, "Look at this image"); + assert_eq!(emitted[0].attachments.len(), 1); + } } diff --git a/channels-src/wechat/src/media.rs b/channels-src/wechat/src/media.rs new file mode 100644 index 00000000..d163a376 --- /dev/null +++ b/channels-src/wechat/src/media.rs @@ -0,0 +1,355 @@ +use aes::cipher::{generic_array::GenericArray, BlockEncrypt, KeyInit}; +use aes::Aes128; +use base64::Engine as _; +use md5::{Digest, Md5}; +use rand::RngCore; +use serde_json::json; + +use crate::exports::near::agent::channel::Attachment; +use crate::near::agent::channel_host::{self, InboundAttachment}; +use crate::types::{ + CdnMedia, ImageItem, MessageItem, SendMessageRequest, WechatConfig, MESSAGE_ITEM_IMAGE, + MESSAGE_STATE_FINISH, MESSAGE_TYPE_BOT, UPLOAD_MEDIA_TYPE_IMAGE, +}; + +const AES_BLOCK_SIZE: usize = 16; + +#[derive(Debug, Clone)] +pub struct UploadImage { + pub download_encrypted_query_param: String, + pub aes_key_base64: String, + pub file_size_ciphertext: u64, +} + +pub fn extract_image_attachments( + config: &WechatConfig, + message: &crate::types::WechatMessage, +) -> Result, String> { + message + .item_list + .iter() + .enumerate() + .filter_map(|(index, item)| map_image_attachment(config, message, item, index).transpose()) + .collect() +} + +pub fn send_image_attachment( + config: &WechatConfig, + to_user_id: &str, + attachment: &Attachment, + context_token: Option<&str>, + text: &str, +) -> Result<(), String> { + if attachment.data.is_empty() { + return Err(format!( + "WeChat image attachment '{}' has no data", + attachment.filename + )); + } + + let upload = upload_image(config, to_user_id, attachment)?; + if !text.trim().is_empty() { + crate::api::send_text_message(config, to_user_id, text.trim(), context_token)?; + } + let request = SendMessageRequest { + msg: crate::types::OutboundWechatMessage { + from_user_id: String::new(), + to_user_id: to_user_id.to_string(), + client_id: format!("wechat-{}", channel_host::now_millis()), + message_type: MESSAGE_TYPE_BOT, + message_state: MESSAGE_STATE_FINISH, + item_list: vec![MessageItem { + r#type: Some(MESSAGE_ITEM_IMAGE), + text_item: None, + image_item: Some(ImageItem { + media: Some(CdnMedia { + encrypt_query_param: Some(upload.download_encrypted_query_param.clone()), + aes_key: Some(upload.aes_key_base64.clone()), + encrypt_type: Some(1), + }), + aeskey: None, + mid_size: Some(upload.file_size_ciphertext), + }), + }], + context_token: context_token.map(str::to_string), + }, + base_info: crate::api::base_info(), + }; + + crate::api::send_message_request(config, &request) +} + +fn map_image_attachment( + config: &WechatConfig, + message: &crate::types::WechatMessage, + item: &MessageItem, + index: usize, +) -> Result, String> { + if item.r#type != Some(MESSAGE_ITEM_IMAGE) { + return Ok(None); + } + + let image = item.image_item.as_ref().ok_or_else(|| { + format!( + "WeChat image message {:?} is missing image_item payload", + message.message_id + ) + })?; + let media = image.media.as_ref().ok_or_else(|| { + format!( + "WeChat image message {:?} is missing media payload", + message.message_id + ) + })?; + let encrypt_query_param = media.encrypt_query_param.as_deref().ok_or_else(|| { + format!( + "WeChat image message {:?} is missing encrypt_query_param", + message.message_id + ) + })?; + let message_id = message + .message_id + .ok_or_else(|| "WeChat image message is missing message_id".to_string())?; + let aes_key = preferred_image_aes_key(image, media).map(str::to_string); + + Ok(Some(InboundAttachment { + id: format!("wechat-image-{}-{}", message_id, index), + mime_type: "image/jpeg".to_string(), + filename: Some(format!("wechat-image-{}-{}.jpg", message_id, index)), + size_bytes: image.mid_size, + source_url: Some(build_cdn_download_url( + &config.cdn_base_url, + encrypt_query_param, + )), + storage_key: None, + extracted_text: None, + extras_json: json!({ "wechat_aes_key": aes_key }).to_string(), + })) +} + +fn preferred_image_aes_key<'a>(image: &'a ImageItem, media: &'a CdnMedia) -> Option<&'a str> { + image + .aeskey + .as_deref() + .filter(|value| !value.trim().is_empty()) + .or_else(|| { + media + .aes_key + .as_deref() + .filter(|value| !value.trim().is_empty()) + }) +} + +fn upload_image( + config: &WechatConfig, + to_user_id: &str, + attachment: &Attachment, +) -> Result { + let plaintext = &attachment.data; + let raw_size = plaintext.len() as u64; + let raw_md5 = hex_lower(md5_bytes(plaintext)); + let file_size_ciphertext = padded_size(raw_size); + let filekey = hex_lower(random_bytes(16)?); + let aes_key = random_bytes(16)?; + let aes_key_hex = hex_lower(aes_key.clone()); + + let upload_url = crate::api::get_upload_url( + config, + &crate::types::GetUploadUrlRequest { + filekey: filekey.clone(), + media_type: UPLOAD_MEDIA_TYPE_IMAGE, + to_user_id: to_user_id.to_string(), + rawsize: raw_size, + rawfilemd5: raw_md5, + filesize: file_size_ciphertext, + no_need_thumb: true, + aeskey: aes_key_hex, + base_info: crate::api::base_info(), + }, + )?; + + let upload_param = upload_url + .upload_param + .as_deref() + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| "WeChat getUploadUrl returned no upload_param".to_string())?; + + if upload_url.thumb_upload_param.is_some() { + channel_host::log( + channel_host::LogLevel::Debug, + "WeChat image upload returned thumb_upload_param; ignoring for single-image flow", + ); + } + + let ciphertext = encrypt_aes_ecb_pkcs7(plaintext, &aes_key)?; + let upload_response = channel_host::http_request( + "POST", + &build_cdn_upload_url(&config.cdn_base_url, upload_param, &filekey), + r#"{"Content-Type":"application/octet-stream"}"#, + Some(&ciphertext), + Some(15_000), + ) + .map_err(|e| format!("WeChat CDN upload failed: {e}"))?; + + if upload_response.status != 200 { + let body = String::from_utf8_lossy(&upload_response.body); + return Err(format!( + "WeChat CDN upload returned {}: {}", + upload_response.status, body + )); + } + + let headers: std::collections::HashMap = + serde_json::from_str(&upload_response.headers_json) + .map_err(|e| format!("Failed to parse WeChat CDN upload headers: {e}"))?; + let download_encrypted_query_param = headers + .iter() + .find_map(|(key, value)| { + if key.eq_ignore_ascii_case("x-encrypted-param") { + Some(value.clone()) + } else { + None + } + }) + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| "WeChat CDN upload response missing x-encrypted-param".to_string())?; + + Ok(UploadImage { + download_encrypted_query_param, + aes_key_base64: base64::engine::general_purpose::STANDARD.encode(aes_key), + file_size_ciphertext, + }) +} + +fn build_cdn_download_url(cdn_base_url: &str, encrypted_query_param: &str) -> String { + format!( + "{}/download?encrypted_query_param={}", + cdn_base_url.trim_end_matches('/'), + percent_encode(encrypted_query_param) + ) +} + +fn build_cdn_upload_url(cdn_base_url: &str, upload_param: &str, filekey: &str) -> String { + format!( + "{}/upload?encrypted_query_param={}&filekey={}", + cdn_base_url.trim_end_matches('/'), + percent_encode(upload_param), + percent_encode(filekey) + ) +} + +fn percent_encode(value: &str) -> String { + let mut encoded = String::with_capacity(value.len()); + for byte in value.bytes() { + if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') { + encoded.push(byte as char); + } else { + encoded.push('%'); + encoded.push(nibble_to_hex(byte >> 4)); + encoded.push(nibble_to_hex(byte & 0x0F)); + } + } + encoded +} + +fn nibble_to_hex(nibble: u8) -> char { + match nibble { + 0..=9 => (b'0' + nibble) as char, + 10..=15 => (b'A' + (nibble - 10)) as char, + _ => '0', + } +} + +fn encode_hex(bytes: &[u8]) -> String { + let mut out = String::with_capacity(bytes.len() * 2); + for byte in bytes { + out.push(nibble_to_hex(byte >> 4)); + out.push(nibble_to_hex(byte & 0x0F)); + } + out +} + +fn hex_lower(bytes: Vec) -> String { + encode_hex(&bytes).to_ascii_lowercase() +} + +fn encrypt_aes_ecb_pkcs7(plaintext: &[u8], key: &[u8]) -> Result, String> { + let cipher = Aes128::new_from_slice(key).map_err(|e| format!("Invalid AES key: {e}"))?; + let mut padded = plaintext.to_vec(); + let pad_len = AES_BLOCK_SIZE - (padded.len() % AES_BLOCK_SIZE); + padded.extend(std::iter::repeat_n(pad_len as u8, pad_len)); + + for chunk in padded.chunks_exact_mut(AES_BLOCK_SIZE) { + cipher.encrypt_block(GenericArray::from_mut_slice(chunk)); + } + + Ok(padded) +} + +fn md5_bytes(bytes: &[u8]) -> Vec { + Md5::digest(bytes).to_vec() +} + +fn random_bytes(len: usize) -> Result, String> { + let mut bytes = vec![0u8; len]; + rand::rngs::OsRng.fill_bytes(&mut bytes); + if bytes.iter().all(|byte| *byte == 0) { + return Err("OS RNG returned all-zero bytes unexpectedly".to_string()); + } + Ok(bytes) +} + +fn padded_size(raw_size: u64) -> u64 { + ((raw_size / AES_BLOCK_SIZE as u64) + 1) * AES_BLOCK_SIZE as u64 +} + +#[cfg(test)] +mod tests { + use super::{encode_hex, encrypt_aes_ecb_pkcs7, map_image_attachment, AES_BLOCK_SIZE}; + use crate::types::{ + CdnMedia, ImageItem, MessageItem, WechatConfig, WechatMessage, MESSAGE_ITEM_IMAGE, + }; + + #[test] + fn test_encrypt_aes_ecb_pkcs7_is_block_aligned() { + let key = [0x11u8; 16]; + let plaintext = b"wechat image payload".to_vec(); + let ciphertext = encrypt_aes_ecb_pkcs7(&plaintext, &key).unwrap(); + assert_eq!(ciphertext.len() % AES_BLOCK_SIZE, 0); + assert_ne!(ciphertext, plaintext); + assert_eq!( + encode_hex(&ciphertext).to_ascii_lowercase(), + "a7464c94a03fb2c5aa783597a1d2f5a461f1cd5d83a7bd92721e8ac1853f881f" + ); + } + + #[test] + fn test_map_image_attachment_errors_when_message_id_missing() { + let config = WechatConfig::default(); + let message = WechatMessage { + message_id: None, + from_user_id: Some("user-1".to_string()), + to_user_id: Some("bot-1".to_string()), + session_id: None, + message_type: None, + context_token: None, + item_list: vec![MessageItem { + r#type: Some(MESSAGE_ITEM_IMAGE), + text_item: None, + image_item: Some(ImageItem { + media: Some(CdnMedia { + encrypt_query_param: Some("enc".to_string()), + aes_key: Some("aes".to_string()), + encrypt_type: Some(1), + }), + aeskey: None, + mid_size: Some(128), + }), + }], + }; + + let error = map_image_attachment(&config, &message, &message.item_list[0], 0) + .expect_err("missing message_id should error"); + assert!(error.contains("missing message_id")); + } +} diff --git a/channels-src/wechat/src/state.rs b/channels-src/wechat/src/state.rs index 69f487ba..503e2ed2 100644 --- a/channels-src/wechat/src/state.rs +++ b/channels-src/wechat/src/state.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; use crate::auth::{ - CONFIG_PATH, CONTEXT_TOKENS_PATH, GET_UPDATES_BUF_PATH, SESSION_EXPIRED_PATH, + CONFIG_PATH, CONTEXT_TOKENS_PATH, GET_UPDATES_BUF_PATH, PENDING_INBOUND_PATH, TYPING_TICKETS_PATH, }; use crate::near::agent::channel_host; @@ -15,6 +15,59 @@ pub struct TypingTicketEntry { pub fetched_at_ms: u64, } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct StoredInboundAttachment { + pub id: String, + pub mime_type: String, + pub filename: Option, + pub size_bytes: Option, + pub source_url: Option, + pub storage_key: Option, + pub extracted_text: Option, + pub extras_json: String, +} + +impl From for StoredInboundAttachment { + fn from(value: channel_host::InboundAttachment) -> Self { + Self { + id: value.id, + mime_type: value.mime_type, + filename: value.filename, + size_bytes: value.size_bytes, + source_url: value.source_url, + storage_key: value.storage_key, + extracted_text: value.extracted_text, + extras_json: value.extras_json, + } + } +} + +impl From for channel_host::InboundAttachment { + fn from(value: StoredInboundAttachment) -> Self { + Self { + id: value.id, + mime_type: value.mime_type, + filename: value.filename, + size_bytes: value.size_bytes, + source_url: value.source_url, + storage_key: value.storage_key, + extracted_text: value.extracted_text, + extras_json: value.extras_json, + } + } +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct PendingInboundBundle { + pub from_user_id: String, + pub to_user_id: Option, + pub session_id: Option, + pub context_token: Option, + pub message_id: Option, + pub text: String, + pub attachments: Vec, +} + pub fn load_config() -> WechatConfig { channel_host::workspace_read(CONFIG_PATH) .and_then(|raw| serde_json::from_str::(&raw).ok()) @@ -63,17 +116,42 @@ pub fn persist_typing_tickets(tickets: &HashMap) -> R channel_host::workspace_write(TYPING_TICKETS_PATH, &serialized).map_err(|e| e.to_string()) } -pub fn session_expired() -> bool { - matches!( - channel_host::workspace_read(SESSION_EXPIRED_PATH).as_deref(), - Some("1") - ) +pub fn load_pending_inbound_bundles() -> Result, String> { + parse_pending_inbound_bundles(channel_host::workspace_read(PENDING_INBOUND_PATH).as_deref()) } -pub fn clear_session_expired() { - let _ = channel_host::workspace_write(SESSION_EXPIRED_PATH, "0"); +pub fn persist_pending_inbound_bundles( + bundles: &HashMap, +) -> Result<(), String> { + let serialized = + serde_json::to_string(bundles).map_err(|e| format!("Failed to serialize bundles: {e}"))?; + channel_host::workspace_write(PENDING_INBOUND_PATH, &serialized).map_err(|e| e.to_string()) } -pub fn mark_session_expired() { - let _ = channel_host::workspace_write(SESSION_EXPIRED_PATH, "1"); +fn parse_pending_inbound_bundles( + raw: Option<&str>, +) -> Result, String> { + match raw { + None => Ok(HashMap::new()), + Some(raw) => serde_json::from_str(raw) + .map_err(|e| format!("Failed to parse pending inbound bundles: {e}")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_pending_inbound_bundles_missing_file_returns_empty_map() { + let bundles = parse_pending_inbound_bundles(None).expect("missing state should be empty"); + assert!(bundles.is_empty()); + } + + #[test] + fn test_parse_pending_inbound_bundles_invalid_json_returns_error() { + let error = + parse_pending_inbound_bundles(Some("{not json")).expect_err("invalid json should err"); + assert!(error.contains("Failed to parse pending inbound bundles")); + } } diff --git a/channels-src/wechat/src/types.rs b/channels-src/wechat/src/types.rs index 03b8ad9f..07c5f4c8 100644 --- a/channels-src/wechat/src/types.rs +++ b/channels-src/wechat/src/types.rs @@ -4,6 +4,8 @@ use serde::{Deserialize, Serialize}; pub struct WechatConfig { #[serde(default = "default_base_url")] pub base_url: String, + #[serde(default = "default_cdn_base_url")] + pub cdn_base_url: String, #[serde(default = "default_bot_type")] pub bot_type: String, #[serde(default = "default_poll_interval_ms")] @@ -16,6 +18,10 @@ fn default_base_url() -> String { "https://ilinkai.weixin.qq.com".to_string() } +fn default_cdn_base_url() -> String { + "https://novac2c.cdn.weixin.qq.com/c2c".to_string() +} + fn default_bot_type() -> String { "3".to_string() } @@ -32,6 +38,7 @@ impl Default for WechatConfig { fn default() -> Self { Self { base_url: default_base_url(), + cdn_base_url: default_cdn_base_url(), bot_type: default_bot_type(), poll_interval_ms: default_poll_interval_ms(), long_poll_timeout_ms: default_long_poll_timeout_ms(), @@ -44,6 +51,19 @@ pub struct BaseInfo { pub channel_version: String, } +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct GetUploadUrlRequest { + pub filekey: String, + pub media_type: i32, + pub to_user_id: String, + pub rawsize: u64, + pub rawfilemd5: String, + pub filesize: u64, + pub no_need_thumb: bool, + pub aeskey: String, + pub base_info: BaseInfo, +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct GetUpdatesRequest { pub get_updates_buf: String, @@ -72,6 +92,14 @@ pub struct GetUpdatesResponse { pub get_updates_buf: Option, } +#[derive(Debug, Clone, Deserialize)] +pub struct GetUploadUrlResponse { + #[serde(default)] + pub upload_param: Option, + #[serde(default)] + pub thumb_upload_param: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct SendMessageRequest { pub msg: OutboundWechatMessage, @@ -140,6 +168,8 @@ pub struct MessageItem { pub r#type: Option, #[serde(default)] pub text_item: Option, + #[serde(default)] + pub image_item: Option, } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -147,6 +177,26 @@ pub struct TextItem { pub text: String, } +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct CdnMedia { + #[serde(default)] + pub encrypt_query_param: Option, + #[serde(default)] + pub aes_key: Option, + #[serde(default)] + pub encrypt_type: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct ImageItem { + #[serde(default)] + pub media: Option, + #[serde(default)] + pub aeskey: Option, + #[serde(default)] + pub mid_size: Option, +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct OutboundMetadata { pub from_user_id: String, @@ -164,5 +214,7 @@ pub const MESSAGE_TYPE_USER: i32 = 1; pub const MESSAGE_TYPE_BOT: i32 = 2; pub const MESSAGE_STATE_FINISH: i32 = 2; pub const MESSAGE_ITEM_TEXT: i32 = 1; +pub const MESSAGE_ITEM_IMAGE: i32 = 2; pub const TYPING_STATUS_TYPING: i32 = 1; pub const TYPING_STATUS_CANCEL: i32 = 2; +pub const UPLOAD_MEDIA_TYPE_IMAGE: i32 = 1; diff --git a/channels-src/wechat/wechat.capabilities.json b/channels-src/wechat/wechat.capabilities.json index cc262616..75a28d01 100644 --- a/channels-src/wechat/wechat.capabilities.json +++ b/channels-src/wechat/wechat.capabilities.json @@ -17,7 +17,8 @@ "capabilities": { "http": { "allowlist": [ - { "host": "ilinkai.weixin.qq.com", "path_prefix": "/" } + { "host": "ilinkai.weixin.qq.com", "path_prefix": "/" }, + { "host": "novac2c.cdn.weixin.qq.com", "path_prefix": "/c2c/" } ], "rate_limit": { "requests_per_minute": 60, @@ -41,6 +42,7 @@ }, "config": { "base_url": "https://ilinkai.weixin.qq.com", + "cdn_base_url": "https://novac2c.cdn.weixin.qq.com/c2c", "bot_type": "3", "poll_interval_ms": 30000, "long_poll_timeout_ms": 35000 diff --git a/docs/plans/2026-03-25-wechat-integration-design.md b/docs/plans/2026-03-25-wechat-integration-design.md index 43eb82cb..e741fc97 100644 --- a/docs/plans/2026-03-25-wechat-integration-design.md +++ b/docs/plans/2026-03-25-wechat-integration-design.md @@ -77,6 +77,8 @@ The point of this phase is to keep the channel aligned with upstream behavior wh - `getupdates` long-poll loop - `sendmessage` outbound replies - typing indicators via `getconfig` and `sendtyping` +- inbound image download/decrypt for vision +- outbound image upload/send via `getuploadurl` - `context_token` persistence - `get_updates_buf` persistence - login persistence across restart @@ -90,7 +92,7 @@ The point of this phase is to keep the channel aligned with upstream behavior wh These are upstream features, so they belong on the roadmap, but they do not need to block the first implementation cut: -- media upload/send via `getuploadurl` +- broader media parity beyond images (files/video/voice) We should not spend time listing non-goals that come from outside the upstream capability boundary. diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index a195458d..aba51954 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -295,7 +295,6 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { } else { tool_defs }; - // Update context for this iteration reason_ctx.available_tools = tool_defs; reason_ctx.system_prompt = Some(if force_text { diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index ddfd0c0f..93f53c4a 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -31,7 +31,58 @@ fn requires_preexisting_uuid_thread(channel: &str) -> bool { matches!(channel, "gateway" | "test") } +fn validate_inbound_text_for_message( + safety: &crate::safety::SafetyLayer, + content: &str, + attachments: &[crate::channels::IncomingAttachment], +) -> crate::safety::ValidationResult { + if content.trim().is_empty() && !attachments.is_empty() { + crate::safety::ValidationResult::ok() + } else { + safety.validate_input(content) + } +} + impl Agent { + fn reject_unsafe_inbound_user_message( + &self, + message: &IncomingMessage, + content: &str, + ) -> Option { + let validation = + validate_inbound_text_for_message(self.safety(), content, &message.attachments); + if !validation.is_valid { + let details = validation + .errors + .iter() + .map(|e| format!("{}: {}", e.field, e.message)) + .collect::>() + .join("; "); + return Some(SubmissionResult::error(format!( + "Input rejected by safety validation: {details}", + ))); + } + + let violations = self.safety().check_policy(content); + if violations + .iter() + .any(|rule| rule.action == crate::safety::PolicyAction::Block) + { + return Some(SubmissionResult::error("Input rejected by safety policy.")); + } + + if let Some(warning) = self.safety().scan_inbound_for_secrets(content) { + tracing::warn!( + user = %message.user_id, + channel = %message.channel, + "Inbound message blocked: contains leaked secret" + ); + return Some(SubmissionResult::error(warning)); + } + + None + } + /// Hydrate a historical thread from DB into memory if not already present. /// /// Called before `resolve_thread` so that the session manager finds the @@ -226,34 +277,11 @@ impl Agent { } // Run the same safety checks that the normal path applies - // (validation, policy, secret scan) so that blocked content - // is never stored in pending_messages or serialized. - let validation = self.safety().validate_input(content); - if !validation.is_valid { - let details = validation - .errors - .iter() - .map(|e| format!("{}: {}", e.field, e.message)) - .collect::>() - .join("; "); - return Ok(SubmissionResult::error(format!( - "Input rejected by safety validation: {details}", - ))); - } - let violations = self.safety().check_policy(content); - if violations - .iter() - .any(|rule| rule.action == crate::safety::PolicyAction::Block) + // so blocked content is never stored in pending_messages. + if let Some(rejection) = + self.reject_unsafe_inbound_user_message(message, content) { - return Ok(SubmissionResult::error("Input rejected by safety policy.")); - } - if let Some(warning) = self.safety().scan_inbound_for_secrets(content) { - tracing::warn!( - user = %message.user_id, - channel = %message.channel, - "Queued message blocked: contains leaked secret" - ); - return Ok(SubmissionResult::error(warning)); + return Ok(rejection); } if !thread.queue_message(content.to_string()) { @@ -307,39 +335,11 @@ impl Agent { } } - // Safety validation for user input - let validation = self.safety().validate_input(content); - if !validation.is_valid { - let details = validation - .errors - .iter() - .map(|e| format!("{}: {}", e.field, e.message)) - .collect::>() - .join("; "); - return Ok(SubmissionResult::error(format!( - "Input rejected by safety validation: {}", - details - ))); - } - - let violations = self.safety().check_policy(content); - if violations - .iter() - .any(|rule| rule.action == crate::safety::PolicyAction::Block) - { - return Ok(SubmissionResult::error("Input rejected by safety policy.")); - } - - // Scan inbound messages for secrets (API keys, tokens). - // Catching them here prevents the LLM from echoing them back, which - // would trigger the outbound leak detector and create error loops. - if let Some(warning) = self.safety().scan_inbound_for_secrets(content) { - tracing::warn!( - user = %message.user_id, - channel = %message.channel, - "Inbound message blocked: contains leaked secret" - ); - return Ok(SubmissionResult::error(warning)); + // Validate inbound content before the turn is created. Attachment-only + // messages are allowed to pass through so multimodal channels can send + // an empty text body alongside real image/document payloads. + if let Some(rejection) = self.reject_unsafe_inbound_user_message(message, content) { + return Ok(rejection); } // Handle explicit commands (starting with /) directly @@ -1880,6 +1880,9 @@ fn rebuild_chat_messages_from_db( #[cfg(test)] mod tests { use super::*; + use crate::channels::{AttachmentKind, IncomingAttachment}; + use crate::config::SafetyConfig; + use crate::safety::SafetyLayer; #[test] fn test_rebuild_chat_messages_user_assistant_only() { @@ -2017,6 +2020,45 @@ mod tests { assert_eq!(result[7].content, "Written"); } + #[test] + fn test_validate_inbound_text_rejects_empty_text_without_attachments() { + let safety = SafetyLayer::new(&SafetyConfig { + max_output_length: 10_000, + injection_check_enabled: true, + }); + + let result = validate_inbound_text_for_message(&safety, "", &[]); + assert!(!result.is_valid); + assert_eq!(result.errors.len(), 1); + assert_eq!(result.errors[0].field, "input"); + assert_eq!(result.errors[0].message, "Input cannot be empty"); + } + + #[test] + fn test_validate_inbound_text_allows_empty_text_when_attachments_exist() { + let safety = SafetyLayer::new(&SafetyConfig { + max_output_length: 10_000, + injection_check_enabled: true, + }); + + let attachments = vec![IncomingAttachment { + id: "image-1".to_string(), + kind: AttachmentKind::Image, + mime_type: "image/jpeg".to_string(), + filename: Some("photo.jpg".to_string()), + size_bytes: Some(128), + source_url: Some("https://example.com/photo.jpg".to_string()), + storage_key: None, + extracted_text: None, + data: vec![1, 2, 3], + duration_secs: None, + }]; + + let result = validate_inbound_text_for_message(&safety, "", &attachments); + assert!(result.is_valid); + assert!(result.errors.is_empty()); + } + fn make_db_msg(role: &str, content: &str) -> crate::history::ConversationMessage { crate::history::ConversationMessage { id: uuid::Uuid::new_v4(), diff --git a/src/channels/wasm/attachment_hydration.rs b/src/channels/wasm/attachment_hydration.rs new file mode 100644 index 00000000..5870013d --- /dev/null +++ b/src/channels/wasm/attachment_hydration.rs @@ -0,0 +1,313 @@ +use std::time::Duration; + +use aes::Aes128; +use aes::cipher::{BlockDecrypt, KeyInit, generic_array::GenericArray}; +use base64::Engine as _; +use serde::Deserialize; + +use crate::channels::wasm::capabilities::ChannelCapabilities; +use crate::channels::wasm::host::{Attachment, ChannelHostState}; + +const AES_BLOCK_SIZE: usize = 16; +const MAX_ATTACHMENT_BYTES: usize = 20 * 1024 * 1024; +const WECHAT_CHANNEL_NAME: &str = "wechat"; + +#[derive(Debug, Deserialize)] +struct WechatAttachmentExtras { + #[serde(default)] + wechat_aes_key: Option, +} + +pub(crate) async fn hydrate_attachment_for_channel( + channel_name: &str, + capabilities: &ChannelCapabilities, + attachment: &mut Attachment, +) { + if !should_hydrate_wechat_attachment(channel_name, attachment) { + return; + } + + let Some(source_url) = attachment.source_url.as_deref() else { + return; + }; + let Some(encoded_aes_key) = wechat_aes_key(&attachment.extras_json) else { + tracing::warn!( + channel = %channel_name, + attachment_id = %attachment.id, + "Skipping WeChat image hydration: missing AES key metadata" + ); + return; + }; + + match download_wechat_attachment_bytes(channel_name, capabilities, source_url).await { + Ok(ciphertext) => match decrypt_wechat_image_bytes(&ciphertext, &encoded_aes_key) { + Ok(plaintext) => { + attachment.size_bytes = Some(plaintext.len() as u64); + attachment.mime_type = detect_image_mime(&plaintext).to_string(); + attachment.data = plaintext; + } + Err(error) => { + tracing::warn!( + channel = %channel_name, + attachment_id = %attachment.id, + error = %error, + "Failed to decrypt WeChat image attachment" + ); + } + }, + Err(error) => { + tracing::warn!( + channel = %channel_name, + attachment_id = %attachment.id, + error = %error, + "Failed to download WeChat image attachment" + ); + } + } +} + +fn should_hydrate_wechat_attachment(channel_name: &str, attachment: &Attachment) -> bool { + channel_name == WECHAT_CHANNEL_NAME + && attachment.data.is_empty() + && attachment.mime_type.starts_with("image/") +} + +fn wechat_aes_key(extras_json: &str) -> Option { + if extras_json.trim().is_empty() { + return None; + } + + serde_json::from_str::(extras_json) + .ok() + .and_then(|extras| extras.wechat_aes_key) + .filter(|value| !value.trim().is_empty()) +} + +async fn download_wechat_attachment_bytes( + channel_name: &str, + capabilities: &ChannelCapabilities, + source_url: &str, +) -> Result, String> { + let host_state = ChannelHostState::new(channel_name, capabilities.clone()); + host_state.check_http_allowed(source_url, "GET")?; + + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|e| format!("Failed to build HTTP client: {e}"))?; + + let response = client + .get(source_url) + .timeout(Duration::from_secs(15)) + .send() + .await + .map_err(|e| format!("WeChat CDN download failed: {e}"))?; + + if response.status() != reqwest::StatusCode::OK { + return Err(format!( + "WeChat CDN download returned {}", + response.status() + )); + } + + let bytes = response + .bytes() + .await + .map_err(|e| format!("Failed to read WeChat CDN response body: {e}"))? + .to_vec(); + + if bytes.is_empty() { + return Err("WeChat CDN download returned an empty body".to_string()); + } + if bytes.len() > MAX_ATTACHMENT_BYTES { + return Err(format!( + "WeChat image attachment exceeds {MAX_ATTACHMENT_BYTES} bytes" + )); + } + + Ok(bytes) +} + +fn decrypt_wechat_image_bytes(ciphertext: &[u8], encoded_aes_key: &str) -> Result, String> { + let key = parse_aes_key(encoded_aes_key)?; + decrypt_aes_ecb_pkcs7(ciphertext, &key) +} + +fn parse_aes_key(encoded: &str) -> Result, String> { + let decoded = if encoded.len() == 32 && encoded.bytes().all(|byte| byte.is_ascii_hexdigit()) { + decode_hex(encoded)? + } else { + base64::engine::general_purpose::STANDARD + .decode(encoded) + .map_err(|e| format!("Failed to decode WeChat AES key: {e}"))? + }; + + if decoded.len() == AES_BLOCK_SIZE { + return Ok(decoded); + } + + if decoded.len() == 32 && decoded.iter().all(|byte| byte.is_ascii_hexdigit()) { + return decode_hex( + std::str::from_utf8(&decoded) + .map_err(|e| format!("WeChat AES key hex payload is not valid UTF-8: {e}"))?, + ); + } + + Err(format!( + "WeChat AES key must decode to 16 bytes or a 32-char hex string, got {} bytes", + decoded.len() + )) +} + +fn decode_hex(input: &str) -> Result, String> { + if !input.len().is_multiple_of(2) { + return Err("hex input length must be even".to_string()); + } + let mut bytes = Vec::with_capacity(input.len() / 2); + let chars: Vec = input.as_bytes().to_vec(); + for idx in (0..chars.len()).step_by(2) { + let high = from_hex_digit(chars[idx])?; + let low = from_hex_digit(chars[idx + 1])?; + bytes.push((high << 4) | low); + } + Ok(bytes) +} + +fn from_hex_digit(value: u8) -> Result { + match value { + b'0'..=b'9' => Ok(value - b'0'), + b'a'..=b'f' => Ok(value - b'a' + 10), + b'A'..=b'F' => Ok(value - b'A' + 10), + _ => Err(format!("invalid hex digit '{}'", value as char)), + } +} + +fn decrypt_aes_ecb_pkcs7(ciphertext: &[u8], key: &[u8]) -> Result, String> { + if !ciphertext.len().is_multiple_of(AES_BLOCK_SIZE) { + return Err("ciphertext length is not a multiple of 16 bytes".to_string()); + } + + let cipher = Aes128::new_from_slice(key).map_err(|e| format!("Invalid AES key: {e}"))?; + let mut plaintext = ciphertext.to_vec(); + for chunk in plaintext.chunks_exact_mut(AES_BLOCK_SIZE) { + cipher.decrypt_block(GenericArray::from_mut_slice(chunk)); + } + + let pad_len = *plaintext + .last() + .ok_or_else(|| "ciphertext decrypted to an empty buffer".to_string())? + as usize; + if pad_len == 0 || pad_len > AES_BLOCK_SIZE || pad_len > plaintext.len() { + return Err("invalid PKCS7 padding".to_string()); + } + if !plaintext[plaintext.len() - pad_len..] + .iter() + .all(|byte| *byte as usize == pad_len) + { + return Err("invalid PKCS7 padding bytes".to_string()); + } + plaintext.truncate(plaintext.len() - pad_len); + Ok(plaintext) +} + +fn detect_image_mime(bytes: &[u8]) -> &'static str { + if bytes.starts_with(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A]) { + "image/png" + } else if bytes.starts_with(&[0xFF, 0xD8, 0xFF]) { + "image/jpeg" + } else if bytes.starts_with(b"GIF87a") || bytes.starts_with(b"GIF89a") { + "image/gif" + } else if bytes.len() >= 12 && &bytes[..4] == b"RIFF" && &bytes[8..12] == b"WEBP" { + "image/webp" + } else { + "image/jpeg" + } +} + +#[cfg(test)] +fn encrypt_aes_ecb_pkcs7(plaintext: &[u8], key: &[u8]) -> Result, String> { + use aes::cipher::BlockEncrypt; + + let cipher = Aes128::new_from_slice(key).map_err(|e| format!("Invalid AES key: {e}"))?; + let mut padded = plaintext.to_vec(); + let pad_len = AES_BLOCK_SIZE - (padded.len() % AES_BLOCK_SIZE); + padded.extend(std::iter::repeat_n(pad_len as u8, pad_len)); + + for chunk in padded.chunks_exact_mut(AES_BLOCK_SIZE) { + cipher.encrypt_block(GenericArray::from_mut_slice(chunk)); + } + + Ok(padded) +} + +#[cfg(test)] +mod tests { + use super::{ + Attachment, decrypt_wechat_image_bytes, detect_image_mime, encrypt_aes_ecb_pkcs7, + hydrate_attachment_for_channel, should_hydrate_wechat_attachment, + }; + use crate::channels::wasm::ChannelCapabilities; + use base64::Engine as _; + + fn make_attachment() -> Attachment { + Attachment { + id: "wechat-image-1".to_string(), + mime_type: "image/jpeg".to_string(), + filename: Some("wechat-image.jpg".to_string()), + size_bytes: None, + source_url: Some( + "https://novac2c.cdn.weixin.qq.com/c2c/download?encrypted_query_param=test" + .to_string(), + ), + storage_key: None, + extracted_text: None, + extras_json: String::new(), + data: Vec::new(), + duration_secs: None, + } + } + + fn encode_test_extras_json(aes_key: &str) -> String { + serde_json::json!({ "wechat_aes_key": aes_key }).to_string() + } + + #[test] + fn decrypt_wechat_image_bytes_round_trips() { + let key = [7u8; 16]; + let plaintext = vec![0xFF, 0xD8, 0xFF, 0xDB, 0x00, 0x11]; + let ciphertext = encrypt_aes_ecb_pkcs7(&plaintext, &key).unwrap(); + let encoded_key = base64::engine::general_purpose::STANDARD.encode(key); + let decrypted = decrypt_wechat_image_bytes(&ciphertext, &encoded_key).unwrap(); + assert_eq!(decrypted, plaintext); + } + + #[test] + fn detect_image_mime_prefers_magic_bytes() { + assert_eq!(detect_image_mime(&[0xFF, 0xD8, 0xFF, 0x00]), "image/jpeg"); + assert_eq!( + detect_image_mime(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A]), + "image/png" + ); + } + + #[test] + fn wechat_attachment_hydration_only_applies_to_wechat_images() { + let mut attachment = make_attachment(); + attachment.extras_json = encode_test_extras_json("ZmFrZS1rZXk="); + assert!(should_hydrate_wechat_attachment("wechat", &attachment)); + assert!(!should_hydrate_wechat_attachment("telegram", &attachment)); + + attachment.mime_type = "application/pdf".to_string(); + assert!(!should_hydrate_wechat_attachment("wechat", &attachment)); + } + + #[tokio::test] + async fn hydration_skips_when_metadata_is_missing() { + let mut attachment = make_attachment(); + let caps = ChannelCapabilities::for_channel("wechat"); + hydrate_attachment_for_channel("wechat", &caps, &mut attachment).await; + assert!(attachment.data.is_empty()); + assert_eq!(attachment.size_bytes, None); + } +} diff --git a/src/channels/wasm/host.rs b/src/channels/wasm/host.rs index eeaccb20..24a207ce 100644 --- a/src/channels/wasm/host.rs +++ b/src/channels/wasm/host.rs @@ -35,6 +35,8 @@ pub struct Attachment { pub storage_key: Option, /// Extracted text content (e.g., OCR result, PDF text, audio transcript). pub extracted_text: Option, + /// Extensible metadata from the channel payload. + pub extras_json: String, /// Raw file bytes (for small files downloaded by the channel). pub data: Vec, /// Duration in seconds (for audio/video). @@ -995,6 +997,7 @@ mod tests { source_url: None, storage_key: None, extracted_text: None, + extras_json: String::new(), data: Vec::new(), duration_secs: None, } diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs index 882709a9..19d3df9a 100644 --- a/src/channels/wasm/mod.rs +++ b/src/channels/wasm/mod.rs @@ -78,6 +78,7 @@ //! } //! ``` +mod attachment_hydration; mod bundled; mod capabilities; mod error; diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index 510bc461..596ab6ef 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -159,6 +159,11 @@ impl WasmChannelRouter { self.channels.read().await.get(channel_name).cloned() } + /// Get a registered channel directly by name. + pub async fn get_channel_by_name(&self, channel_name: &str) -> Option> { + self.channels.read().await.get(channel_name).cloned() + } + /// Validate a secret for a channel. pub async fn validate_secret(&self, channel_name: &str, provided: &str) -> bool { let secrets = self.secrets.read().await; @@ -710,6 +715,10 @@ mod tests { // Should not find non-existent path let not_found = router.get_channel_for_path("/webhook/telegram").await; assert!(not_found.is_none()); + + let found_by_name = router.get_channel_by_name("slack").await; + assert!(found_by_name.is_some()); + assert_eq!(found_by_name.unwrap().channel_name(), "slack"); } #[tokio::test] diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 65e4de88..b60bdaf2 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -573,6 +573,7 @@ impl near::agent::channel_host::Host for ChannelStoreData { source_url: a.source_url, storage_key: a.storage_key, extracted_text: a.extracted_text, + extras_json: a.extras_json, data, duration_secs, } @@ -1181,22 +1182,32 @@ impl WasmChannel { ) } - fn log_on_start_host_state(&self, host_state: &mut ChannelHostState) { + fn log_host_state_entries(channel_name: &str, host_state: &mut ChannelHostState) { for entry in host_state.take_logs() { match entry.level { + crate::tools::wasm::LogLevel::Trace => { + tracing::trace!(channel = %channel_name, "{}", entry.message); + } + crate::tools::wasm::LogLevel::Debug => { + tracing::debug!(channel = %channel_name, "{}", entry.message); + } + crate::tools::wasm::LogLevel::Info => { + tracing::info!(channel = %channel_name, "{}", entry.message); + } crate::tools::wasm::LogLevel::Error => { - tracing::error!(channel = %self.name, "{}", entry.message); + tracing::error!(channel = %channel_name, "{}", entry.message); } crate::tools::wasm::LogLevel::Warn => { - tracing::warn!(channel = %self.name, "{}", entry.message); - } - _ => { - tracing::debug!(channel = %self.name, "{}", entry.message); + tracing::warn!(channel = %channel_name, "{}", entry.message); } } } } + fn log_on_start_host_state(&self, host_state: &mut ChannelHostState) { + Self::log_host_state_entries(&self.name, host_state); + } + async fn execute_on_start_with_state( &self, ) -> Result<(Result, ChannelHostState), WasmChannelError> { @@ -1480,18 +1491,20 @@ impl WasmChannel { // Call on_poll using the generated typed interface let channel_iface = instance.near_agent_channel(); - channel_iface + let poll_result = channel_iface .call_on_poll(&mut store) - .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel)); let mut host_state = Self::extract_host_state(&mut store, &prepared.name, &capabilities); - // Commit pending workspace writes to the persistent store - let pending_writes = host_state.take_pending_writes(); - workspace_store.commit_writes(&pending_writes); + if poll_result.is_ok() { + // Commit pending workspace writes only after a successful callback. + let pending_writes = host_state.take_pending_writes(); + workspace_store.commit_writes(&pending_writes); + } - Ok(((), host_state)) + Ok((poll_result, host_state)) }) .await .map_err(|e| WasmChannelError::ExecutionPanicked { @@ -1503,7 +1516,10 @@ impl WasmChannel { let channel_name = self.name.clone(); match result { - Ok(Ok(((), mut host_state))) => { + Ok(Ok((poll_result, mut host_state))) => { + Self::log_host_state_entries(&channel_name, &mut host_state); + poll_result?; + // Process emitted messages let emitted = host_state.take_emitted_messages(); self.process_emitted_messages(emitted).await?; @@ -2181,6 +2197,16 @@ impl WasmChannel { }; for emitted in messages { + let EmittedMessage { + user_id, + user_name, + content, + thread_id, + metadata_json, + attachments, + .. + } = emitted; + // Check rate limit — acquire and release the write lock before send().await { let mut rate_limiter = self.rate_limiter.write().await; @@ -2198,55 +2224,41 @@ impl WasmChannel { let (resolved_user_id, is_owner_sender) = resolve_message_scope( &self.owner_scope_id, self.owner_actor_id.as_deref(), - &emitted.user_id, + &user_id, ); // Convert to IncomingMessage - let mut msg = IncomingMessage::new(&self.name, &resolved_user_id, &emitted.content) + let mut msg = IncomingMessage::new(&self.name, &resolved_user_id, &content) .with_owner_id(&self.owner_scope_id) - .with_sender_id(&emitted.user_id); + .with_sender_id(&user_id); - if let Some(name) = emitted.user_name { + if let Some(name) = user_name { msg = msg.with_user_name(name); } - if let Some(thread_id) = emitted.thread_id { + if let Some(thread_id) = thread_id { msg = msg.with_thread(thread_id); } // Convert attachments - if !emitted.attachments.is_empty() { - let incoming_attachments = emitted - .attachments - .iter() - .map(|a| crate::channels::IncomingAttachment { - id: a.id.clone(), - kind: crate::channels::AttachmentKind::from_mime_type(&a.mime_type), - mime_type: a.mime_type.clone(), - filename: a.filename.clone(), - size_bytes: a.size_bytes, - source_url: a.source_url.clone(), - storage_key: a.storage_key.clone(), - extracted_text: a.extracted_text.clone(), - data: a.data.clone(), - duration_secs: a.duration_secs, - }) - .collect(); + if !attachments.is_empty() { + let incoming_attachments = + convert_emitted_attachments(&self.name, &self.capabilities, attachments).await; msg = msg.with_attachments(incoming_attachments); } // Parse metadata JSON - msg = apply_emitted_metadata(msg, &emitted.metadata_json); + msg = apply_emitted_metadata(msg, &metadata_json); if is_owner_sender { // Store for owner-target routing (chat_id etc.). - self.update_broadcast_metadata(&emitted.metadata_json).await; + self.update_broadcast_metadata(&metadata_json).await; } // Send to stream — no locks held across this await tracing::info!( channel = %self.name, - user_id = %emitted.user_id, - content_len = emitted.content.len(), + user_id = %user_id, + content_len = content.len(), attachment_count = msg.attachments.len(), "Sending emitted message to agent" ); @@ -2331,6 +2343,7 @@ impl WasmChannel { && let Err(e) = Self::dispatch_emitted_messages( EmitDispatchContext { channel_name: &channel_name, + capabilities: &capabilities, owner_scope_id: &owner_scope_id, owner_actor_id: owner_actor_id.as_deref(), message_tx: &message_tx, @@ -2416,18 +2429,20 @@ impl WasmChannel { // Call on_poll using the generated typed interface let channel_iface = instance.near_agent_channel(); - channel_iface + let poll_result = channel_iface .call_on_poll(&mut store) - .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel)); let mut host_state = Self::extract_host_state(&mut store, &prepared.name, &capabilities); - // Commit pending workspace writes to the persistent store - let pending_writes = host_state.take_pending_writes(); - workspace_store.commit_writes(&pending_writes); + if poll_result.is_ok() { + // Commit pending workspace writes only after a successful callback. + let pending_writes = host_state.take_pending_writes(); + workspace_store.commit_writes(&pending_writes); + } - Ok(host_state) + Ok((poll_result, host_state)) }) .await .map_err(|e| WasmChannelError::ExecutionPanicked { @@ -2438,7 +2453,10 @@ impl WasmChannel { .await; match result { - Ok(Ok(mut host_state)) => { + Ok(Ok((poll_result, mut host_state))) => { + Self::log_host_state_entries(channel_name, &mut host_state); + poll_result?; + let emitted = host_state.take_emitted_messages(); tracing::debug!( channel = %channel_name, @@ -2484,6 +2502,16 @@ impl WasmChannel { }; for emitted in messages { + let EmittedMessage { + user_id, + user_name, + content, + thread_id, + metadata_json, + attachments, + .. + } = emitted; + // Check rate limit — acquire and release the write lock before send().await { let mut limiter = dispatch.rate_limiter.write().await; @@ -2498,54 +2526,40 @@ impl WasmChannel { } } - let (resolved_user_id, is_owner_sender) = resolve_message_scope( - dispatch.owner_scope_id, - dispatch.owner_actor_id, - &emitted.user_id, - ); + let (resolved_user_id, is_owner_sender) = + resolve_message_scope(dispatch.owner_scope_id, dispatch.owner_actor_id, &user_id); // Convert to IncomingMessage - let mut msg = - IncomingMessage::new(dispatch.channel_name, &resolved_user_id, &emitted.content) - .with_owner_id(dispatch.owner_scope_id) - .with_sender_id(&emitted.user_id); + let mut msg = IncomingMessage::new(dispatch.channel_name, &resolved_user_id, &content) + .with_owner_id(dispatch.owner_scope_id) + .with_sender_id(&user_id); - if let Some(name) = emitted.user_name { + if let Some(name) = user_name { msg = msg.with_user_name(name); } - if let Some(thread_id) = emitted.thread_id { + if let Some(thread_id) = thread_id { msg = msg.with_thread(thread_id); } // Convert attachments - if !emitted.attachments.is_empty() { - let incoming_attachments = emitted - .attachments - .iter() - .map(|a| crate::channels::IncomingAttachment { - id: a.id.clone(), - kind: crate::channels::AttachmentKind::from_mime_type(&a.mime_type), - mime_type: a.mime_type.clone(), - filename: a.filename.clone(), - size_bytes: a.size_bytes, - source_url: a.source_url.clone(), - storage_key: a.storage_key.clone(), - extracted_text: a.extracted_text.clone(), - data: a.data.clone(), - duration_secs: a.duration_secs, - }) - .collect(); + if !attachments.is_empty() { + let incoming_attachments = convert_emitted_attachments( + dispatch.channel_name, + dispatch.capabilities, + attachments, + ) + .await; msg = msg.with_attachments(incoming_attachments); } - msg = apply_emitted_metadata(msg, &emitted.metadata_json); + msg = apply_emitted_metadata(msg, &metadata_json); if is_owner_sender { // Store for owner-target routing (chat_id etc.) do_update_broadcast_metadata( dispatch.channel_name, dispatch.owner_scope_id, - &emitted.metadata_json, + &metadata_json, dispatch.last_broadcast_metadata, dispatch.settings_store, ) @@ -2555,8 +2569,8 @@ impl WasmChannel { // Send to stream — no locks held across this await tracing::info!( channel = %dispatch.channel_name, - user_id = %emitted.user_id, - content_len = emitted.content.len(), + user_id = %user_id, + content_len = content.len(), attachment_count = msg.attachments.len(), "Sending polled message to agent" ); @@ -2581,6 +2595,7 @@ impl WasmChannel { struct EmitDispatchContext<'a> { channel_name: &'a str, + capabilities: &'a ChannelCapabilities, owner_scope_id: &'a str, owner_actor_id: Option<&'a str>, message_tx: &'a RwLock>>, @@ -3243,6 +3258,38 @@ async fn resolve_channel_host_credentials( /// Maximum total attachment size (50 MB). const MAX_TOTAL_ATTACHMENT_BYTES: u64 = 50 * 1024 * 1024; +async fn convert_emitted_attachments( + channel_name: &str, + capabilities: &ChannelCapabilities, + attachments: Vec, +) -> Vec { + let mut hydrated = attachments; + for attachment in &mut hydrated { + crate::channels::wasm::attachment_hydration::hydrate_attachment_for_channel( + channel_name, + capabilities, + attachment, + ) + .await; + } + + hydrated + .into_iter() + .map(|attachment| crate::channels::IncomingAttachment { + id: attachment.id, + kind: crate::channels::AttachmentKind::from_mime_type(&attachment.mime_type), + mime_type: attachment.mime_type, + filename: attachment.filename, + size_bytes: attachment.size_bytes, + source_url: attachment.source_url, + storage_key: attachment.storage_key, + extracted_text: attachment.extracted_text, + data: attachment.data, + duration_secs: attachment.duration_secs, + }) + .collect() +} + /// Detect MIME type from file extension using the `mime_guess` crate. fn mime_from_extension(path: &str) -> String { mime_guess::from_path(path) @@ -3455,6 +3502,8 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(10); let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("test-channel"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( @@ -3471,6 +3520,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "test-channel", + capabilities: &capabilities, owner_scope_id: "default", owner_actor_id: None, message_tx: &message_tx, @@ -3503,6 +3553,8 @@ mod tests { // No sender available (channel not started) let message_tx = Arc::new(tokio::sync::RwLock::new(None)); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("test-channel"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( crate::channels::wasm::capabilities::EmitRateLimitConfig::default(), @@ -3516,6 +3568,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "test-channel", + capabilities: &capabilities, owner_scope_id: "default", owner_actor_id: None, message_tx: &message_tx, @@ -4506,6 +4559,8 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(10); let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("test-channel"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( @@ -4522,6 +4577,7 @@ mod tests { source_url: Some("https://api.telegram.org/file/photo123".to_string()), storage_key: None, extracted_text: None, + extras_json: String::new(), data: Vec::new(), duration_secs: None, }, @@ -4533,6 +4589,7 @@ mod tests { source_url: None, storage_key: Some("store/doc456".to_string()), extracted_text: Some("Report contents...".to_string()), + extras_json: String::new(), data: Vec::new(), duration_secs: None, }, @@ -4545,6 +4602,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "test-channel", + capabilities: &capabilities, owner_scope_id: "default", owner_actor_id: None, message_tx: &message_tx, @@ -4591,6 +4649,8 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(10); let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("telegram"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( crate::channels::wasm::capabilities::EmitRateLimitConfig::default(), @@ -4606,6 +4666,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "telegram", + capabilities: &capabilities, owner_scope_id: "owner-scope", owner_actor_id: Some("telegram-owner"), message_tx: &message_tx, @@ -4634,6 +4695,8 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(10); let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("telegram"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( crate::channels::wasm::capabilities::EmitRateLimitConfig::default(), @@ -4648,6 +4711,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "telegram", + capabilities: &capabilities, owner_scope_id: "owner-scope", owner_actor_id: Some("telegram-owner"), message_tx: &message_tx, @@ -4717,6 +4781,8 @@ mod tests { let (tx, mut rx) = tokio::sync::mpsc::channel(10); let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + let capabilities = + crate::channels::wasm::capabilities::ChannelCapabilities::for_channel("test-channel"); let rate_limiter = Arc::new(tokio::sync::RwLock::new( crate::channels::wasm::host::ChannelEmitRateLimiter::new( @@ -4730,6 +4796,7 @@ mod tests { let result = WasmChannel::dispatch_emitted_messages( EmitDispatchContext { channel_name: "test-channel", + capabilities: &capabilities, owner_scope_id: "default", owner_actor_id: None, message_tx: &message_tx, diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 5c9fb570..48679345 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -4089,6 +4089,10 @@ impl ExtensionManager { let webhook_path = format!("/webhook/{}", name); let existing_channel = match router.get_channel_for_path(&webhook_path).await { + Some(ch) => Some(ch), + None => router.get_channel_by_name(name).await, + }; + let existing_channel = match existing_channel { Some(ch) => ch, None => { return Ok(ActivateResult {