Add WeChat typing indicators to the DM channel

channels-src/wechat/src/api.rs

@@ -2,9 +2,9 @@ use base64::Engine as _;
 
 use crate::near::agent::channel_host;
 use crate::types::{
-    BaseInfo, GetUpdatesRequest, GetUpdatesResponse, MessageItem, OutboundWechatMessage,
-    SendMessageRequest, TextItem, WechatConfig, MESSAGE_ITEM_TEXT, MESSAGE_STATE_FINISH,
-    MESSAGE_TYPE_BOT,
+    BaseInfo, GetConfigRequest, GetConfigResponse, GetUpdatesRequest, GetUpdatesResponse,
+    MessageItem, OutboundWechatMessage, SendMessageRequest, SendTypingRequest, SendTypingResponse,
+    TextItem, WechatConfig, MESSAGE_ITEM_TEXT, MESSAGE_STATE_FINISH, MESSAGE_TYPE_BOT,
 };
 
 fn base_info() -> BaseInfo {
@@ -114,3 +114,76 @@ pub fn send_text_message(
 
     Ok(())
 }
+
+pub fn get_config(
+    config: &WechatConfig,
+    ilink_user_id: &str,
+    context_token: Option<&str>,
+) -> Result<GetConfigResponse, String> {
+    let body = serde_json::to_vec(&GetConfigRequest {
+        ilink_user_id: ilink_user_id.to_string(),
+        context_token: context_token.map(str::to_string),
+        base_info: base_info(),
+    })
+    .map_err(|e| format!("Failed to encode getConfig request: {e}"))?;
+    let headers = request_headers(&body);
+    let url = format!(
+        "{}ilink/bot/getconfig",
+        ensure_trailing_slash(&config.base_url)
+    );
+
+    let response = channel_host::http_request("POST", &url, &headers, Some(&body), Some(10_000))
+        .map_err(|e| format!("getConfig request failed: {e}"))?;
+
+    if response.status != 200 {
+        let body = String::from_utf8_lossy(&response.body);
+        return Err(format!("getConfig returned {}: {}", response.status, body));
+    }
+
+    serde_json::from_slice(&response.body)
+        .map_err(|e| format!("Failed to parse getConfig response: {e}"))
+}
+
+pub fn send_typing(
+    config: &WechatConfig,
+    ilink_user_id: &str,
+    typing_ticket: &str,
+    status: i32,
+) -> Result<(), String> {
+    let body = serde_json::to_vec(&SendTypingRequest {
+        ilink_user_id: ilink_user_id.to_string(),
+        typing_ticket: typing_ticket.to_string(),
+        status,
+        base_info: base_info(),
+    })
+    .map_err(|e| format!("Failed to encode sendTyping request: {e}"))?;
+    let headers = request_headers(&body);
+    let url = format!(
+        "{}ilink/bot/sendtyping",
+        ensure_trailing_slash(&config.base_url)
+    );
+
+    let response = channel_host::http_request("POST", &url, &headers, Some(&body), Some(10_000))
+        .map_err(|e| format!("sendTyping request failed: {e}"))?;
+
+    if response.status != 200 {
+        let body = String::from_utf8_lossy(&response.body);
+        return Err(format!("sendTyping returned {}: {}", response.status, body));
+    }
+
+    let parsed: SendTypingResponse = serde_json::from_slice(&response.body)
+        .map_err(|e| format!("Failed to parse sendTyping response: {e}"))?;
+
+    if parsed.ret.unwrap_or(0) != 0 {
+        let errmsg = parsed
+            .errmsg
+            .as_deref()
+            .unwrap_or("unknown WeChat sendTyping error");
+        return Err(format!(
+            "sendTyping returned ret={} errmsg={errmsg}",
+            parsed.ret.unwrap_or(-1)
+        ));
+    }
+
+    Ok(())
+}

channels-src/wechat/src/auth.rs

@@ -2,4 +2,5 @@ pub const TOKEN_SECRET_NAME: &str = "wechat_bot_token";
 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";

channels-src/wechat/src/lib.rs

@@ -9,7 +9,7 @@ mod state;
 mod types;
 
 use exports::near::agent::channel::{
-    AgentResponse, ChannelConfig, Guest, PollConfig, StatusUpdate,
+    AgentResponse, ChannelConfig, Guest, PollConfig, StatusType, StatusUpdate,
 };
 use near::agent::channel_host::{self, EmittedMessage};
 use serde_json::json;
@@ -17,13 +17,22 @@ 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,
-    mark_session_expired, persist_config, persist_context_tokens, persist_get_updates_buf,
-    session_expired,
+    load_typing_tickets, mark_session_expired, persist_config, persist_context_tokens,
+    persist_get_updates_buf, persist_typing_tickets, session_expired, TypingTicketEntry,
 };
 use crate::types::{
     OutboundMetadata, WechatConfig, WechatMessage, MESSAGE_ITEM_TEXT, MESSAGE_TYPE_USER,
+    TYPING_STATUS_CANCEL, TYPING_STATUS_TYPING,
 };
 
