diff --git a/src/apps/desktop/README.md b/src/apps/desktop/README.md index b17676b1af..f6be961b6d 100644 --- a/src/apps/desktop/README.md +++ b/src/apps/desktop/README.md @@ -252,3 +252,34 @@ local Desktop scene. Peer mode and older hosts show unsupported states. SSH/Dock project installation remains unsupported; choose user scope to install on the serving host. Mobile/bot controls, CLI peers and Detached Dispatch do not gain a marketplace configuration or installation entry point. + +## WeChat session result notifications + +While the WeChat bot is running, completed turns in its currently selected local +session can send their text to WeChat without a new incoming message. This includes +scheduled jobs and turns started in the desktop window. Turns started by that same +WeChat bot retain their normal reply path and are not pushed a second time. Failed +turns and empty output are not pushed. Use a dedicated session if desktop activity +should not appear in WeChat. + +Delivery still needs a valid WeChat reply context and available channel quota; +this feature does not bypass either restriction or impose an additional daily +three-message limit. Several queued results are combined into one reply of at most +4,000 UTF-8 bytes, favoring the newest results. Longer output is truncated; the full +answer remains in the session. These pushes share the channel quota with normal +replies, including any split replies. + +Unavailable reply context and send failures keep output in a bounded, in-memory +queue for retry. Sending another WeChat message refreshes the reply context and +wakes the sender. The queue holds at most 20 results per recipient for 24 hours; +older entries are discarded. Restarting or replacing the bot clears pending output. +This is best-effort notification, not a durable message inbox. Switching sessions +or devices discards output from the previous selection before sending. + +Proactive notifications currently require the session runtime and WeChat bot to +run on the same OpenBitFun host. A session on that host may use an SSH workspace; +this does not make it an account-device session. When switching the bot to another +account device, the existing reply includes an explicit notice that proactive +scheduled-job and desktop-result notifications are unavailable there. Ordinary +WeChat-initiated requests keep their existing cross-device reply path. This feature +does not add cross-host Peer Device or Detached Dispatch result delivery. diff --git a/src/crates/assembly/core/AGENTS.md b/src/crates/assembly/core/AGENTS.md index c14ebb6507..f10a47194a 100644 --- a/src/crates/assembly/core/AGENTS.md +++ b/src/crates/assembly/core/AGENTS.md @@ -279,6 +279,7 @@ IM bot reply routing, account-device observation, and interaction delivery: ```bash cargo test --locked -p openbitfun-core --no-default-features --features remote-connect --lib service::remote_connect::bot:: +cargo test --locked -p openbitfun-core --no-default-features --features remote-connect --lib service::remote_connect::bot::weixin::tests ``` Pages account publication and tool gates (including remote directory rejection): diff --git a/src/crates/assembly/core/src/service/remote_connect/bot/weixin.rs b/src/crates/assembly/core/src/service/remote_connect/bot/weixin.rs index 5b5ddca09d..c7360b9e87 100644 --- a/src/crates/assembly/core/src/service/remote_connect/bot/weixin.rs +++ b/src/crates/assembly/core/src/service/remote_connect/bot/weixin.rs @@ -6,7 +6,7 @@ use anyhow::{anyhow, Result}; use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; -use log::{error, info, warn}; +use log::{debug, error, info, warn}; use openbitfun_services_integrations::remote_connect::bot::weixin as weixin_provider; use openbitfun_services_integrations::remote_connect::bot::weixin::WeixinProviderClient; pub use openbitfun_services_integrations::remote_connect::bot::weixin::{ @@ -14,10 +14,10 @@ pub use openbitfun_services_integrations::remote_connect::bot::weixin::{ MAX_INBOUND_IMAGES, MAX_WEIXIN_FILE_BYTES, WEIXIN_SESSION_EXPIRED_ERRCODE, }; use serde_json::Value; -use std::collections::HashMap; -use std::sync::Arc; +use std::collections::{HashMap, VecDeque}; +use std::sync::{Arc, Weak}; use std::time::Duration; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, Notify, RwLock}; use super::command_router::{ complete_im_bot_pairing, current_bot_language, execute_forwarded_turn, handle_command, @@ -27,10 +27,189 @@ use super::command_router::{ use super::{ load_bot_persistence, update_bot_persistence, BotConfig, BotRuntimeFence, SavedBotConnection, }; +use crate::agentic::coordination::get_global_coordinator; +use crate::agentic::events::{AgenticEvent, EventSubscriber}; use crate::service::remote_connect::remote_server::ImageAttachment; +use openbitfun_agent_runtime::event_bus::EventSubscriberResult; const LONG_POLL_TIMEOUT_SECS: u64 = 36; +/// Maximum proactive messages held for one peer before they are merged into a +/// single reply. Older entries are dropped so a peer that never comes back +/// cannot grow the backlog without bound. +const MAX_PENDING_OUTBOUND_PER_PEER: usize = 20; + +/// Bound retries to one channel activation window. This is a best-effort, +/// in-memory backlog, not a durable delivery receipt. +const PENDING_OUTBOUND_TTL_SECS: i64 = 24 * 60 * 60; + +/// Separator between merged turn outputs, so one reply built from several +/// answers still reads as several answers. +const PENDING_SEPARATOR: &str = "\n\n---\n\n"; + +/// How often the outbound worker retries a backlog that no new event woke it +/// for, so a token refreshed by an inbound message is picked up even if the +/// wakeup notification races the enqueue. +const OUTBOUND_RETRY_INTERVAL_SECS: u64 = 5; + +/// Bound on the remembered bot-owned turn ids. The set only needs to cover +/// turns that can still complete, so a small window is enough. +const MAX_TRACKED_OWN_TURNS: usize = 256; + +/// Largest payload one proactive push may merge into. One channel reply is the +/// unit the quota counts, so capping the merge at a single part avoids spending +/// several replies on one scheduled result. The provider enforces its quota. +const MAX_PROACTIVE_PUSH_BYTES: usize = weixin_provider::MAX_TEXT_CHUNK; + +/// One piece of assistant output waiting for a deliverable context_token. +#[derive(Debug, Clone)] +struct PendingOutbound { + id: String, + session_id: String, + text: String, + queued_at: i64, +} + +/// One peer's proactive backlog. +#[derive(Debug, Default)] +struct PendingPeer { + queue: VecDeque, + /// Reason the last delivery attempt held the backlog back, so a peer that + /// stays blocked logs once per distinct reason instead of on every retry. + deferred_reason: Option<&'static str>, +} + +fn pending_matches_binding(state: &BotChatState, item: &PendingOutbound) -> bool { + state.paired + && !state.account_remote_context + && state.active_remote_device.is_none() + && state.current_session_id.as_deref() == Some(item.session_id.as_str()) +} + +fn append_remote_push_notice(result: &mut HandleResult, language: super::BotLanguage) { + let notice = openbitfun_services_integrations::remote_connect::bot::strings_for(language) + .weixin_remote_push_unsupported; + result.reply.push_str("\n\n"); + result.reply.push_str(notice); + let footer = result.menu.footer_hint.get_or_insert_with(String::new); + if !footer.is_empty() { + footer.push_str("\n\n"); + } + footer.push_str(notice); +} + +/// Drops entries older than [`PENDING_OUTBOUND_TTL_SECS`]. +/// +/// Entries are appended in time order, so expired ones are always at the +/// front; returns how many were dropped so the caller can report them outside +/// the lock. +fn drop_expired_pending(queue: &mut VecDeque, now: i64) -> usize { + let mut dropped = 0; + while let Some(front) = queue.front() { + if now.saturating_sub(front.queued_at) <= PENDING_OUTBOUND_TTL_SECS { + break; + } + queue.pop_front(); + dropped += 1; + } + dropped +} + +/// Queues one proactive message, enforcing the retention window and the count +/// cap. Returns how many entries were dropped for logging. +fn push_pending(queue: &mut VecDeque, item: PendingOutbound, now: i64) -> usize { + let mut dropped = drop_expired_pending(queue, now); + while queue.len() >= MAX_PENDING_OUTBOUND_PER_PEER { + queue.pop_front(); + dropped += 1; + } + queue.push_back(item); + dropped +} + +fn push_tracked_turn(turns: &mut VecDeque, turn_id: &str) { + while turns.len() >= MAX_TRACKED_OWN_TURNS { + turns.pop_front(); + } + turns.push_back(turn_id.to_string()); +} + +/// Only the persistence fence identifies a completed answer that is safe to +/// read. The earlier DialogTurnCompleted event may precede the final write. +fn settled_session_turn(event: &AgenticEvent) -> Option<(&str, &str)> { + match event { + AgenticEvent::SessionHistoryChanged { + session_id, + settled_turn_id: Some(turn_id), + } => Some((session_id, turn_id)), + _ => None, + } +} + +fn settled_turn_text(mut poll: Value, turn_id: &str) -> Option { + // A newly created tracker can seed an empty active turn, or still contain + // a partial overlay until its own subscriber sees the persistence fence. + // Prefer the persisted record regardless of subscriber execution order. + poll["active_turn"] = Value::Null; + let turn = openbitfun_services_integrations::remote_connect::bot::remote_turn::observe_turn( + &poll, turn_id, + )?; + (matches!(turn.status.as_str(), "done" | "completed") && turn.error.is_none()) + .then_some(turn.text) +} + +fn bounded_proactive_text(text: &str) -> String { + let text = text.trim(); + if text.len() <= MAX_PROACTIVE_PUSH_BYTES { + return text.to_string(); + } + const TRUNCATED: &str = "\n…"; + format!( + "{}{TRUNCATED}", + crate::util::truncate_at_char_boundary(text, MAX_PROACTIVE_PUSH_BYTES - TRUNCATED.len()) + ) +} + +/// Joins a peer's backlog into the single reply that carries all of it. +/// +/// The quota counts replies, so a backlog must not be delivered one reply per +/// turn. The merged payload is capped at [`MAX_PROACTIVE_PUSH_BYTES`]: the +/// oldest outputs are dropped first, and when the newest output alone exceeds +/// the cap its head is sent, because the newest answer is the one the user is +/// waiting for. +fn merge_pending(pending: &[PendingOutbound]) -> String { + let mut kept: Vec<&str> = Vec::new(); + let mut bytes = 0usize; + // Walk newest first so the byte cap drops the oldest output. + for item in pending.iter().rev() { + let text = item.text.trim(); + if text.is_empty() { + continue; + } + let separator = if kept.is_empty() { + 0 + } else { + PENDING_SEPARATOR.len() + }; + if bytes + separator + text.len() > MAX_PROACTIVE_PUSH_BYTES { + break; + } + bytes += separator + text.len(); + kept.push(text); + } + if kept.is_empty() { + return pending + .iter() + .rev() + .map(|item| item.text.trim()) + .find(|text| !text.is_empty()) + .map(bounded_proactive_text) + .unwrap_or_default(); + } + kept.reverse(); + kept.join(PENDING_SEPARATOR) +} + #[derive(Debug, Clone)] struct PendingPairing { created_at: i64, @@ -42,6 +221,17 @@ pub struct WeixinBot { chat_states: Arc>>, context_tokens: Arc>>, runtime_fence: BotRuntimeFence, + /// Turn ids this bot started from an inbound message. The inbound path + /// already answers those turns, so the proactive channel must skip them or + /// every inbound message would be delivered twice. + own_turns: Arc>>, + /// Proactive output per peer that has no deliverable context_token yet, + /// oldest first. + pending_outbound: Arc>>, + outbound_wakeup: Arc, + /// Identity of this instance's proactive session-output subscription, so a + /// replaced bot retires its own hook instead of the replacement's. + session_output_hook_id: String, } pub async fn weixin_qr_start(base_url_override: Option) -> Result { @@ -83,6 +273,10 @@ impl WeixinBot { chat_states: Arc::new(RwLock::new(HashMap::new())), context_tokens: Arc::new(RwLock::new(context_tokens)), runtime_fence, + own_turns: Arc::new(Mutex::new(VecDeque::new())), + pending_outbound: Arc::new(Mutex::new(HashMap::new())), + outbound_wakeup: Arc::new(Notify::new()), + session_output_hook_id: format!("weixin_session_output_{}", uuid::Uuid::new_v4()), } } @@ -174,6 +368,315 @@ impl WeixinBot { let mut tokens = self.context_tokens.write().await; tokens.insert(peer_id.to_string(), token); weixin_provider::save_context_tokens(&self.api.config().bot_account_id, &tokens); + drop(tokens); + // A fresh token is the only thing that makes a stalled proactive + // backlog deliverable, so retry it as soon as one arrives. + self.outbound_wakeup.notify_one(); + } + + /// Registers the proactive session-output hook for this bot instance. + /// + /// Turns started outside the inbound path — a scheduled job, the desktop + /// window, another controller — produce assistant output that the reply + /// path never sees. The hook forwards that output to the peer bound to the + /// turn's session instead of dropping it. + pub fn subscribe_session_output(self: &Arc) { + let Some(coordinator) = get_global_coordinator() else { + warn!("weixin: session output hook unavailable; proactive push disabled"); + return; + }; + coordinator.subscribe_internal( + self.session_output_hook_id.clone(), + WeixinSessionOutputSubscriber { + bot: Arc::downgrade(self), + }, + ); + } + + /// Retires this instance's proactive session-output hook. + pub fn unsubscribe_session_output(&self) { + if let Some(coordinator) = get_global_coordinator() { + coordinator.unsubscribe_internal(&self.session_output_hook_id); + } + } + + /// Starts the single worker that drains proactive backlogs. + /// + /// One worker per bot preserves order and keeps two triggers from + /// delivering the same queued message. + fn spawn_outbound_worker(self: &Arc, mut stop: tokio::sync::watch::Receiver) { + let bot = self.clone(); + let wakeup = self.outbound_wakeup.clone(); + tokio::spawn(async move { + loop { + if *stop.borrow() || !bot.runtime_fence.is_lifecycle_current() { + break; + } + tokio::select! { + _ = stop.changed() => break, + _ = wakeup.notified() => {} + _ = tokio::time::sleep(Duration::from_secs(OUTBOUND_RETRY_INTERVAL_SECS)) => {} + } + bot.flush_pending_outbound().await; + } + }); + } + + /// Delivers assistant output produced by a turn this bot did not start. + /// + /// This is the explicit entry point behind the session-output hook, so the + /// proactive path can also be driven directly when no coordinator event is + /// available. + pub async fn notify_session_output(&self, session_id: &str, text: &str) { + if !self.runtime_fence.is_lifecycle_current() { + return; + } + for peer_id in self.peers_for_session(session_id).await { + self.queue_proactive(&peer_id, session_id, text, &format!("session {session_id}")) + .await; + } + } + + async fn handle_session_turn_settled(&self, session_id: String, turn_id: String) { + if !self.runtime_fence.is_lifecycle_current() { + return; + } + if self.is_own_turn(&turn_id).await { + return; + } + let peers = self.peers_for_session(&session_id).await; + if peers.is_empty() { + return; + } + let text = self.read_turn_text(&session_id, &turn_id).await; + // The session read is an await point: a bot replaced meanwhile must not + // push, even though the turn itself is worth delivering. + if !self.runtime_fence.is_lifecycle_current() { + return; + } + for peer_id in peers { + self.queue_proactive(&peer_id, &session_id, &text, &format!("turn {turn_id}")) + .await; + } + } + + async fn queue_proactive(&self, peer_id: &str, session_id: &str, text: &str, source: &str) { + if text.trim().is_empty() { + debug!("weixin: proactive push skipped for peer {peer_id}: {source} has no text"); + return; + } + info!("weixin: queueing proactive push for peer {peer_id} from {source}"); + self.enqueue_outbound(peer_id, session_id, text).await; + self.outbound_wakeup.notify_one(); + } + + /// The peers whose bot chats are bound to `session_id`, if this bot owns it. + /// + /// A chat routed to another device executes the turn elsewhere, so its text + /// is unreadable here and must not be pushed from this host. + async fn peers_for_session(&self, session_id: &str) -> Vec { + let states = self.chat_states.read().await; + states + .iter() + .filter(|(_, state)| { + state.paired + && !state.account_remote_context + && state.active_remote_device.is_none() + && state.current_session_id.as_deref() == Some(session_id) + }) + .map(|(peer_id, _)| peer_id.clone()) + .collect() + } + + /// Reads the assistant text of `turn_id` through the replay-safe turn + /// projection the inbound reply path already relies on. + async fn read_turn_text(&self, session_id: &str, turn_id: &str) -> String { + use openbitfun_services_integrations::remote_connect::{ + handle_remote_poll_command, RemoteCommand, + }; + + let dispatcher = + crate::service::remote_connect::remote_server::get_or_init_global_dispatcher(); + let poll_host = + crate::service_agent_runtime::CoreServiceAgentRuntime::remote_poll_host(&dispatcher); + let poll = handle_remote_poll_command( + &poll_host, + &RemoteCommand::PollSession { + session_id: session_id.to_string(), + since_version: 0, + known_msg_count: 0, + known_model_catalog_version: None, + }, + ) + .await; + serde_json::to_value(poll) + .ok() + .and_then(|poll| settled_turn_text(poll, turn_id)) + .unwrap_or_default() + } + + async fn mark_own_turn(&self, turn_id: &str) { + let mut turns = self.own_turns.lock().await; + push_tracked_turn(&mut turns, turn_id); + } + + async fn is_own_turn(&self, turn_id: &str) -> bool { + let turns = self.own_turns.lock().await; + turns.iter().any(|tracked| tracked == turn_id) + } + + async fn enqueue_outbound(&self, peer_id: &str, session_id: &str, text: &str) { + let now = chrono::Utc::now().timestamp(); + let dropped = { + let mut pending = self.pending_outbound.lock().await; + let peer = pending.entry(peer_id.to_string()).or_default(); + push_pending( + &mut peer.queue, + PendingOutbound { + id: uuid::Uuid::new_v4().to_string(), + session_id: session_id.to_string(), + text: bounded_proactive_text(text), + queued_at: now, + }, + now, + ) + }; + if dropped > 0 { + warn!( + "weixin: proactive push backlog for peer {peer_id} exceeded its limit or retention window; dropped {dropped} oldest message(s)" + ); + } + } + + async fn flush_pending_outbound(&self) { + let peers: Vec = { + let pending = self.pending_outbound.lock().await; + pending.keys().cloned().collect() + }; + for peer_id in peers { + self.flush_pending_for_peer(&peer_id).await; + } + } + + /// Delivers one peer's backlog as a single reply. + /// + /// The channel counts every reply against a 24 hour quota that also answers + /// the user's own messages, so the merged backlog is capped at a single part. No additional daily quota is imposed. + async fn flush_pending_for_peer(&self, peer_id: &str) { + if !self.runtime_fence.is_lifecycle_current() { + return; + } + let expired = self.expire_pending(peer_id).await; + if expired > 0 { + warn!( + "weixin: dropped {expired} proactive message(s) for peer {peer_id} past the {PENDING_OUTBOUND_TTL_SECS}s retention window" + ); + } + let pending = self.pending_snapshot(peer_id).await; + if pending.is_empty() { + return; + } + let merged = merge_pending(&pending); + if merged.trim().is_empty() { + // Nothing deliverable in the backlog, so it is not worth a reply. + self.confirm_pending_sent(peer_id, &pending).await; + return; + } + // A command can switch sessions or devices while output is queued. + // Never deliver an old selection's backlog into the new context. + if !self.runtime_fence.is_lifecycle_current() + || !self.pending_binding_is_current(peer_id, &pending).await + { + self.confirm_pending_sent(peer_id, &pending).await; + return; + } + if let Err(err) = self.send_text(peer_id, &merged).await { + // Keep the backlog. A stale token is replaced by the next inbound + // message and `send_text` has already dropped it, so the retry has a + // chance to succeed; the retention window bounds it. + if self.mark_deferred(peer_id, "send_failed").await { + warn!( + "weixin: proactive push to peer {peer_id} deferred; waiting for a retry or fresh context_token: {err}" + ); + } + return; + } + info!( + "weixin: proactive push delivered to peer {peer_id} (one reply for {} queued message(s))", + pending.len() + ); + self.confirm_pending_sent(peer_id, &pending).await; + } + + /// Drops queued entries past the retention window and returns how many. + async fn expire_pending(&self, peer_id: &str) -> usize { + let now = chrono::Utc::now().timestamp(); + let mut pending = self.pending_outbound.lock().await; + let (dropped, now_empty) = match pending.get_mut(peer_id) { + Some(peer) => { + let dropped = drop_expired_pending(&mut peer.queue, now); + (dropped, peer.queue.is_empty()) + } + None => (0, false), + }; + if now_empty { + pending.remove(peer_id); + } + dropped + } + + /// Clones a peer's backlog so the send below runs without holding the lock. + async fn pending_snapshot(&self, peer_id: &str) -> Vec { + let states = self.chat_states.read().await; + let state = states.get(peer_id); + let mut pending = self.pending_outbound.lock().await; + let Some(peer) = pending.get_mut(peer_id) else { + return Vec::new(); + }; + peer.queue + .retain(|item| state.is_some_and(|state| pending_matches_binding(state, item))); + peer.queue.iter().cloned().collect() + } + + /// Removes the entries a successful send covered and clears the blocker + /// flag. Anything enqueued while the send was in flight stays queued. + async fn confirm_pending_sent(&self, peer_id: &str, sent: &[PendingOutbound]) { + let mut pending = self.pending_outbound.lock().await; + let now_empty = match pending.get_mut(peer_id) { + Some(peer) => { + peer.queue + .retain(|item| !sent.iter().any(|sent| sent.id == item.id)); + peer.deferred_reason = None; + peer.queue.is_empty() + } + None => false, + }; + if now_empty { + pending.remove(peer_id); + } + } + + /// Flags the backlog's current blocker. Returns whether this is the first + /// attempt blocked for that reason, so the caller logs once instead of on + /// every retry. + async fn mark_deferred(&self, peer_id: &str, reason: &'static str) -> bool { + let mut pending = self.pending_outbound.lock().await; + match pending.get_mut(peer_id) { + Some(peer) if peer.deferred_reason != Some(reason) => { + peer.deferred_reason = Some(reason); + true + } + _ => false, + } + } + + async fn pending_binding_is_current(&self, peer_id: &str, pending: &[PendingOutbound]) -> bool { + let states = self.chat_states.read().await; + states.get(peer_id).is_some_and(|state| { + pending + .iter() + .all(|item| pending_matches_binding(state, item)) + }) } pub async fn notify_start(&self) -> Result<()> { @@ -455,6 +958,8 @@ impl WeixinBot { pub async fn run_message_loop(self: Arc, stop_rx: tokio::sync::watch::Receiver) { info!("Weixin message loop started"); + self.subscribe_session_output(); + self.spawn_outbound_worker(stop_rx.clone()); let mut stop = stop_rx; let mut buf = weixin_provider::load_sync_buf(&self.api.config().bot_account_id); let mut long_poll_timeout = Duration::from_secs(LONG_POLL_TIMEOUT_SECS); @@ -553,6 +1058,7 @@ impl WeixinBot { }); } } + self.unsubscribe_session_output(); info!("Weixin message loop stopped"); } @@ -621,8 +1127,12 @@ impl WeixinBot { if self.runtime_fence.identity_epoch() != command_identity_epoch { return; } + let was_remote = state.account_remote_context || state.active_remote_device.is_some(); let command = parse_command(text); - let result = handle_command(state, command, images).await; + let mut result = handle_command(state, command, images).await; + if !was_remote && (state.account_remote_context || state.active_remote_device.is_some()) { + append_remote_push_notice(&mut result, language); + } self.runtime_fence.reconcile_states(&mut states); if let Some(state) = states.get(&peer_id) { self.persist_chat_state(&peer_id, state).await; @@ -638,6 +1148,9 @@ impl WeixinBot { self.send_handle_result(&peer_id, &result).await; if let Some(forward) = result.forward_to_session { + // The inbound path answers this turn itself; keep the proactive + // hook from delivering the same answer a second time. + self.mark_own_turn(&forward.turn_id).await; let output_session_id = forward.session_id.clone(); let output_remote_target = forward.remote_target.clone(); let output_identity_epoch = command_identity_epoch; @@ -757,6 +1270,34 @@ impl WeixinBot { } } +/// Forwards completed turns that the inbound reply path does not own. +/// +/// The bot is held weakly so a retired instance cannot keep pushing after its +/// lifecycle slot was replaced. +struct WeixinSessionOutputSubscriber { + bot: Weak, +} + +#[async_trait::async_trait] +impl EventSubscriber for WeixinSessionOutputSubscriber { + async fn on_event(&self, event: &AgenticEvent) -> EventSubscriberResult { + let Some((session_id, turn_id)) = settled_session_turn(event) else { + return Ok(()); + }; + let Some(bot) = self.bot.upgrade() else { + return Ok(()); + }; + let session_id = session_id.to_string(); + let turn_id = turn_id.to_string(); + // The router awaits every subscriber inline while this path reads the + // session and sends over the network, so the work must not run here. + tokio::spawn(async move { + bot.handle_session_turn_settled(session_id, turn_id).await; + }); + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -794,4 +1335,442 @@ mod tests { assert!(body.contains("[\u{5f15}\u{7528}:")); assert!(body.contains("reply")); } + + fn pending(text: &str, queued_at: i64) -> PendingOutbound { + PendingOutbound { + id: uuid::Uuid::new_v4().to_string(), + session_id: "session".to_string(), + text: text.to_string(), + queued_at, + } + } + + fn backlog(texts: &[&str]) -> Vec { + texts + .iter() + .enumerate() + .map(|(index, text)| pending(*text, 1_000 + index as i64)) + .collect() + } + + #[test] + fn proactive_push_waits_for_durable_completion_and_ignores_history_edits() { + let early = AgenticEvent::DialogTurnCompleted { + session_id: "session".into(), + turn_id: "turn".into(), + total_rounds: 1, + total_tools: 0, + duration_ms: 10, + partial_recovery_reason: None, + success: Some(true), + finish_reason: Some("complete".into()), + has_final_response: Some(true), + }; + assert!(settled_session_turn(&early).is_none()); + assert!(settled_session_turn(&AgenticEvent::SessionHistoryChanged { + session_id: "session".into(), + settled_turn_id: None, + }) + .is_none()); + assert_eq!( + settled_session_turn(&AgenticEvent::SessionHistoryChanged { + session_id: "session".into(), + settled_turn_id: Some("turn".into()), + }), + Some(("session", "turn")) + ); + } + + #[test] + fn settled_projection_ignores_empty_or_partial_active_overlay() { + for text in ["", "partial answer"] { + let poll = json!({ + "active_turn": {"turn_id":"turn", "status":"active", "text":text}, + "new_messages": [{"turn_id":"turn", "role":"assistant", "status":"done", "content":"final scheduled result"}] + }); + assert_eq!( + settled_turn_text(poll, "turn").as_deref(), + Some("final scheduled result") + ); + } + let poll = json!({ + "active_turn": {"turn_id":"next", "status":"active", "text":"another turn"}, + "message_snapshot": [{"id":"turn_assistant", "role":"assistant", "status":"done", "content":"legacy final result"}] + }); + assert_eq!( + settled_turn_text(poll, "turn").as_deref(), + Some("legacy final result") + ); + } + + #[test] + fn settled_projection_skips_failed_cancelled_and_unrelated_turns() { + for status in ["active", "failed", "cancelled"] { + let poll = json!({"new_messages": [{"turn_id":"turn", "role":"assistant", "status":status, "content":"partial"}]}); + assert!(settled_turn_text(poll, "turn").is_none()); + } + let poll = json!({"new_messages": [{"turn_id":"other", "role":"assistant", "status":"done", "content":"another answer"}]}); + assert!(settled_turn_text(poll, "turn").is_none()); + } + + #[tokio::test] + async fn bot_owned_settled_turn_keeps_the_original_reply_path() { + let bot = test_bot("http://127.0.0.1:1/".into()).await; + bot.mark_own_turn("inbound-turn").await; + bot.handle_session_turn_settled("session".into(), "inbound-turn".into()) + .await; + assert!(bot.pending_outbound.lock().await.is_empty()); + } + + #[test] + fn tracked_bot_turns_stay_within_the_window() { + let mut turns = VecDeque::new(); + for index in 0..(MAX_TRACKED_OWN_TURNS + 5) { + push_tracked_turn(&mut turns, &format!("turn_{index}")); + } + assert_eq!(turns.len(), MAX_TRACKED_OWN_TURNS); + let newest = format!("turn_{}", MAX_TRACKED_OWN_TURNS + 4); + assert!(turns.iter().any(|id| id == &newest)); + assert!(!turns.iter().any(|id| id == "turn_0")); + } + + #[test] + fn pending_backlog_drops_the_oldest_past_the_pre_merge_cap() { + // The count cap bounds what is held before merging; the merged payload + // has its own byte cap. + let mut queue = VecDeque::new(); + let now = 1_000_000; + let mut dropped = 0; + for index in 0..(MAX_PENDING_OUTBOUND_PER_PEER + 3) { + dropped += push_pending(&mut queue, pending(&format!("m{index}"), now), now); + } + assert_eq!(queue.len(), MAX_PENDING_OUTBOUND_PER_PEER); + assert_eq!(dropped, 3); + assert_eq!(queue.front().map(|item| item.text.as_str()), Some("m3")); + assert_eq!(queue.back().map(|item| item.text.as_str()), Some("m22")); + } + + #[test] + fn pending_backlog_expires_entries_past_the_retention_window() { + let mut queue = VecDeque::new(); + let now = 1_000_000; + push_pending( + &mut queue, + pending("stale", now - PENDING_OUTBOUND_TTL_SECS - 1), + now, + ); + push_pending(&mut queue, pending("fresh", now), now); + assert_eq!(queue.len(), 1); + assert_eq!(queue.front().map(|item| item.text.as_str()), Some("fresh")); + assert_eq!(drop_expired_pending(&mut queue, now), 0); + } + + #[test] + fn merged_backlog_becomes_one_reply_in_time_order() { + let merged = merge_pending(&backlog(&["first", "second", "third"])); + assert_eq!( + merged, + format!("first{PENDING_SEPARATOR}second{PENDING_SEPARATOR}third") + ); + // However many turns the backlog carries, it spends one reply. + assert_eq!(weixin_provider::weixin_reply_count(&merged), 1); + } + + #[test] + fn merged_backlog_skips_entries_with_no_text() { + let merged = merge_pending(&backlog(&["first", " ", "", "second"])); + assert_eq!(merged, format!("first{PENDING_SEPARATOR}second")); + } + + #[test] + fn merged_backlog_drops_the_oldest_to_fit_one_reply() { + // The two answers cannot share one reply, so the newest is kept. + let oldest = "o".repeat(MAX_PROACTIVE_PUSH_BYTES - 100); + let newest = "n".repeat(200); + let merged = merge_pending(&backlog(&[oldest.as_str(), newest.as_str()])); + assert_eq!(merged, newest); + assert_eq!(weixin_provider::weixin_reply_count(&merged), 1); + } + + #[test] + fn merged_backlog_truncates_a_single_oversized_answer() { + // The newest answer alone exceeds the cap, so its head is sent instead + // of nothing at all. + let huge = "a".repeat(MAX_PROACTIVE_PUSH_BYTES + 100); + let merged = merge_pending(&backlog(&[huge.as_str()])); + assert_eq!(merged.len(), MAX_PROACTIVE_PUSH_BYTES); + assert!(merged.ends_with("\n…")); + assert_eq!(weixin_provider::weixin_reply_count(&merged), 1); + } + + #[test] + fn merged_backlog_never_splits_a_character() { + // The byte cap lands mid-character, so the head must stay valid UTF-8 + // and still fit inside one reply. + let huge = "\u{5b57}".repeat(MAX_PROACTIVE_PUSH_BYTES); + let merged = merge_pending(&backlog(&[huge.as_str()])); + assert!(merged.len() <= MAX_PROACTIVE_PUSH_BYTES); + assert_eq!(weixin_provider::weixin_reply_count(&merged), 1); + } + + async fn test_bot(base_url: String) -> Arc { + let bot = Arc::new(WeixinBot::new(WeixinConfig { + ilink_token: "fixture-token".into(), + base_url, + bot_account_id: format!("proactive-test-{}", uuid::Uuid::new_v4()), + })); + let mut state = BotChatState::new("peer".into()); + state.paired = true; + state.current_session_id = Some("session".into()); + bot.chat_states.write().await.insert("peer".into(), state); + bot.context_tokens + .write() + .await + .insert("peer".into(), "context".into()); + bot + } + + // A loopback provider with an explicit response gate exercises enqueueing + // while the real async HTTP send is in flight, without timing assumptions. + async fn mock_outbound( + responses: Vec<&'static str>, + first_response: Option>, + ) -> ( + String, + tokio::sync::mpsc::UnboundedReceiver, + tokio::task::JoinHandle<()>, + ) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}/", listener.local_addr().unwrap()); + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let task = tokio::spawn(async move { + let mut first_response = first_response; + for response in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut chunk = [0; 4096]; + loop { + let count = stream.read(&mut chunk).await.unwrap(); + assert!(count > 0, "request ended before the body"); + request.extend_from_slice(&chunk[..count]); + if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&request[..end]); + let length: usize = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse().unwrap()) + }) + .unwrap(); + if request.len() >= end + 4 + length { + tx.send( + serde_json::from_slice(&request[end + 4..end + 4 + length]) + .unwrap(), + ) + .unwrap(); + break; + } + } + } + if let Some(gate) = first_response.take() { + gate.await.unwrap(); + } + stream.write_all(format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response.len(), response + ).as_bytes()).await.unwrap(); + } + }); + (base, rx, task) + } + + #[tokio::test] + async fn proactive_send_ack_preserves_new_output_when_queue_overflows() { + let (release, gate) = tokio::sync::oneshot::channel(); + let (base, mut requests, server) = mock_outbound(vec!["{}", "{}"], Some(gate)).await; + let bot = test_bot(base).await; + for i in 0..MAX_PENDING_OUTBOUND_PER_PEER { + bot.notify_session_output("session", &format!("old-{i}")) + .await; + } + let sender = bot.clone(); + let sending = tokio::spawn(async move { sender.flush_pending_outbound().await }); + let first = tokio::time::timeout(Duration::from_secs(5), requests.recv()) + .await + .unwrap() + .unwrap(); + assert!(first["msg"]["item_list"][0]["text_item"]["text"] + .as_str() + .unwrap() + .contains("old-19")); + bot.notify_session_output("session", "new-unsent").await; + release.send(()).unwrap(); + sending.await.unwrap(); + let remaining = bot.pending_snapshot("peer").await; + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].text, "new-unsent"); + bot.flush_pending_outbound().await; + let second = requests.recv().await.unwrap(); + assert_eq!( + second["msg"]["item_list"][0]["text_item"]["text"], + "new-unsent" + ); + assert!(bot.pending_snapshot("peer").await.is_empty()); + server.await.unwrap(); + } + + #[tokio::test] + async fn proactive_send_waits_for_context_and_retries_provider_failure() { + let (base, mut requests, server) = mock_outbound( + vec![r#"{"ret":-1,"errmsg":"temporary failure"}"#, "{}"], + None, + ) + .await; + let bot = test_bot(base).await; + bot.context_tokens.write().await.clear(); + bot.notify_session_output("session", "scheduled result") + .await; + bot.flush_pending_outbound().await; + assert!(requests.try_recv().is_err()); + assert_eq!(bot.pending_snapshot("peer").await.len(), 1); + bot.context_tokens + .write() + .await + .insert("peer".into(), "fresh-context".into()); + bot.flush_pending_outbound().await; + assert_eq!(bot.pending_snapshot("peer").await.len(), 1); + bot.flush_pending_outbound().await; + assert!(bot.pending_snapshot("peer").await.is_empty()); + for _ in 0..2 { + let request = requests.recv().await.unwrap(); + assert_eq!(request["msg"]["context_token"], "fresh-context"); + assert_eq!( + request["msg"]["item_list"][0]["text_item"]["text"], + "scheduled result" + ); + } + server.await.unwrap(); + } + + #[tokio::test] + async fn fourth_scheduled_result_is_not_blocked_by_an_artificial_daily_quota() { + let (base, mut requests, server) = mock_outbound(vec!["{}"; 4], None).await; + let bot = test_bot(base).await; + for i in 0..4 { + bot.notify_session_output("session", &format!("result-{i}")) + .await; + bot.flush_pending_outbound().await; + let request = requests.recv().await.unwrap(); + assert_eq!( + request["msg"]["item_list"][0]["text_item"]["text"], + format!("result-{i}") + ); + } + server.await.unwrap(); + } + + #[tokio::test] + async fn switching_session_discards_old_backlog_without_discarding_new_output() { + let bot = test_bot("http://127.0.0.1:1/".into()).await; + bot.notify_session_output("session", "old-session-output") + .await; + bot.chat_states + .write() + .await + .get_mut("peer") + .unwrap() + .current_session_id = Some("next".into()); + bot.notify_session_output("next", "new-session-output") + .await; + let pending = bot.pending_snapshot("peer").await; + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].text, "new-session-output"); + bot.chat_states + .write() + .await + .get_mut("peer") + .unwrap() + .account_remote_context = true; + assert!(bot.pending_snapshot("peer").await.is_empty()); + bot.notify_session_output("next", "must-not-fall-back-to-local") + .await; + assert!(bot.pending_snapshot("peer").await.is_empty()); + } + + #[tokio::test] + async fn pending_output_is_byte_bounded_and_survives_more_than_one_hour() { + let bot = test_bot("http://127.0.0.1:1/".into()).await; + bot.notify_session_output("session", &"\u{5b57}".repeat(10000)) + .await; + let pending = bot.pending_snapshot("peer").await; + assert!(pending[0].text.len() <= MAX_PROACTIVE_PUSH_BYTES); + let mut queue = VecDeque::from(pending); + let queued_at = queue[0].queued_at; + assert_eq!(drop_expired_pending(&mut queue, queued_at + 3601), 0); + assert_eq!( + drop_expired_pending(&mut queue, queued_at + PENDING_OUTBOUND_TTL_SECS + 1), + 1 + ); + } + + #[test] + fn remote_push_limitation_is_visible_in_both_reply_renderers() { + for language in [ + super::super::BotLanguage::EnUS, + super::super::BotLanguage::ZhCN, + super::super::BotLanguage::ZhTW, + ] { + let mut result = HandleResult { + reply: "Device selected".into(), + actions: Vec::new(), + forward_to_session: None, + menu: super::super::MenuView::plain("Device selected"), + }; + append_remote_push_notice(&mut result, language); + let notice = + openbitfun_services_integrations::remote_connect::bot::strings_for(language) + .weixin_remote_push_unsupported; + assert!(result.reply.contains(notice)); + assert!(result.menu.render_plain_text(language).contains(notice)); + } + } + #[tokio::test] + async fn retired_bot_does_not_enqueue_or_send_pending_output() { + let bot = test_bot("http://127.0.0.1:1/".into()).await; + bot.notify_session_output("session", "pending-before-retirement") + .await; + bot.runtime_fence.slot.advance(); + bot.notify_session_output("session", "must-not-enqueue") + .await; + bot.flush_pending_outbound().await; + let pending = bot.pending_outbound.lock().await; + let peer = pending.get("peer").unwrap(); + assert_eq!(peer.queue.len(), 1); + // A send attempt to the closed endpoint would have deferred this queue. + assert!(peer.deferred_reason.is_none()); + } + + #[tokio::test] + async fn every_peer_bound_to_the_completed_session_gets_queued_output() { + let bot = test_bot("http://127.0.0.1:1/".into()).await; + let mut state = BotChatState::new("second-peer".into()); + state.paired = true; + state.current_session_id = Some("session".into()); + bot.chat_states + .write() + .await + .insert("second-peer".into(), state); + bot.notify_session_output("session", "scheduled result") + .await; + assert_eq!( + bot.pending_snapshot("peer").await[0].text, + "scheduled result" + ); + assert_eq!( + bot.pending_snapshot("second-peer").await[0].text, + "scheduled result" + ); + } } diff --git a/src/crates/services/services-integrations/src/remote_connect/bot/locale.rs b/src/crates/services/services-integrations/src/remote_connect/bot/locale.rs index 2c92c4f7a2..906d266581 100644 --- a/src/crates/services/services-integrations/src/remote_connect/bot/locale.rs +++ b/src/crates/services/services-integrations/src/remote_connect/bot/locale.rs @@ -174,6 +174,7 @@ pub struct BotStrings { pub auto_push_intro_many_fmt: &'static str, pub auto_push_skip_too_large_fmt: &'static str, pub auto_push_failed_fmt: &'static str, + pub weixin_remote_push_unsupported: &'static str, // ── Multi-device control ───────────────────────────────────── pub devices_title: &'static str, @@ -336,6 +337,7 @@ const STRINGS_ZH: BotStrings = BotStrings { auto_push_intro_many_fmt: "正在为你发送 {n} 个文件……", auto_push_skip_too_large_fmt: "已跳过「{name}」：{size} 超过 {limit} 上限，请改用桌面端获取。", auto_push_failed_fmt: "发送「{name}」失败：{err}", + weixin_remote_push_unsupported: "微信暂不支持主动推送其他设备上的定时任务或桌面会话结果。你从微信发起的请求仍会正常回复。", devices_title: "多设备控制", devices_account_required: "所连接的桌面端尚未使用 GitHub 登录，无法使用多设备控制。请在桌面端的账号登录对话框中登录，机器人会自动继承账号身份。", @@ -497,6 +499,7 @@ const STRINGS_ZH_TW: BotStrings = BotStrings { auto_push_intro_many_fmt: "正在為你發送 {n} 個文件……", auto_push_skip_too_large_fmt: "已跳過「{name}」：{size} 超過 {limit} 上限，請改用桌面端獲取。", auto_push_failed_fmt: "發送「{name}」失敗：{err}", + weixin_remote_push_unsupported: "微信暫不支援主動推送其他裝置上的排程任務或桌面會話結果。你從微信發起的請求仍會正常回覆。", devices_title: "多裝置控制", devices_account_required: "所連接的桌面端尚未使用 GitHub 登入，無法使用多裝置控制。請在桌面端的帳號登入對話框中登入，機器人會自動繼承帳號身份。", @@ -659,6 +662,7 @@ Open Remote Connect in OpenBitFun Desktop and send the 6-digit pairing code here auto_push_intro_many_fmt: "Sending {n} files for you…", auto_push_skip_too_large_fmt: "Skipping \"{name}\": {size} exceeds the {limit} limit. Please grab it from OpenBitFun Desktop instead.", auto_push_failed_fmt: "Failed to send \"{name}\": {err}", + weixin_remote_push_unsupported: "WeChat cannot proactively push scheduled-job or desktop-session results from another device yet. Requests you send from WeChat still receive replies.", devices_title: "Multi-device Control", devices_account_required: "The paired desktop is not logged into a GitHub account, so multi-device control is unavailable. Log in via the desktop's Account Login dialog and the bot will inherit the account identity.", diff --git a/src/crates/services/services-integrations/src/remote_connect/bot/weixin.rs b/src/crates/services/services-integrations/src/remote_connect/bot/weixin.rs index 5ee960ae9e..f9a0156f97 100644 --- a/src/crates/services/services-integrations/src/remote_connect/bot/weixin.rs +++ b/src/crates/services/services-integrations/src/remote_connect/bot/weixin.rs @@ -30,7 +30,11 @@ const QR_POLL_TIMEOUT_SECS: u64 = 35; const QR_SESSION_TTL_MS: i64 = 5 * 60_000; pub const WEIXIN_SESSION_EXPIRED_ERRCODE: i64 = -14; const SESSION_PAUSE_SECS: u64 = 3600; -const MAX_TEXT_CHUNK: usize = 4000; +/// Largest text payload the channel accepts in one reply. +/// +/// A longer reply is split, and the channel counts every part against its +/// reply quota, so callers that budget replies need this size. +pub const MAX_TEXT_CHUNK: usize = 4000; const MAX_QR_REFRESH: u32 = 3; const DEFAULT_CDN_BASE_URL: &str = "https://novac2c.cdn.weixin.qq.com/c2c"; pub const MAX_WEIXIN_FILE_BYTES: u64 = 30 * 1024 * 1024; @@ -1660,6 +1664,16 @@ fn chunk_text_for_weixin(text: &str) -> Vec { out } +/// Number of replies the channel spends to deliver `text`. +/// +/// The channel counts every split part of a long reply against its reply +/// quota, so a caller that budgets replies must predict this before sending. +/// Defined on top of [`chunk_text_for_weixin`] so the count cannot drift from +/// what `send_text_chunks` actually sends. +pub fn weixin_reply_count(text: &str) -> usize { + chunk_text_for_weixin(text).len() +} + fn is_weixin_media_item_type(type_id: i64) -> bool { matches!(type_id, 2..=5) } @@ -2008,6 +2022,25 @@ mod tests { assert_eq!(chunks[1], "a"); } + #[test] + fn reply_count_follows_the_part_boundary() { + assert_eq!(weixin_reply_count(&"a".repeat(MAX_TEXT_CHUNK)), 1); + assert_eq!(weixin_reply_count(&"a".repeat(MAX_TEXT_CHUNK + 1)), 2); + assert_eq!(weixin_reply_count(&"a".repeat(MAX_TEXT_CHUNK * 2)), 2); + assert_eq!(weixin_reply_count(&"a".repeat(MAX_TEXT_CHUNK * 2 + 1)), 3); + // Empty text still spends one reply, so a caller that budgets replies + // must refuse to send it. + assert_eq!(weixin_reply_count(""), 1); + } + + #[test] + fn reply_count_measures_bytes_not_characters() { + // The splitter cuts on bytes, so a CJK body spends more replies than + // its character count suggests. + assert_eq!(weixin_reply_count(&"\u{5b57}".repeat(1333)), 1); + assert_eq!(weixin_reply_count(&"\u{5b57}".repeat(1334)), 2); + } + #[test] fn accepts_bounded_server_long_poll_timeout() { let fallback = Duration::from_secs(36);