mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
Add WeChat image messaging and QR login polish
This commit is contained in:
Coffee
Generated
+1
@@ -3392,6 +3392,7 @@ dependencies = [
|
||||
name = "ironclaw"
|
||||
version = "0.19.0"
|
||||
dependencies = [
|
||||
"aes",
|
||||
"aes-gcm",
|
||||
"aho-corasick",
|
||||
"anyhow",
|
||||
|
||||
@@ -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"
|
||||
|
||||
+1
-1
@@ -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 |
|
||||
|
||||
Generated
+160
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 ""
|
||||
|
||||
@@ -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<GetUploadUrlResponse, String> {
|
||||
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,
|
||||
|
||||
@@ -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";
|
||||
|
||||
+313
-37
@@ -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<ChannelConfig, String> {
|
||||
let config = serde_json::from_str::<WechatConfig>(&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<String> = 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<Option<PendingInboundBundle>, 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::<Vec<_>>();
|
||||
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<String, PendingInboundBundle>,
|
||||
bundle: PendingInboundBundle,
|
||||
pending_inbound_changed: &mut bool,
|
||||
) -> Vec<PendingInboundBundle> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Vec<InboundAttachment>, 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<Option<InboundAttachment>, 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<UploadImage, String> {
|
||||
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<String, String> =
|
||||
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<u8>) -> String {
|
||||
encode_hex(&bytes).to_ascii_lowercase()
|
||||
}
|
||||
|
||||
fn encrypt_aes_ecb_pkcs7(plaintext: &[u8], key: &[u8]) -> Result<Vec<u8>, 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<u8> {
|
||||
Md5::digest(bytes).to_vec()
|
||||
}
|
||||
|
||||
fn random_bytes(len: usize) -> Result<Vec<u8>, 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"));
|
||||
}
|
||||
}
|
||||
@@ -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<String>,
|
||||
pub size_bytes: Option<u64>,
|
||||
pub source_url: Option<String>,
|
||||
pub storage_key: Option<String>,
|
||||
pub extracted_text: Option<String>,
|
||||
pub extras_json: String,
|
||||
}
|
||||
|
||||
impl From<channel_host::InboundAttachment> 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<StoredInboundAttachment> 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<String>,
|
||||
pub session_id: Option<String>,
|
||||
pub context_token: Option<String>,
|
||||
pub message_id: Option<i64>,
|
||||
pub text: String,
|
||||
pub attachments: Vec<StoredInboundAttachment>,
|
||||
}
|
||||
|
||||
pub fn load_config() -> WechatConfig {
|
||||
channel_host::workspace_read(CONFIG_PATH)
|
||||
.and_then(|raw| serde_json::from_str::<WechatConfig>(&raw).ok())
|
||||
@@ -63,17 +116,42 @@ pub fn persist_typing_tickets(tickets: &HashMap<String, TypingTicketEntry>) -> 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<HashMap<String, PendingInboundBundle>, 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<String, PendingInboundBundle>,
|
||||
) -> 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<HashMap<String, PendingInboundBundle>, 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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct GetUploadUrlResponse {
|
||||
#[serde(default)]
|
||||
pub upload_param: Option<String>,
|
||||
#[serde(default)]
|
||||
pub thumb_upload_param: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct SendMessageRequest {
|
||||
pub msg: OutboundWechatMessage,
|
||||
@@ -140,6 +168,8 @@ pub struct MessageItem {
|
||||
pub r#type: Option<i32>,
|
||||
#[serde(default)]
|
||||
pub text_item: Option<TextItem>,
|
||||
#[serde(default)]
|
||||
pub image_item: Option<ImageItem>,
|
||||
}
|
||||
|
||||
#[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<String>,
|
||||
#[serde(default)]
|
||||
pub aes_key: Option<String>,
|
||||
#[serde(default)]
|
||||
pub encrypt_type: Option<i32>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct ImageItem {
|
||||
#[serde(default)]
|
||||
pub media: Option<CdnMedia>,
|
||||
#[serde(default)]
|
||||
pub aeskey: Option<String>,
|
||||
#[serde(default)]
|
||||
pub mid_size: Option<u64>,
|
||||
}
|
||||
|
||||
#[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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+102
-60
@@ -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<SubmissionResult> {
|
||||
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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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(),
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
if extras_json.trim().is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
serde_json::from_str::<WechatAttachmentExtras>(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<Vec<u8>, 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<Vec<u8>, String> {
|
||||
let key = parse_aes_key(encoded_aes_key)?;
|
||||
decrypt_aes_ecb_pkcs7(ciphertext, &key)
|
||||
}
|
||||
|
||||
fn parse_aes_key(encoded: &str) -> Result<Vec<u8>, 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<Vec<u8>, 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<u8> = 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<u8, String> {
|
||||
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<Vec<u8>, 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<Vec<u8>, 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);
|
||||
}
|
||||
}
|
||||
@@ -35,6 +35,8 @@ pub struct Attachment {
|
||||
pub storage_key: Option<String>,
|
||||
/// Extracted text content (e.g., OCR result, PDF text, audio transcript).
|
||||
pub extracted_text: Option<String>,
|
||||
/// Extensible metadata from the channel payload.
|
||||
pub extras_json: String,
|
||||
/// Raw file bytes (for small files downloaded by the channel).
|
||||
pub data: Vec<u8>,
|
||||
/// 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,
|
||||
}
|
||||
|
||||
@@ -78,6 +78,7 @@
|
||||
//! }
|
||||
//! ```
|
||||
|
||||
mod attachment_hydration;
|
||||
mod bundled;
|
||||
mod capabilities;
|
||||
mod error;
|
||||
|
||||
@@ -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<Arc<WasmChannel>> {
|
||||
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]
|
||||
|
||||
+145
-78
@@ -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<ChannelConfig, WasmChannelError>, 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<Option<mpsc::Sender<IncomingMessage>>>,
|
||||
@@ -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<crate::channels::wasm::host::Attachment>,
|
||||
) -> Vec<crate::channels::IncomingAttachment> {
|
||||
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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user