1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
//! Background-task resume producer for WhatsApp (#731).
//!
//! Mirrors Telegram's `build_enqueue_callback`: when a detached long command
//! finishes, resume the originating session and send the result to its chat.
//! The session→JID map is populated per turn in the handler (`register_session_jid`).
use super::WhatsAppState;
use crate::brain::agent::service::MessageEnqueueCallback;
use crate::channels::bg_resume::{self, AgentHolder};
use std::sync::Arc;
pub(crate) fn build_enqueue_callback(
state: Arc<WhatsAppState>,
agent_holder: AgentHolder,
wa_cfg: crate::config::types::WhatsAppConfig,
) -> MessageEnqueueCallback {
Arc::new(move |session_id, msg| {
let state = state.clone();
let agent_holder = agent_holder.clone();
let wa_cfg = wa_cfg.clone();
tokio::spawn(async move {
let Some(jid_str) = state.session_jid(session_id).await else {
tracing::warn!(
"[bg-resume] whatsapp: no chat jid for session {session_id}; dropping"
);
return;
};
// #1989: the in-turn tail loop stops when the task is marked
// finished, which happens BEFORE this resume turn runs, so the
// chat sat silent through the slowest message of the whole
// cycle. Composing every 5 s across the turn and the delivery,
// first ping immediately; ticks without a client are skipped.
let typing = tokio::spawn({
let state = state.clone();
let jid_str = jid_str.clone();
async move {
if let Err(e) = send_composing_now(&state, &jid_str).await {
tracing::debug!(error = %e, "whatsapp resume: first composing skipped");
}
loop {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
if let Err(e) = send_composing_now(&state, &jid_str).await {
tracing::debug!(error = %e, "whatsapp resume: composing skipped");
}
}
}
});
let resume_flow = async {
let Some(agent) = bg_resume::upgrade(&agent_holder) else {
tracing::warn!("[bg-resume] whatsapp: agent gone; dropping resume");
return;
};
let Some(content) = bg_resume::run_resume_turn(
agent,
session_id,
msg.context_text,
"whatsapp",
&jid_str,
)
.await
else {
return;
};
// Bounded wait rather than a drop (#1242). Worse here than on
// the surfaces that check first: the turn above has already run,
// so returning threw away a completed answer AND the provider
// call that produced it.
let Some(client) = crate::channels::transport_ready::await_transport(
"whatsapp",
session_id,
|| state.client(),
)
.await
else {
return;
};
let Ok(jid) = jid_str.parse::<wacore_binary::jid::Jid>() else {
tracing::warn!("[bg-resume] whatsapp: bad jid '{jid_str}'; dropping delivery");
return;
};
// #1407: bg-resume results are agent-output sends: gate them
// through the shared limiter. Over-budget sends park in the
// FIFO queue (the drainer flushes and persists them as the
// rolling 24h window slides); owner-bound resumes bypass.
let rl_owner = wa_cfg.is_owner(jid_str.split('@').next().unwrap_or(&jid_str));
match state
.rate_limiter
.gate(&wa_cfg.rate_limit, &jid_str, &content, rl_owner)
.await
{
super::rate_limit::GateOutcome::Queued { .. } => {
tracing::info!(
"[bg-resume] whatsapp: daily cap reached; result queued for drainer flush"
);
return;
}
super::rate_limit::GateOutcome::SendNow => {}
}
let out = waproto::whatsapp::Message {
conversation: Some(content),
..Default::default()
};
if let Err(e) = client.send_message(jid, out).await {
tracing::warn!("[bg-resume] whatsapp: send_message failed: {e}");
}
};
resume_flow.await;
// Reached on every exit of the flow, including its early returns:
// the indicator lives exactly as long as the resume work does.
typing.abort();
});
})
}
/// One composing ping for the resume ticker: no-ops quietly when the
/// transport is down (`state.client()` is None until it reconnects).
async fn send_composing_now(state: &WhatsAppState, jid_str: &str) -> Result<(), String> {
let client = state.client().await.ok_or("whatsapp transport not ready")?;
let jid = jid_str
.parse::<wacore_binary::jid::Jid>()
.map_err(|e| e.to_string())?;
client
.chatstate()
.send_composing(&jid)
.await
.map_err(|e| e.to_string())
}