From 44c62d3919a7a7335dcd9d80fb8fc4816e7b69cc Mon Sep 17 00:00:00 2001 From: TunaTung <103393065+TunaTung@users.noreply.github.com> Date: Sun, 20 Sep 2026 23:59:24 +0800 Subject: [PATCH 1/3] feat(remote-connect): push weixin bot output from non-inbound turns The Weixin iLink bot only ever answered messages it received: the sole send path was the inbound forward (handle_incoming_message -> execute_forwarded_turn -> send_text). Assistant output produced by a turn the bot did not start - a scheduled job, the desktop window, another controller - was never delivered, so a scheduled job in the peer's bound session could run to completion while the user's phone stayed silent. Subscribe the bot to AgenticEvent::DialogTurnCompleted and deliver the turn's text to the peer bound to that session. A turn is skipped when this bot started it, because the inbound path already answered it and pushing it again would deliver every inbound message twice. Failed turns are skipped too; whether there is anything to send is decided by the turn's own text, so a degraded turn that still carries an explanation is not lost. The text is read through the existing replay-safe observe_turn projection, so bot delivery agrees with what the other remote surfaces show. Delivery respects the channel reply quota. The OpenBitFun product copy for this channel records that WeChat ClawBot allows the bot at most 10 replies in the 24 hours after the user sends a message, and that a split long reply spends one reply per part. That quota is also what answers the user's own messages, so this path: - merges a peer's whole backlog into a single reply instead of spending one reply per turn, keeps the turns in time order, and separates them readably; - caps the merged payload at one channel part, dropping the oldest output first and truncating the newest on a character boundary only when it alone exceeds the cap; - keeps its own share of 3 replies per rolling 24 hour window, which it can never overspend, leaving the rest of the channel quota untouched; - counts the replies a push will actually spend - the parts the channel splits it into - and refuses to send when the share cannot cover them, holding the backlog until the window rolls instead; - still queues when no context_token is available and keeps the backlog when a send fails on a stale token, so output waits for the next inbound message rather than being dropped. The backlog is bounded per peer, by age, and by merged payload size. One worker per bot drains it, which preserves order and keeps two triggers from delivering the same queued message. The subscriber spawns its work instead of running it inline: the event router awaits every subscriber while this path reads the session and sends over the network. is_lifecycle_current() is checked on entry, after the session read, and in the worker loop, so a replaced bot cannot push, and the subscription id is unique per instance so a retired bot unsubscribes only its own hook. Reuses send_text / send_text_chunks and the existing context_tokens cache; no protocol change. weixin_reply_count() is added next to chunk_text_for_weixin so the predicted reply count cannot drift from what send_text_chunks sends. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/service/remote_connect/bot/weixin.rs | 757 +++++++++++++++++- .../src/remote_connect/bot/weixin.rs | 35 +- 2 files changed, 787 insertions(+), 5 deletions(-) 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..6258d41bd8 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,203 @@ 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; + +/// Maximum age of a queued proactive message. Anything older is dropped +/// instead of replayed: the answer it carries is stale by then, and a restart +/// must not push a batch of expired turns. +const PENDING_OUTBOUND_TTL_SECS: i64 = 3600; + +/// 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; + +/// Reply quota of the WeChat ClawBot channel over one activation window, as +/// the OpenBitFun product copy for this channel records it (`botWeixinRestriction`): +/// after the user sends a message the bot may send at most 10 replies in the +/// following 24 hours, a split long reply spends one reply per part, and the +/// window restarts when the user sends another message. +/// +/// This is a product-level channel restriction, not a field the wire protocol +/// documents — `Tencent/openclaw-weixin` only specifies passing the inbound +/// `context_token` back on a reply. +const REPLY_QUOTA_WINDOW_SECS: i64 = 24 * 60 * 60; +const REPLY_QUOTA_PER_WINDOW: usize = 10; + +/// Replies this path may spend inside one quota window. +/// +/// The quota is shared with the replies the user's own messages receive, and +/// this path cannot observe what the inbound reply path already spent, so it +/// keeps a small share and leaves the rest alone. A proactive answer is never +/// urgent, and three unprompted replies in one window already exceeds what a +/// user expects from a channel they did not just talk to. +const MAX_PROACTIVE_REPLIES_PER_WINDOW: usize = 3; + +/// 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 keeps one push +/// from spending the whole share at once. +const MAX_PROACTIVE_PUSH_BYTES: usize = weixin_provider::MAX_TEXT_CHUNK; + +/// The proactive share must stay well clear of the channel quota, so the +/// inbound reply path always keeps room to answer the user. +const _: () = assert!(MAX_PROACTIVE_REPLIES_PER_WINDOW < REPLY_QUOTA_PER_WINDOW); + +/// One piece of assistant output waiting for a deliverable context_token. +#[derive(Debug, Clone)] +struct PendingOutbound { + 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>, +} + +/// 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()); +} + +/// A completed turn is worth pushing when this bot did not start it (the +/// inbound path already answered it) and it did not end in failure. Whether +/// there is anything to send is decided by the turn's own text, which keeps a +/// degraded turn that still carries an explanation from being dropped. +fn proactive_push_eligible(own_turn: bool, success: Option) -> bool { + !own_turn && success != Some(false) +} + +/// Whether `planned_replies` more replies still fit the proactive share. +/// +/// The unit is one channel reply, so a message the channel splits into two +/// parts asks for two. The share is never allowed to overspend, which is what +/// keeps the rest of the channel quota available for inbound replies. +fn proactive_reply_budget_allows(used_replies: usize, planned_replies: usize) -> bool { + used_replies + planned_replies <= MAX_PROACTIVE_REPLIES_PER_WINDOW +} + +/// Drops reply spend that has left the quota window. +/// +/// The window rolls with each reply instead of resetting on a fixed boundary, +/// so this path can never spend the whole share in one burst at the end of a +/// window. Spend is recorded in order, so expired entries are always at the +/// front. +fn drop_expired_replies(spent: &mut VecDeque, now: i64) -> usize { + let mut dropped = 0; + while let Some(front) = spent.front() { + if now.saturating_sub(*front) < REPLY_QUOTA_WINDOW_SECS { + break; + } + spent.pop_front(); + dropped += 1; + } + dropped +} + +/// Records `replies` replies spent at `now`. The unit is one channel reply, so +/// a message the channel splits into two parts records two. +fn record_replies_spent(spent: &mut VecDeque, replies: usize, now: i64) { + for _ in 0..replies { + spent.push_back(now); + } +} + +/// 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(|text| { + crate::util::truncate_at_char_boundary(text, MAX_PROACTIVE_PUSH_BYTES).to_string() + }) + .unwrap_or_default(); + } + kept.reverse(); + kept.join(PENDING_SEPARATOR) +} + #[derive(Debug, Clone)] struct PendingPairing { created_at: i64, @@ -42,6 +235,20 @@ 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>>, + /// Timestamps of the channel replies this path already spent, one entry + /// per reply including the parts the channel split it into. + proactive_replies: 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 +290,11 @@ 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())), + proactive_replies: Arc::new(Mutex::new(VecDeque::new())), + outbound_wakeup: Arc::new(Notify::new()), + session_output_hook_id: format!("weixin_session_output_{}", uuid::Uuid::new_v4()), } } @@ -174,6 +386,326 @@ 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; + } + let Some(peer_id) = self.peer_for_session(session_id).await else { + return; + }; + self.queue_proactive(&peer_id, text, &format!("session {session_id}")) + .await; + } + + async fn handle_session_turn_completed( + &self, + session_id: String, + turn_id: String, + success: Option, + ) { + if !self.runtime_fence.is_lifecycle_current() { + return; + } + if !proactive_push_eligible(self.is_own_turn(&turn_id).await, success) { + return; + } + let Some(peer_id) = self.peer_for_session(&session_id).await else { + 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; + } + self.queue_proactive(&peer_id, &text, &format!("turn {turn_id}")) + .await; + } + + async fn queue_proactive(&self, peer_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, text).await; + self.outbound_wakeup.notify_one(); + } + + /// The peer whose bot chat is 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 peer_for_session(&self, session_id: &str) -> Option { + let states = self.chat_states.read().await; + states + .iter() + .find(|(_, 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()) + } + + /// 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::bot::remote_turn::observe_turn; + 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| observe_turn(&poll, turn_id)) + .map(|turn| turn.text) + .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, 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 { + text: text.to_string(), + 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 backlog is merged and the reply count the + /// merge will spend is checked before anything is sent. + 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.len()).await; + return; + } + let planned_replies = weixin_provider::weixin_reply_count(&merged); + let now = chrono::Utc::now().timestamp(); + if !self.proactive_budget_allows(now, planned_replies).await { + if self + .mark_deferred(peer_id, "proactive_reply_share_spent") + .await + { + warn!( + "weixin: proactive push to peer {peer_id} waiting; the proactive share of {MAX_PROACTIVE_REPLIES_PER_WINDOW} of the channel's {REPLY_QUOTA_PER_WINDOW} replies per {}h window is spent and this push needs {planned_replies}", + REPLY_QUOTA_WINDOW_SECS / 3600 + ); + } + 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, "no_context_token").await { + warn!( + "weixin: proactive push to peer {peer_id} deferred until a fresh context_token arrives: {err}" + ); + } + return; + } + self.record_spent_replies(now, planned_replies).await; + info!( + "weixin: proactive push delivered to peer {peer_id} ({planned_replies} reply(ies) for {} queued message(s))", + pending.len() + ); + self.confirm_pending_sent(peer_id, pending.len()).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 pending = self.pending_outbound.lock().await; + pending + .get(peer_id) + .map(|peer| peer.queue.iter().cloned().collect()) + .unwrap_or_default() + } + + /// 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: usize) { + let mut pending = self.pending_outbound.lock().await; + let now_empty = match pending.get_mut(peer_id) { + Some(peer) => { + for _ in 0..sent { + peer.queue.pop_front(); + } + 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, + } + } + + /// Whether `planned_replies` more replies fit the proactive share, after + /// retiring spend that has left the quota window. + async fn proactive_budget_allows(&self, now: i64, planned_replies: usize) -> bool { + let mut spent = self.proactive_replies.lock().await; + drop_expired_replies(&mut spent, now); + proactive_reply_budget_allows(spent.len(), planned_replies) + } + + async fn record_spent_replies(&self, now: i64, replies: usize) { + let mut spent = self.proactive_replies.lock().await; + record_replies_spent(&mut spent, replies, now); } pub async fn notify_start(&self) -> Result<()> { @@ -455,6 +987,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 +1087,7 @@ impl WeixinBot { }); } } + self.unsubscribe_session_output(); info!("Weixin message loop stopped"); } @@ -638,6 +1173,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 +1295,42 @@ 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 AgenticEvent::DialogTurnCompleted { + session_id, + turn_id, + success, + .. + } = event + else { + return Ok(()); + }; + let Some(bot) = self.bot.upgrade() else { + return Ok(()); + }; + let session_id = session_id.clone(); + let turn_id = turn_id.clone(); + let success = *success; + // 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_completed(session_id, turn_id, success) + .await; + }); + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -794,4 +1368,179 @@ mod tests { assert!(body.contains("[\u{5f15}\u{7528}:")); assert!(body.contains("reply")); } + + fn pending(text: &str, queued_at: i64) -> PendingOutbound { + PendingOutbound { + 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_skips_bot_owned_and_failed_turns() { + // A turn this bot started already got its reply on the inbound path, so + // pushing it again would deliver every inbound message twice. + assert!(!proactive_push_eligible(true, Some(true))); + assert!(proactive_push_eligible(false, Some(true))); + // A turn that ended in failure must not be pushed. + assert!(!proactive_push_eligible(false, Some(false))); + // Older runtimes omit `success`; the turn stays eligible and the + // turn's own text decides whether there is anything to send. + assert!(proactive_push_eligible(false, None)); + } + + #[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_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); + } + + #[test] + fn proactive_reply_budget_keeps_room_for_inbound_replies() { + // The share is a minority slice of the channel quota, so the replies + // the user's own messages receive still work. + assert!(MAX_PROACTIVE_REPLIES_PER_WINDOW * 2 < REPLY_QUOTA_PER_WINDOW); + // A push fits while the share has room for it. + assert!(proactive_reply_budget_allows(0, 1)); + assert!(proactive_reply_budget_allows( + MAX_PROACTIVE_REPLIES_PER_WINDOW - 1, + 1 + )); + // Once the share is spent, nothing more is sent. + assert!(!proactive_reply_budget_allows( + MAX_PROACTIVE_REPLIES_PER_WINDOW, + 1 + )); + // A push that needs more replies than the whole share is refused, which + // is what keeps one push from draining it. + assert!(!proactive_reply_budget_allows( + 0, + MAX_PROACTIVE_REPLIES_PER_WINDOW + 1 + )); + } + + #[test] + fn reply_budget_allows_a_push_that_exactly_spends_the_share() { + assert!(proactive_reply_budget_allows( + MAX_PROACTIVE_REPLIES_PER_WINDOW - 2, + 2 + )); + } + + #[test] + fn reply_spend_leaves_the_window_on_a_rolling_basis() { + let mut spent = VecDeque::new(); + let now = 1_000_000; + record_replies_spent(&mut spent, 2, now); + assert_eq!(spent.len(), 2); + assert!(!proactive_reply_budget_allows( + spent.len(), + MAX_PROACTIVE_REPLIES_PER_WINDOW - 1 + )); + + // Still inside the window: the spend stands, so no burst is possible + // at the end of a window. + drop_expired_replies(&mut spent, now + REPLY_QUOTA_WINDOW_SECS - 1); + assert_eq!(spent.len(), 2); + + // Past the window the spend is retired and the share is free again. + assert_eq!( + drop_expired_replies(&mut spent, now + REPLY_QUOTA_WINDOW_SECS), + 2 + ); + assert!(spent.is_empty()); + assert!(proactive_reply_budget_allows(spent.len(), 1)); + } } 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); From b0ffd19bf320dc16a2f2c99f246dc61ab1e927c1 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Tue, 22 Sep 2026 09:38:34 +0800 Subject: [PATCH 2/3] fix(weixin): preserve pending results and remove artificial push quota --- src/apps/desktop/README.md | 31 ++ src/crates/assembly/core/AGENTS.md | 1 + .../src/service/remote_connect/bot/weixin.rs | 523 ++++++++++++------ .../src/remote_connect/bot/locale.rs | 4 + 4 files changed, 384 insertions(+), 175 deletions(-) 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 6258d41bd8..cf130a250d 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 @@ -39,10 +39,9 @@ const LONG_POLL_TIMEOUT_SECS: u64 = 36; /// cannot grow the backlog without bound. const MAX_PENDING_OUTBOUND_PER_PEER: usize = 20; -/// Maximum age of a queued proactive message. Anything older is dropped -/// instead of replayed: the answer it carries is stale by then, and a restart -/// must not push a batch of expired turns. -const PENDING_OUTBOUND_TTL_SECS: i64 = 3600; +/// 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. @@ -57,39 +56,16 @@ const OUTBOUND_RETRY_INTERVAL_SECS: u64 = 5; /// turns that can still complete, so a small window is enough. const MAX_TRACKED_OWN_TURNS: usize = 256; -/// Reply quota of the WeChat ClawBot channel over one activation window, as -/// the OpenBitFun product copy for this channel records it (`botWeixinRestriction`): -/// after the user sends a message the bot may send at most 10 replies in the -/// following 24 hours, a split long reply spends one reply per part, and the -/// window restarts when the user sends another message. -/// -/// This is a product-level channel restriction, not a field the wire protocol -/// documents — `Tencent/openclaw-weixin` only specifies passing the inbound -/// `context_token` back on a reply. -const REPLY_QUOTA_WINDOW_SECS: i64 = 24 * 60 * 60; -const REPLY_QUOTA_PER_WINDOW: usize = 10; - -/// Replies this path may spend inside one quota window. -/// -/// The quota is shared with the replies the user's own messages receive, and -/// this path cannot observe what the inbound reply path already spent, so it -/// keeps a small share and leaves the rest alone. A proactive answer is never -/// urgent, and three unprompted replies in one window already exceeds what a -/// user expects from a channel they did not just talk to. -const MAX_PROACTIVE_REPLIES_PER_WINDOW: usize = 3; - /// 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 keeps one push -/// from spending the whole share at once. +/// 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; -/// The proactive share must stay well clear of the channel quota, so the -/// inbound reply path always keeps room to answer the user. -const _: () = assert!(MAX_PROACTIVE_REPLIES_PER_WINDOW < REPLY_QUOTA_PER_WINDOW); - /// 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, } @@ -103,6 +79,25 @@ struct PendingPeer { 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 @@ -147,39 +142,16 @@ fn proactive_push_eligible(own_turn: bool, success: Option) -> bool { !own_turn && success != Some(false) } -/// Whether `planned_replies` more replies still fit the proactive share. -/// -/// The unit is one channel reply, so a message the channel splits into two -/// parts asks for two. The share is never allowed to overspend, which is what -/// keeps the rest of the channel quota available for inbound replies. -fn proactive_reply_budget_allows(used_replies: usize, planned_replies: usize) -> bool { - used_replies + planned_replies <= MAX_PROACTIVE_REPLIES_PER_WINDOW -} - -/// Drops reply spend that has left the quota window. -/// -/// The window rolls with each reply instead of resetting on a fixed boundary, -/// so this path can never spend the whole share in one burst at the end of a -/// window. Spend is recorded in order, so expired entries are always at the -/// front. -fn drop_expired_replies(spent: &mut VecDeque, now: i64) -> usize { - let mut dropped = 0; - while let Some(front) = spent.front() { - if now.saturating_sub(*front) < REPLY_QUOTA_WINDOW_SECS { - break; - } - spent.pop_front(); - dropped += 1; - } - dropped -} - -/// Records `replies` replies spent at `now`. The unit is one channel reply, so -/// a message the channel splits into two parts records two. -fn record_replies_spent(spent: &mut VecDeque, replies: usize, now: i64) { - for _ in 0..replies { - spent.push_back(now); +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. @@ -215,9 +187,7 @@ fn merge_pending(pending: &[PendingOutbound]) -> String { .rev() .map(|item| item.text.trim()) .find(|text| !text.is_empty()) - .map(|text| { - crate::util::truncate_at_char_boundary(text, MAX_PROACTIVE_PUSH_BYTES).to_string() - }) + .map(bounded_proactive_text) .unwrap_or_default(); } kept.reverse(); @@ -242,9 +212,6 @@ pub struct WeixinBot { /// Proactive output per peer that has no deliverable context_token yet, /// oldest first. pending_outbound: Arc>>, - /// Timestamps of the channel replies this path already spent, one entry - /// per reply including the parts the channel split it into. - proactive_replies: 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. @@ -292,7 +259,6 @@ impl WeixinBot { runtime_fence, own_turns: Arc::new(Mutex::new(VecDeque::new())), pending_outbound: Arc::new(Mutex::new(HashMap::new())), - proactive_replies: Arc::new(Mutex::new(VecDeque::new())), outbound_wakeup: Arc::new(Notify::new()), session_output_hook_id: format!("weixin_session_output_{}", uuid::Uuid::new_v4()), } @@ -449,11 +415,10 @@ impl WeixinBot { if !self.runtime_fence.is_lifecycle_current() { return; } - let Some(peer_id) = self.peer_for_session(session_id).await else { - return; - }; - self.queue_proactive(&peer_id, text, &format!("session {session_id}")) - .await; + 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_completed( @@ -468,44 +433,48 @@ impl WeixinBot { if !proactive_push_eligible(self.is_own_turn(&turn_id).await, success) { return; } - let Some(peer_id) = self.peer_for_session(&session_id).await else { + 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; } - self.queue_proactive(&peer_id, &text, &format!("turn {turn_id}")) - .await; + 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, text: &str, source: &str) { + 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, text).await; + self.enqueue_outbound(peer_id, session_id, text).await; self.outbound_wakeup.notify_one(); } - /// The peer whose bot chat is bound to `session_id`, if this bot owns it. + /// 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 peer_for_session(&self, session_id: &str) -> Option { + async fn peers_for_session(&self, session_id: &str) -> Vec { let states = self.chat_states.read().await; states .iter() - .find(|(_, state)| { + .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 @@ -547,7 +516,7 @@ impl WeixinBot { turns.iter().any(|tracked| tracked == turn_id) } - async fn enqueue_outbound(&self, peer_id: &str, text: &str) { + 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; @@ -555,7 +524,9 @@ impl WeixinBot { push_pending( &mut peer.queue, PendingOutbound { - text: text.to_string(), + id: uuid::Uuid::new_v4().to_string(), + session_id: session_id.to_string(), + text: bounded_proactive_text(text), queued_at: now, }, now, @@ -581,8 +552,7 @@ impl WeixinBot { /// 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 backlog is merged and the reply count the - /// merge will spend is checked before anything is sent. + /// 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; @@ -600,40 +570,33 @@ impl WeixinBot { 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.len()).await; + self.confirm_pending_sent(peer_id, &pending).await; return; } - let planned_replies = weixin_provider::weixin_reply_count(&merged); - let now = chrono::Utc::now().timestamp(); - if !self.proactive_budget_allows(now, planned_replies).await { - if self - .mark_deferred(peer_id, "proactive_reply_share_spent") - .await - { - warn!( - "weixin: proactive push to peer {peer_id} waiting; the proactive share of {MAX_PROACTIVE_REPLIES_PER_WINDOW} of the channel's {REPLY_QUOTA_PER_WINDOW} replies per {}h window is spent and this push needs {planned_replies}", - REPLY_QUOTA_WINDOW_SECS / 3600 - ); - } + // 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, "no_context_token").await { + if self.mark_deferred(peer_id, "send_failed").await { warn!( - "weixin: proactive push to peer {peer_id} deferred until a fresh context_token arrives: {err}" + "weixin: proactive push to peer {peer_id} deferred; waiting for a retry or fresh context_token: {err}" ); } return; } - self.record_spent_replies(now, planned_replies).await; info!( - "weixin: proactive push delivered to peer {peer_id} ({planned_replies} reply(ies) for {} queued message(s))", + "weixin: proactive push delivered to peer {peer_id} (one reply for {} queued message(s))", pending.len() ); - self.confirm_pending_sent(peer_id, pending.len()).await; + self.confirm_pending_sent(peer_id, &pending).await; } /// Drops queued entries past the retention window and returns how many. @@ -655,22 +618,25 @@ impl WeixinBot { /// Clones a peer's backlog so the send below runs without holding the lock. async fn pending_snapshot(&self, peer_id: &str) -> Vec { - let pending = self.pending_outbound.lock().await; - pending - .get(peer_id) - .map(|peer| peer.queue.iter().cloned().collect()) - .unwrap_or_default() + 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: usize) { + 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) => { - for _ in 0..sent { - peer.queue.pop_front(); - } + peer.queue + .retain(|item| !sent.iter().any(|sent| sent.id == item.id)); peer.deferred_reason = None; peer.queue.is_empty() } @@ -695,17 +661,13 @@ impl WeixinBot { } } - /// Whether `planned_replies` more replies fit the proactive share, after - /// retiring spend that has left the quota window. - async fn proactive_budget_allows(&self, now: i64, planned_replies: usize) -> bool { - let mut spent = self.proactive_replies.lock().await; - drop_expired_replies(&mut spent, now); - proactive_reply_budget_allows(spent.len(), planned_replies) - } - - async fn record_spent_replies(&self, now: i64, replies: usize) { - let mut spent = self.proactive_replies.lock().await; - record_replies_spent(&mut spent, replies, now); + 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<()> { @@ -1156,8 +1118,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; @@ -1371,6 +1337,8 @@ mod tests { 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, } @@ -1474,6 +1442,7 @@ mod tests { 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); } @@ -1487,60 +1456,264 @@ mod tests { assert_eq!(weixin_provider::weixin_reply_count(&merged), 1); } - #[test] - fn proactive_reply_budget_keeps_room_for_inbound_replies() { - // The share is a minority slice of the channel quota, so the replies - // the user's own messages receive still work. - assert!(MAX_PROACTIVE_REPLIES_PER_WINDOW * 2 < REPLY_QUOTA_PER_WINDOW); - // A push fits while the share has room for it. - assert!(proactive_reply_budget_allows(0, 1)); - assert!(proactive_reply_budget_allows( - MAX_PROACTIVE_REPLIES_PER_WINDOW - 1, - 1 - )); - // Once the share is spent, nothing more is sent. - assert!(!proactive_reply_budget_allows( - MAX_PROACTIVE_REPLIES_PER_WINDOW, - 1 - )); - // A push that needs more replies than the whole share is refused, which - // is what keeps one push from draining it. - assert!(!proactive_reply_budget_allows( - 0, - MAX_PROACTIVE_REPLIES_PER_WINDOW + 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) } - #[test] - fn reply_budget_allows_a_push_that_exactly_spends_the_share() { - assert!(proactive_reply_budget_allows( - MAX_PROACTIVE_REPLIES_PER_WINDOW - 2, - 2 - )); + #[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 reply_spend_leaves_the_window_on_a_rolling_basis() { - let mut spent = VecDeque::new(); - let now = 1_000_000; - record_replies_spent(&mut spent, 2, now); - assert_eq!(spent.len(), 2); - assert!(!proactive_reply_budget_allows( - spent.len(), - MAX_PROACTIVE_REPLIES_PER_WINDOW - 1 - )); - - // Still inside the window: the spend stands, so no burst is possible - // at the end of a window. - drop_expired_replies(&mut spent, now + REPLY_QUOTA_WINDOW_SECS - 1); - assert_eq!(spent.len(), 2); - - // Past the window the spend is retired and the share is free again. + 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!( - drop_expired_replies(&mut spent, now + REPLY_QUOTA_WINDOW_SECS), - 2 + bot.pending_snapshot("second-peer").await[0].text, + "scheduled result" ); - assert!(spent.is_empty()); - assert!(proactive_reply_budget_allows(spent.len(), 1)); } } 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.", From 59dfb818aee320a677045aab7ac129aa82379bf6 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Tue, 22 Sep 2026 09:48:00 +0800 Subject: [PATCH 3/3] fix(weixin): wait for persisted turn before proactive delivery --- .../src/service/remote_connect/bot/weixin.rs | 133 +++++++++++++----- 1 file changed, 95 insertions(+), 38 deletions(-) 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 cf130a250d..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 @@ -134,12 +134,28 @@ fn push_tracked_turn(turns: &mut VecDeque, turn_id: &str) { turns.push_back(turn_id.to_string()); } -/// A completed turn is worth pushing when this bot did not start it (the -/// inbound path already answered it) and it did not end in failure. Whether -/// there is anything to send is decided by the turn's own text, which keeps a -/// degraded turn that still carries an explanation from being dropped. -fn proactive_push_eligible(own_turn: bool, success: Option) -> bool { - !own_turn && success != Some(false) +/// 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 { @@ -421,16 +437,11 @@ impl WeixinBot { } } - async fn handle_session_turn_completed( - &self, - session_id: String, - turn_id: String, - success: Option, - ) { + async fn handle_session_turn_settled(&self, session_id: String, turn_id: String) { if !self.runtime_fence.is_lifecycle_current() { return; } - if !proactive_push_eligible(self.is_own_turn(&turn_id).await, success) { + if self.is_own_turn(&turn_id).await { return; } let peers = self.peers_for_session(&session_id).await; @@ -480,7 +491,6 @@ impl WeixinBot { /// 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::bot::remote_turn::observe_turn; use openbitfun_services_integrations::remote_connect::{ handle_remote_poll_command, RemoteCommand, }; @@ -501,8 +511,7 @@ impl WeixinBot { .await; serde_json::to_value(poll) .ok() - .and_then(|poll| observe_turn(&poll, turn_id)) - .map(|turn| turn.text) + .and_then(|poll| settled_turn_text(poll, turn_id)) .unwrap_or_default() } @@ -1272,26 +1281,18 @@ struct WeixinSessionOutputSubscriber { #[async_trait::async_trait] impl EventSubscriber for WeixinSessionOutputSubscriber { async fn on_event(&self, event: &AgenticEvent) -> EventSubscriberResult { - let AgenticEvent::DialogTurnCompleted { - session_id, - turn_id, - success, - .. - } = event - else { + 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.clone(); - let turn_id = turn_id.clone(); - let success = *success; + 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_completed(session_id, turn_id, success) - .await; + bot.handle_session_turn_settled(session_id, turn_id).await; }); Ok(()) } @@ -1353,16 +1354,72 @@ mod tests { } #[test] - fn proactive_push_skips_bot_owned_and_failed_turns() { - // A turn this bot started already got its reply on the inbound path, so - // pushing it again would deliver every inbound message twice. - assert!(!proactive_push_eligible(true, Some(true))); - assert!(proactive_push_eligible(false, Some(true))); - // A turn that ended in failure must not be pushed. - assert!(!proactive_push_eligible(false, Some(false))); - // Older runtimes omit `success`; the turn stays eligible and the - // turn's own text decides whether there is anything to send. - assert!(proactive_push_eligible(false, None)); + 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]