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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
//! Slack Integration
//!
//! Runs a Slack bot via Socket Mode alongside the TUI, forwarding messages from
//! allowlisted users to the AgentService and replying with responses.
mod agent;
pub(crate) mod blocks;
pub(crate) mod follow_up_question;
pub(crate) mod handler;
pub(crate) mod reactions;
pub(crate) mod suggest_followups;
pub(crate) mod tool_group;
pub(crate) mod upload;
pub use agent::SlackAgent;
use slack_morphism::prelude::SlackHyperClient;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, oneshot};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
/// One pending `follow_up_question` on Slack: oneshot half + options.
type PendingSlackQuestion = (oneshot::Sender<String>, Vec<String>);
/// Shared Slack state for proactive messaging.
///
/// Set when the bot connects via Socket Mode.
/// Read by the `slack_send` tool to send messages on demand.
pub struct SlackState {
client: Mutex<Option<Arc<SlackHyperClient>>>,
bot_token: Mutex<Option<String>>,
/// Channel ID of the owner's last message — used as default for proactive sends
owner_channel_id: Mutex<Option<String>>,
/// Maps session_id → channel_id for approval routing
session_channels: Mutex<HashMap<Uuid, String>>,
/// Pending approval channels: approval_id → oneshot sender of (approved, always)
pending_approvals: Mutex<HashMap<String, oneshot::Sender<(bool, bool)>>>,
/// Pending follow-up questions: question_id → (oneshot sender,
/// options). Same shape as the other channels — action_id only
/// carries the option index, the click handler maps it back via
/// the stored options list.
pending_questions: Mutex<HashMap<String, PendingSlackQuestion>>,
/// Per-session OPTIONAL follow-up suggestions from `suggest_followups`
/// (#599). Non-blocking: buttons ride under the response and a tap injects
/// the chosen suggestion as a new turn. Keyed by session; the tap handler
/// resolves `idx -> text`. Cleared on tap or when the user sends anything.
pending_followups: Mutex<HashMap<Uuid, Vec<String>>>,
/// Per-session cancel tokens for aborting in-flight agent tasks via /stop
cancel_tokens: Mutex<HashMap<Uuid, CancellationToken>>,
/// Collapsible tool groups keyed by their message ts, so the Expand /
/// Collapse interaction can re-render long after the turn ended.
/// Insertion-ordered for pruning; bounded at [`Self::TOOL_GROUP_CAP`].
tool_groups: Mutex<(Vec<String>, HashMap<String, tool_group::GroupState>)>,
}
impl Default for SlackState {
fn default() -> Self {
Self::new()
}
}
impl SlackState {
pub fn new() -> Self {
Self {
client: Mutex::new(None),
bot_token: Mutex::new(None),
owner_channel_id: Mutex::new(None),
session_channels: Mutex::new(HashMap::new()),
pending_approvals: Mutex::new(HashMap::new()),
pending_questions: Mutex::new(HashMap::new()),
pending_followups: Mutex::new(HashMap::new()),
cancel_tokens: Mutex::new(HashMap::new()),
tool_groups: Mutex::new((Vec::new(), HashMap::new())),
}
}
/// Retained tool groups; older ones stop being toggleable (their last
/// rendered state stays on screen, like Telegram's frozen blocks).
const TOOL_GROUP_CAP: usize = 20;
/// Insert or update the group for a message ts, PRESERVING the user's
/// expanded/collapsed choice on updates (a completing tool must not
/// snap an expanded group shut). Prunes the oldest beyond the cap.
/// Returns the stored state so callers render exactly what is kept.
pub(crate) async fn upsert_tool_group(
&self,
ts: String,
mut group: tool_group::GroupState,
) -> tool_group::GroupState {
let mut guard = self.tool_groups.lock().await;
let (order, map) = &mut *guard;
match map.get(&ts) {
Some(existing) => group.expanded = existing.expanded,
None => {
order.push(ts.clone());
while order.len() > Self::TOOL_GROUP_CAP {
let oldest = order.remove(0);
map.remove(&oldest);
}
}
}
map.insert(ts, group.clone());
group
}
/// Flip a group's expanded state; returns the new state for re-render,
/// or None when the group aged out of retention.
pub(crate) async fn toggle_tool_group(&self, ts: &str) -> Option<tool_group::GroupState> {
let mut guard = self.tool_groups.lock().await;
let (_, map) = &mut *guard;
let group = map.get_mut(ts)?;
group.expanded = !group.expanded;
Some(group.clone())
}
/// Register a pending `follow_up_question`. The action-block click
/// handler resolves by option index.
pub async fn register_pending_question(
&self,
id: String,
tx: oneshot::Sender<String>,
options: Vec<String>,
) {
self.pending_questions
.lock()
.await
.insert(id, (tx, options));
}
/// Resolve a pending question by option index. Returns the chosen
/// option string if the question + index are both valid.
pub async fn resolve_pending_question(&self, id: &str, idx: usize) -> Option<String> {
let (tx, options) = self.pending_questions.lock().await.remove(id)?;
let answer = options.get(idx)?.clone();
let _ = tx.send(answer.clone());
Some(answer)
}
/// Stash this session's optional follow-up suggestions (#599).
pub async fn set_pending_followups(&self, session_id: Uuid, options: Vec<String>) {
self.pending_followups
.lock()
.await
.insert(session_id, options);
}
/// Take a tapped follow-up suggestion by index, consuming the whole set.
pub async fn take_pending_followup(&self, session_id: Uuid, idx: usize) -> Option<String> {
let options = self.pending_followups.lock().await.remove(&session_id)?;
options.get(idx).cloned()
}
/// Drop this session's pending follow-up suggestions (user sent their own).
pub async fn clear_pending_followups(&self, session_id: Uuid) {
self.pending_followups.lock().await.remove(&session_id);
}
/// Store the connected client, bot token, and optionally the owner's channel.
pub async fn set_connected(
&self,
client: Arc<SlackHyperClient>,
bot_token: String,
channel_id: Option<String>,
) {
*self.client.lock().await = Some(client);
*self.bot_token.lock().await = Some(bot_token);
if let Some(id) = channel_id {
*self.owner_channel_id.lock().await = Some(id);
}
}
/// Update the owner's channel ID (called on each owner message).
pub async fn set_owner_channel(&self, channel_id: String) {
*self.owner_channel_id.lock().await = Some(channel_id);
}
/// Get a clone of the connected client, if any.
pub async fn client(&self) -> Option<Arc<SlackHyperClient>> {
self.client.lock().await.clone()
}
/// Get the bot token for opening API sessions.
pub async fn bot_token(&self) -> Option<String> {
self.bot_token.lock().await.clone()
}
/// Get the owner's last channel ID for proactive messaging.
pub async fn owner_channel_id(&self) -> Option<String> {
self.owner_channel_id.lock().await.clone()
}
/// Check if Slack is currently connected.
pub async fn is_connected(&self) -> bool {
self.client.lock().await.is_some()
}
/// Record which channel_id corresponds to a given session.
pub async fn register_session_channel(&self, session_id: Uuid, channel_id: String) {
self.session_channels
.lock()
.await
.insert(session_id, channel_id);
}
/// Look up the channel_id for a session.
pub async fn session_channel(&self, session_id: Uuid) -> Option<String> {
self.session_channels.lock().await.get(&session_id).cloned()
}
/// Register a pending approval oneshot channel.
pub async fn register_pending_approval(&self, id: String, tx: oneshot::Sender<(bool, bool)>) {
self.pending_approvals.lock().await.insert(id, tx);
}
/// Resolve a pending approval. Returns true if one existed.
pub async fn resolve_pending_approval(&self, id: &str, approved: bool, always: bool) -> bool {
if let Some(tx) = self.pending_approvals.lock().await.remove(id) {
let _ = tx.send((approved, always));
true
} else {
false
}
}
/// Store a cancel token for a session (before starting agent call).
/// If a token already exists for this session, cancel it first to abort the
/// previous in-flight agent call — prevents concurrent uncancellable agents.
pub async fn store_cancel_token(&self, session_id: Uuid, token: CancellationToken) {
let mut tokens = self.cancel_tokens.lock().await;
if let Some(old) = tokens.remove(&session_id) {
tracing::warn!(
"Slack: cancelling previous in-flight agent call for session {}",
session_id
);
old.cancel();
}
tokens.insert(session_id, token);
}
/// Cancel and remove the token for a session. Returns true if a token existed.
pub async fn cancel_session(&self, session_id: Uuid) -> bool {
if let Some(token) = self.cancel_tokens.lock().await.remove(&session_id) {
token.cancel();
true
} else {
false
}
}
/// Remove the cancel token after the agent call completes (cleanup).
/// Only removes if the stored token is already cancelled — prevents a
/// finishing old call from removing a newer call's live token.
pub async fn remove_cancel_token(&self, session_id: Uuid) {
let mut tokens = self.cancel_tokens.lock().await;
if let Some(token) = tokens.get(&session_id)
&& token.is_cancelled()
{
tokens.remove(&session_id);
}
}
}