+const TYPING_TICKET_TTL_MS: u64 = 24 * 60 * 60 * 1000;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum WechatStatusAction {
+    Typing,
+    Cancel,
+}
+
 struct WechatChannel;
 
 impl Guest for WechatChannel {
@@ -151,6 +160,18 @@ impl Guest for WechatChannel {
             .context_token
             .clone()
             .or_else(|| context_tokens.get(&metadata.from_user_id).cloned());
+        if let Err(error) = send_typing_indicator(
+            &config,
+            &metadata,
+            context_token.as_deref(),
+            TYPING_STATUS_CANCEL,
+            false,
+        ) {
+            channel_host::log(
+                channel_host::LogLevel::Debug,
+                &format!("Failed to cancel WeChat typing indicator before reply: {error}"),
+            );
+        }
 
         api::send_text_message(
             &config,
@@ -160,7 +181,42 @@ impl Guest for WechatChannel {
         )
     }
 
-    fn on_status(_update: StatusUpdate) {}
+    fn on_status(update: StatusUpdate) {
+        let Some(action) = classify_status_update(&update) else {
+            return;
+        };
+        let metadata = match serde_json::from_str::<OutboundMetadata>(&update.metadata_json) {
+            Ok(metadata) => metadata,
+            Err(_) => {
+                channel_host::log(
+                    channel_host::LogLevel::Debug,
+                    "on_status: no valid WeChat metadata, skipping typing update",
+                );
+                return;
+            }
+        };
+        let config = load_config();
+        let context_tokens = load_context_tokens();
+        let context_token = resolve_context_token(&metadata, &context_tokens);
+
+        let (typing_status, allow_ticket_fetch) = match action {
+            WechatStatusAction::Typing => (TYPING_STATUS_TYPING, true),
+            WechatStatusAction::Cancel => (TYPING_STATUS_CANCEL, false),
+        };
+
+        if let Err(error) = send_typing_indicator(
+            &config,
+            &metadata,
+            context_token.as_deref(),
+            typing_status,
+            allow_ticket_fetch,
+        ) {
+            channel_host::log(
+                channel_host::LogLevel::Debug,
+                &format!("WeChat typing update failed: {error}"),
+            );
+        }
+    }
 
     fn on_broadcast(_user_id: String, _response: AgentResponse) -> Result<(), String> {
         Ok(())
@@ -215,4 +271,226 @@ fn extract_text(message: &WechatMessage) -> String {
         .unwrap_or_default()
 }
 
+fn is_terminal_text_status(message: &str) -> bool {
+    let trimmed = message.trim();
+    trimmed.eq_ignore_ascii_case("done")
+        || trimmed.eq_ignore_ascii_case("interrupted")
+        || trimmed.eq_ignore_ascii_case("awaiting approval")
+        || trimmed.eq_ignore_ascii_case("rejected")
+}
+
+fn classify_status_update(update: &StatusUpdate) -> Option<WechatStatusAction> {
+    match update.status {
+        StatusType::Thinking => Some(WechatStatusAction::Typing),
+        StatusType::Done
+        | StatusType::Interrupted
+        | StatusType::ApprovalNeeded
+        | StatusType::AuthRequired => Some(WechatStatusAction::Cancel),
+        StatusType::Status if is_terminal_text_status(&update.message) => {
+            Some(WechatStatusAction::Cancel)
+        }
+        StatusType::ToolStarted
+        | StatusType::ToolCompleted
+        | StatusType::ToolResult
+        | StatusType::Status
+        | StatusType::JobStarted
+        | StatusType::AuthCompleted => None,
+    }
+}
+
+fn resolve_context_token(
+    metadata: &OutboundMetadata,
+    context_tokens: &std::collections::HashMap<String, String>,
+) -> Option<String> {
+    metadata
+        .context_token
+        .clone()
+        .or_else(|| context_tokens.get(&metadata.from_user_id).cloned())
+}
+
+fn cached_typing_ticket(user_id: &str) -> Option<String> {
+    let tickets = load_typing_tickets();
+    let ticket = tickets.get(user_id)?;
+    let trimmed = ticket.ticket.trim();
+    if trimmed.is_empty() {
+        return None;
+    }
+
+    let age_ms = channel_host::now_millis().saturating_sub(ticket.fetched_at_ms);
+    if age_ms >= TYPING_TICKET_TTL_MS {
+        return None;
+    }
+
+    Some(trimmed.to_string())
+}
+
+fn persist_typing_ticket(user_id: &str, ticket: &str) -> Result<(), String> {
+    let mut tickets = load_typing_tickets();
+    tickets.insert(
+        user_id.to_string(),
+        TypingTicketEntry {
+            ticket: ticket.to_string(),
+            fetched_at_ms: channel_host::now_millis(),
+        },
+    );
+    persist_typing_tickets(&tickets)
+}
+
+fn clear_typing_ticket(user_id: &str) -> Result<(), String> {
+    let mut tickets = load_typing_tickets();
+    if tickets.remove(user_id).is_some() {
+        persist_typing_tickets(&tickets)?;
+    }
+    Ok(())
+}
+
+fn resolve_typing_ticket(
+    config: &WechatConfig,
+    user_id: &str,
+    context_token: Option<&str>,
+) -> Result<Option<String>, String> {
+    if let Some(ticket) = cached_typing_ticket(user_id) {
+        return Ok(Some(ticket));
+    }
+
+    let response = api::get_config(config, user_id, context_token)?;
+    if response.ret.unwrap_or(0) != 0 {
+        let errmsg = response
+            .errmsg
+            .as_deref()
+            .unwrap_or("unknown WeChat getConfig error");
+        return Err(format!(
+            "WeChat getConfig returned ret={} errmsg={errmsg}",
+            response.ret.unwrap_or(-1)
+        ));
+    }
+
+    let Some(ticket) = response
+        .typing_ticket
+        .as_deref()
+        .map(str::trim)
+        .filter(|ticket| !ticket.is_empty())
+    else {
+        return Ok(None);
+    };
+
+    if let Err(error) = persist_typing_ticket(user_id, ticket) {
+        channel_host::log(
+            channel_host::LogLevel::Warn,
+            &format!("Failed to persist WeChat typing ticket: {error}"),
+        );
+    }
+
+    Ok(Some(ticket.to_string()))
+}
+
+fn send_typing_indicator(
+    config: &WechatConfig,
+    metadata: &OutboundMetadata,
+    context_token: Option<&str>,
+    status: i32,
+    allow_ticket_fetch: bool,
+) -> Result<(), String> {
+    let ticket = if allow_ticket_fetch {
+        resolve_typing_ticket(config, &metadata.from_user_id, context_token)?
+    } else {
+        cached_typing_ticket(&metadata.from_user_id)
+    };
+
+    let Some(ticket) = ticket else {
+        return Ok(());
+    };
+
+    if let Err(error) = api::send_typing(config, &metadata.from_user_id, &ticket, status) {
+        let _ = clear_typing_ticket(&metadata.from_user_id);
+        return Err(error);
+    }
+
+    Ok(())
+}
+
 export!(WechatChannel);
+
+#[cfg(test)]
+mod tests {
+    use super::{classify_status_update, WechatStatusAction};
+    use crate::exports::near::agent::channel::{StatusType, StatusUpdate};
+
+    #[test]
+    fn test_classify_status_update_thinking_starts_typing() {
+        let update = StatusUpdate {
+            status: StatusType::Thinking,
+            message: "Thinking...".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(
+            classify_status_update(&update),
+            Some(WechatStatusAction::Typing)
+        );
+    }
+
+    #[test]
+    fn test_classify_status_update_done_cancels_typing() {
+        let update = StatusUpdate {
+            status: StatusType::Done,
+            message: "Done".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(
+            classify_status_update(&update),
+            Some(WechatStatusAction::Cancel)
+        );
+    }
+
+    #[test]
+    fn test_classify_status_update_approval_needed_cancels_typing() {
+        let update = StatusUpdate {
+            status: StatusType::ApprovalNeeded,
+            message: "Approval needed".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(
+            classify_status_update(&update),
+            Some(WechatStatusAction::Cancel)
+        );
+    }
+
+    #[test]
+    fn test_classify_status_update_tool_started_is_ignored() {
+        let update = StatusUpdate {
+            status: StatusType::ToolStarted,
+            message: "Tool started".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(classify_status_update(&update), None);
+    }
+
+    #[test]
+    fn test_classify_status_update_terminal_text_status_cancels_typing() {
+        let update = StatusUpdate {
+            status: StatusType::Status,
+            message: "Awaiting approval".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(
+            classify_status_update(&update),
+            Some(WechatStatusAction::Cancel)
+        );
+    }
+
+    #[test]
+    fn test_classify_status_update_progress_status_is_ignored() {
+        let update = StatusUpdate {
+            status: StatusType::Status,
+            message: "Context compaction started".to_string(),
+            metadata_json: "{}".to_string(),
+        };
+
+        assert_eq!(classify_status_update(&update), None);
+    }
+}

channels-src/wechat/src/state.rs

@@ -1,9 +1,20 @@
 use std::collections::HashMap;
 
-use crate::auth::{CONFIG_PATH, CONTEXT_TOKENS_PATH, GET_UPDATES_BUF_PATH, SESSION_EXPIRED_PATH};
+use serde::{Deserialize, Serialize};
+
+use crate::auth::{
+    CONFIG_PATH, CONTEXT_TOKENS_PATH, GET_UPDATES_BUF_PATH, SESSION_EXPIRED_PATH,
+    TYPING_TICKETS_PATH,
+};
 use crate::near::agent::channel_host;
 use crate::types::WechatConfig;
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct TypingTicketEntry {
+    pub ticket: String,
+    pub fetched_at_ms: u64,
+}
+
 pub fn load_config() -> WechatConfig {
     channel_host::workspace_read(CONFIG_PATH)
         .and_then(|raw| serde_json::from_str::<WechatConfig>(&raw).ok())
@@ -40,6 +51,18 @@ pub fn persist_context_tokens(tokens: &HashMap<String, String>) -> Result<(), St
     channel_host::workspace_write(CONTEXT_TOKENS_PATH, &serialized).map_err(|e| e.to_string())
 }
 
+pub fn load_typing_tickets() -> HashMap<String, TypingTicketEntry> {
+    channel_host::workspace_read(TYPING_TICKETS_PATH)
+        .and_then(|raw| serde_json::from_str::<HashMap<String, TypingTicketEntry>>(&raw).ok())
+        .unwrap_or_default()
+}
+
+pub fn persist_typing_tickets(tickets: &HashMap<String, TypingTicketEntry>) -> Result<(), String> {
+    let serialized =
+        serde_json::to_string(tickets).map_err(|e| format!("Failed to serialize tickets: {e}"))?;
+    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(),

channels-src/wechat/src/types.rs

@@ -50,6 +50,14 @@ pub struct GetUpdatesRequest {
     pub base_info: BaseInfo,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct GetConfigRequest {
+    pub ilink_user_id: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub context_token: Option<String>,
+    pub base_info: BaseInfo,
+}
+
 #[derive(Debug, Clone, Deserialize)]
 pub struct GetUpdatesResponse {
     #[serde(default)]
@@ -70,6 +78,14 @@ pub struct SendMessageRequest {
     pub base_info: BaseInfo,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct SendTypingRequest {
+    pub ilink_user_id: String,
+    pub typing_ticket: String,
+    pub status: i32,
+    pub base_info: BaseInfo,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct OutboundWechatMessage {
     pub from_user_id: String,
@@ -100,6 +116,24 @@ pub struct WechatMessage {
     pub item_list: Vec<MessageItem>,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+pub struct GetConfigResponse {
+    #[serde(default)]
+    pub ret: Option<i32>,
+    #[serde(default)]
+    pub errmsg: Option<String>,
+    #[serde(default)]
+    pub typing_ticket: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct SendTypingResponse {
+    #[serde(default)]
+    pub ret: Option<i32>,
+    #[serde(default)]
+    pub errmsg: Option<String>,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct MessageItem {
     #[serde(default)]
@@ -130,3 +164,5 @@ 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 TYPING_STATUS_TYPING: i32 = 1;
+pub const TYPING_STATUS_CANCEL: i32 = 2;