Vakyartha documentation

1//! Slack surface adapter (docs/design/34-channel-onboarding.md Phase 3),
2//! same contract as `telegram.rs`/`discord.rs`.
3//!
4//! **Transport choice.** Socket Mode (the option that needs no public
5//! URL) is a websocket, and the workspace has no websocket dependency
6//! today; the Events API alternative needs an internet-reachable webhook
7//! endpoint, which a laptop bridge does not have. This bridge therefore
8//! polls `conversations.history` for an explicitly configured set of
9//! channel ids and replies with `chat.postMessage` — no new dependency, no
10//! public URL, and the same "watch a few channels" shape as the Discord
11//! bridge. Socket Mode is the follow-up for real-time delivery and
12//! interactive Block Kit buttons.
13//!
14//! Launched via `vak slack --server URL --token GATEWAY_TOKEN` with
15//! `SLACK_BOT_TOKEN` in the environment (the credential store included).
16
17use serde_json::Value;
18
19use crate::gateway::{InboundChannel, InboundRequest};
20
21const API_BASE_DEFAULT: &str = "https://slack.com/api";
22
23pub struct SlackBridge {
24 /// API base, e.g. `https://slack.com/api`. Overridable for tests via
25 /// `SLACK_API_BASE`.
26 pub api_base: String,
27 pub bot_token: String,
28 /// Channel/DM ids this bot watches (`SLACK_CHANNEL_IDS`).
29 pub channel_ids: Vec<String>,
30 pub gateway_url: String,
31 pub gateway_token: String,
32 pub poll_secs: u64,
33 /// See `TelegramBridge::bot_id`.
34 pub bot_id: Option<String>,
35}
36
37impl InboundChannel for SlackBridge {
38 fn surface(&self) -> &'static str {
39 "slack"
40 }
41}
42
43/// One routable Slack message. `ts` is Slack's own message timestamp,
44/// which doubles as its id and as the `oldest` cursor.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct SlackMessage {
47 pub ts: String,
48 pub channel_id: String,
49 pub user_id: String,
50 pub text: String,
51 pub audio_url: Option<String>,
52 pub audio_mime: Option<String>,
53}
54
55#[derive(Debug, Default)]
56struct GatewayReply {
57 chunks: Vec<String>,
58 session_id: Option<String>,
59}
60
61/// Parse a `conversations.history` response into routable messages,
62/// oldest first (Slack returns newest first). Bot messages and message
63/// subtypes (joins, edits, thread broadcasts) are dropped: only a real
64/// human message should open or continue a session.
65pub fn parse_history(channel_id: &str, body: &Value) -> Vec<SlackMessage> {
66 let mut out: Vec<SlackMessage> = body["messages"]
67 .as_array()
68 .map(|items| {
69 items
70 .iter()
71 .filter(|m| m["bot_id"].is_null() && m["subtype"].is_null())
72 .filter_map(|m| {
73 let text = m["text"].as_str().unwrap_or_default().to_string();
74 let audio = m["files"].as_array().and_then(|items| {
75 items.iter().find(|f| {
76 f["mimetype"]
77 .as_str()
78 .is_some_and(|mime| mime.starts_with("audio/"))
79 && f["url_private_download"].as_str().is_some()
80 })
81 });
82 if text.trim().is_empty() && audio.is_none() {
83 return None;
84 }
85 Some(SlackMessage {
86 ts: m["ts"].as_str()?.to_string(),
87 channel_id: channel_id.to_string(),
88 user_id: m["user"].as_str()?.to_string(),
89 text,
90 audio_url: audio
91 .and_then(|f| f["url_private_download"].as_str())
92 .map(str::to_string),
93 audio_mime: audio
94 .and_then(|f| f["mimetype"].as_str())
95 .map(str::to_string),
96 })
97 })
98 .collect()
99 })
100 .unwrap_or_default();
101 // Slack timestamps are "<seconds>.<microseconds>" — compare
102 // numerically so a shorter second-part never sorts wrong.
103 out.sort_by(|a, b| {
104 ts_value(&a.ts)
105 .partial_cmp(&ts_value(&b.ts))
106 .unwrap_or(std::cmp::Ordering::Equal)
107 });
108 out
109}
110
111fn ts_value(ts: &str) -> f64 {
112 ts.parse::<f64>().unwrap_or(0.0)
113}
114
115fn http() -> reqwest::Client {
116 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
117 CLIENT
118 .get_or_init(|| {
119 reqwest::Client::builder()
120 .timeout(std::time::Duration::from_secs(300))
121 .build()
122 .unwrap_or_default()
123 })
124 .clone()
125}
126
127impl SlackBridge {
128 pub fn from_env(
129 gateway_url: String,
130 gateway_token: String,
131 bot_token: String,
132 bot_id: Option<String>,
133 ) -> Self {
134 let channel_ids = vak_config::get_var("SLACK_CHANNEL_IDS")
135 .unwrap_or_default()
136 .split(',')
137 .map(|s| s.trim().to_string())
138 .filter(|s| !s.is_empty())
139 .collect();
140 SlackBridge {
141 api_base: vak_config::get_var("SLACK_API_BASE")
142 .unwrap_or_else(|| API_BASE_DEFAULT.to_string()),
143 bot_token,
144 channel_ids,
145 gateway_url: gateway_url.trim_end_matches('/').to_string(),
146 gateway_token,
147 poll_secs: 3,
148 bot_id,
149 }
150 }
151
152 pub async fn tick(
153 &self,
154 cursors: &mut std::collections::HashMap<String, String>,
155 ) -> Result<(), String> {
156 for channel_id in &self.channel_ids {
157 let messages = self.fetch(channel_id, cursors.get(channel_id)).await?;
158 for message in messages {
159 let cold_start = !cursors.contains_key(channel_id);
160 cursors.insert(channel_id.clone(), message.ts.clone());
161 if cold_start {
162 continue;
163 }
164 let reply = self.process(&message).await;
165 if let Err(e) = self.send_message(channel_id, &reply.chunks).await {
166 eprintln!("[slack] send to {channel_id} failed: {e}");
167 }
168 if message.audio_url.is_some()
169 && let Err(e) = self
170 .send_voice(
171 channel_id,
172 &reply.chunks.join("\n"),
173 reply.session_id.as_deref(),
174 )
175 .await
176 {
177 eprintln!("[slack] voice reply unavailable: {e}");
178 }
179 }
180 }
181 Ok(())
182 }
183
184 async fn fetch(
185 &self,
186 channel_id: &str,
187 oldest: Option<&String>,
188 ) -> Result<Vec<SlackMessage>, String> {
189 let mut query: Vec<(&str, String)> = vec![
190 ("channel", channel_id.to_string()),
191 (
192 "limit",
193 if oldest.is_some() { "25" } else { "1" }.to_string(),
194 ),
195 ];
196 if let Some(ts) = oldest {
197 query.push(("oldest", ts.clone()));
198 // `oldest` is inclusive by default; excluding it means the
199 // cursor message is never re-routed.
200 query.push(("inclusive", "false".to_string()));
201 }
202 let resp = http()
203 .get(format!("{}/conversations.history", self.api_base))
204 .bearer_auth(&self.bot_token)
205 .query(&query)
206 .send()
207 .await
208 .map_err(|e| format!("slack conversations.history: {e}"))?;
209 if !resp.status().is_success() {
210 return Err(format!(
211 "slack conversations.history returned {}",
212 resp.status()
213 ));
214 }
215 let body: Value = resp
216 .json()
217 .await
218 .map_err(|e| format!("slack conversations.history body: {e}"))?;
219 // Slack answers 200 with `{"ok": false, "error": ...}`; treating
220 // that as success would silently poll forever against a bad token.
221 if body["ok"].as_bool() != Some(true) {
222 return Err(format!(
223 "slack conversations.history not ok: {}",
224 body["error"].as_str().unwrap_or("?")
225 ));
226 }
227 Ok(parse_history(channel_id, &body))
228 }
229
230 async fn process(&self, message: &SlackMessage) -> GatewayReply {
231 let text = message.text.clone();
232 let mut attachments = Vec::new();
233 if let Some(url) = &message.audio_url
234 && let Ok(response) = http().get(url).bearer_auth(&self.bot_token).send().await
235 && response.status().is_success()
236 && let Ok(bytes) = response.bytes().await
237 && bytes.len() <= 16 * 1024 * 1024
238 {
239 use base64::Engine as _;
240 let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
241 attachments.push(serde_json::json!({
242 "data": encoded,
243 "mime": message.audio_mime.as_deref().unwrap_or("audio/ogg"),
244 "kind": "audio",
245 "filename": "voice",
246 }));
247 }
248 // 0c-03: real per-user chat/sender, never a fixed placeholder.
249 let req = match InboundRequest::new(
250 self,
251 message.channel_id.clone(),
252 message.user_id.clone(),
253 text,
254 ) {
255 Ok(req) => req
256 .with_attachments(attachments)
257 .waiting()
258 .with_bot_id(self.bot_id.clone()),
259 Err(e) => {
260 return GatewayReply {
261 chunks: vec![format!("(bridge refused to send: {e})")],
262 ..Default::default()
263 };
264 }
265 };
266 let res = http()
267 .post(format!("{}/gateway/inbound", self.gateway_url))
268 .bearer_auth(&self.gateway_token)
269 .json(&req)
270 .send()
271 .await;
272 match res {
273 Ok(r) if r.status().as_u16() == 202 => GatewayReply {
274 chunks: vec!["(queued: I'm still working on your previous message)".into()],
275 ..Default::default()
276 },
277 Ok(r) if r.status().is_success() => match r.json::<Value>().await {
278 Ok(v) => {
279 let session_id = v["session_id"].as_str().map(String::from);
280 let chunks = super::prepared_chunks(v, "slack")
281 .unwrap_or_else(|e| vec![format!("(delivery failed: {e})")]);
282 GatewayReply { chunks, session_id }
283 }
284 Err(e) => GatewayReply {
285 chunks: vec![format!("(bad gateway reply: {e})")],
286 ..Default::default()
287 },
288 },
289 Ok(r) => GatewayReply {
290 chunks: vec![format!("(gateway error: {})", r.status())],
291 ..Default::default()
292 },
293 Err(e) => GatewayReply {
294 chunks: vec![format!("(gateway unreachable: {e})")],
295 ..Default::default()
296 },
297 }
298 }
299
300 async fn send_message(&self, channel_id: &str, chunks: &[String]) -> Result<(), String> {
301 for chunk in chunks {
302 let resp = http()
303 .post(format!("{}/chat.postMessage", self.api_base))
304 .bearer_auth(&self.bot_token)
305 .json(&serde_json::json!({ "channel": channel_id, "text": chunk }))
306 .send()
307 .await
308 .map_err(|e| format!("slack chat.postMessage: {e}"))?;
309 if !resp.status().is_success() {
310 return Err(format!("slack chat.postMessage returned {}", resp.status()));
311 }
312 let body: Value = resp
313 .json()
314 .await
315 .map_err(|e| format!("slack chat.postMessage body: {e}"))?;
316 if body["ok"].as_bool() != Some(true) {
317 return Err(format!(
318 "slack chat.postMessage not ok: {}",
319 body["error"].as_str().unwrap_or("?")
320 ));
321 }
322 }
323 Ok(())
324 }
325
326 async fn send_voice(
327 &self,
328 channel_id: &str,
329 text: &str,
330 session_id: Option<&str>,
331 ) -> Result<(), String> {
332 let response = http()
333 .post(format!("{}/voice/speak", self.gateway_url))
334 .bearer_auth(&self.gateway_token)
335 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
336 .send()
337 .await
338 .map_err(|e| e.to_string())?;
339 if !response.status().is_success() {
340 return Err(format!("voice speak returned {}", response.status()));
341 }
342 let part = reqwest::multipart::Part::bytes(
343 response.bytes().await.map_err(|e| e.to_string())?.to_vec(),
344 )
345 .file_name("reply.wav")
346 .mime_str("audio/wav")
347 .map_err(|e| e.to_string())?;
348 let sent = http()
349 .post(format!("{}/files.uploadV2", self.api_base))
350 .bearer_auth(&self.bot_token)
351 .multipart(
352 reqwest::multipart::Form::new()
353 .text("channel_id", channel_id.to_string())
354 .part("file", part),
355 )
356 .send()
357 .await
358 .map_err(|e| e.to_string())?;
359 if !sent.status().is_success() {
360 return Err(format!("slack voice upload returned {}", sent.status()));
361 }
362 Ok(())
363 }
364
365 pub async fn run(&self) -> Result<(), String> {
366 if self.channel_ids.is_empty() {
367 return Err(
368 "no channels to watch — set SLACK_CHANNEL_IDS to a comma-separated \
369 list of Slack channel or DM ids the bot has been invited to"
370 .into(),
371 );
372 }
373 let mut cursors = std::collections::HashMap::new();
374 let mut failures: u32 = 0;
375 loop {
376 match self.tick(&mut cursors).await {
377 Ok(()) => {
378 failures = 0;
379 tokio::time::sleep(std::time::Duration::from_secs(self.poll_secs)).await;
380 }
381 Err(e) => {
382 failures += 1;
383 if failures == 1 || failures.is_multiple_of(10) {
384 eprintln!("[slack] poll failed ({failures} consecutive): {e}");
385 }
386 tokio::time::sleep(std::time::Duration::from_secs(
387 crate::surfaces::discord::backoff_secs(failures),
388 ))
389 .await;
390 }
391 }
392 }
393 }
394}
395
396#[cfg(test)]
397#[allow(clippy::unwrap_used, clippy::expect_used)]
398mod tests {
399 use super::*;
400
401 fn bridge() -> SlackBridge {
402 SlackBridge {
403 api_base: "http://localhost".into(),
404 bot_token: "t".into(),
405 channel_ids: vec!["C1".into()],
406 gateway_url: "http://localhost".into(),
407 gateway_token: "g".into(),
408 poll_secs: 1,
409 bot_id: None,
410 }
411 }
412
413 #[test]
414 fn surface_is_the_allowlist_key_prefix() {
415 assert_eq!(bridge().surface(), "slack");
416 }
417
418 #[test]
419 fn identity_maps_channel_to_chat_and_user_to_sender() {
420 let req = InboundRequest::new(&bridge(), "C1", "U9", "hi").unwrap();
421 assert_eq!(req.surface, "slack");
422 assert_eq!(req.chat, "C1");
423 assert_eq!(req.sender, "U9");
424 }
425
426 #[test]
427 fn empty_or_placeholder_identity_is_refused() {
428 assert!(InboundRequest::new(&bridge(), "", "U9", "hi").is_err());
429 assert!(InboundRequest::new(&bridge(), "C1", "", "hi").is_err());
430 assert!(InboundRequest::new(&bridge(), "slack", "U9", "hi").is_err());
431 assert!(InboundRequest::new(&bridge(), "C1", "slack", "hi").is_err());
432 }
433
434 #[test]
435 fn parses_oldest_first_and_drops_bot_and_subtype_messages() {
436 let body = serde_json::json!({
437 "ok": true,
438 "messages": [
439 { "ts": "1000.000200", "text": "second", "user": "U9" },
440 { "ts": "1000.000300", "text": "beep", "user": "U1", "bot_id": "B1" },
441 { "ts": "1000.000100", "text": "first", "user": "U9" },
442 { "ts": "1000.000400", "text": "joined", "user": "U9", "subtype": "channel_join" },
443 ]
444 });
445 let parsed = parse_history("C1", &body);
446 assert_eq!(parsed.len(), 2);
447 assert_eq!(parsed[0].text, "first");
448 assert_eq!(parsed[0].user_id, "U9");
449 assert_eq!(parsed[0].channel_id, "C1");
450 assert_eq!(parsed[1].text, "second");
451 }
452
453 #[test]
454 fn parses_audio_files_without_text_for_governed_transcription() {
455 let body = serde_json::json!({"messages":[{"ts":"1.1","text":"","user":"U9","files":[{"mimetype":"audio/ogg","url_private_download":"https://files.example/voice"}]}]});
456 let parsed = parse_history("C1", &body);
457 assert_eq!(parsed.len(), 1);
458 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/ogg"));
459 assert_eq!(
460 parsed[0].audio_url.as_deref(),
461 Some("https://files.example/voice")
462 );
463 }
464
465 #[test]
466 fn audio_file_without_download_url_is_not_routable_as_voice() {
467 let body = serde_json::json!({"messages":[{"ts":"1.2","text":"","user":"U9","files":[{"mimetype":"audio/ogg"}]}]});
468 let parsed = parse_history("C1", &body);
469 assert!(
470 parsed.is_empty(),
471 "unfetchable media must not enter the governed path"
472 );
473 }
474
475 #[test]
476 fn gateway_reply_session_id_is_preserved_for_playback() {
477 let value = serde_json::json!({"chunks":["ok"],"session_id":"slack-session"});
478 let reply = GatewayReply {
479 chunks: value["chunks"]
480 .as_array()
481 .unwrap()
482 .iter()
483 .filter_map(|v| v.as_str().map(String::from))
484 .collect(),
485 session_id: value["session_id"].as_str().map(String::from),
486 };
487 assert_eq!(reply.session_id.as_deref(), Some("slack-session"));
488 }
489
490 #[test]
491 fn timestamps_order_numerically_not_lexically() {
492 let body = serde_json::json!({
493 "ok": true,
494 "messages": [
495 { "ts": "1000.5", "text": "later", "user": "U9" },
496 { "ts": "999.9", "text": "earlier", "user": "U9" },
497 ]
498 });
499 let parsed = parse_history("C1", &body);
500 assert_eq!(parsed[0].text, "earlier");
501 assert_eq!(parsed[1].text, "later");
502 }
503}
504

1//! Slack surface adapter (docs/design/34-channel-onboarding.md Phase 3),
2//! same contract as `telegram.rs`/`discord.rs`.
3//!
4//! **Transport choice.** Socket Mode (the option that needs no public
5//! URL) is a websocket, and the workspace has no websocket dependency
6//! today; the Events API alternative needs an internet-reachable webhook
7//! endpoint, which a laptop bridge does not have. This bridge therefore
8//! polls `conversations.history` for an explicitly configured set of
9//! channel ids and replies with `chat.postMessage` — no new dependency, no
10//! public URL, and the same "watch a few channels" shape as the Discord
11//! bridge. Socket Mode is the follow-up for real-time delivery and
12//! interactive Block Kit buttons.
13//!
14//! Launched via `vak slack --server URL --token GATEWAY_TOKEN` with
15//! `SLACK_BOT_TOKEN` in the environment (the credential store included).
16
17use serde_json::Value;
18
19use crate::gateway::{InboundChannel, InboundRequest};
20
21const API_BASE_DEFAULT: &str = "https://slack.com/api";
22
23pub struct SlackBridge {
24 /// API base, e.g. `https://slack.com/api`. Overridable for tests via
25 /// `SLACK_API_BASE`.
26 pub api_base: String,
27 pub bot_token: String,
28 /// Channel/DM ids this bot watches (`SLACK_CHANNEL_IDS`).
29 pub channel_ids: Vec<String>,
30 pub gateway_url: String,
31 pub gateway_token: String,
32 pub poll_secs: u64,
33 /// See `TelegramBridge::bot_id`.
34 pub bot_id: Option<String>,
35}
36
37impl InboundChannel for SlackBridge {
38 fn surface(&self) -> &'static str {
39 "slack"
40 }
41}
42
43/// One routable Slack message. `ts` is Slack's own message timestamp,
44/// which doubles as its id and as the `oldest` cursor.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct SlackMessage {
47 pub ts: String,
48 pub channel_id: String,
49 pub user_id: String,
50 pub text: String,
51 pub audio_url: Option<String>,
52 pub audio_mime: Option<String>,
53}
54
55#[derive(Debug, Default)]
56struct GatewayReply {
57 chunks: Vec<String>,
58 session_id: Option<String>,
59}
60
61/// Parse a `conversations.history` response into routable messages,
62/// oldest first (Slack returns newest first). Bot messages and message
63/// subtypes (joins, edits, thread broadcasts) are dropped: only a real
64/// human message should open or continue a session.
65pub fn parse_history(channel_id: &str, body: &Value) -> Vec<SlackMessage> {
66 let mut out: Vec<SlackMessage> = body["messages"]
67 .as_array()
68 .map(|items| {
69 items
70 .iter()
71 .filter(|m| m["bot_id"].is_null() && m["subtype"].is_null())
72 .filter_map(|m| {
73 let text = m["text"].as_str().unwrap_or_default().to_string();
74 let audio = m["files"].as_array().and_then(|items| {
75 items.iter().find(|f| {
76 f["mimetype"]
77 .as_str()
78 .is_some_and(|mime| mime.starts_with("audio/"))
79 && f["url_private_download"].as_str().is_some()
80 })
81 });
82 if text.trim().is_empty() && audio.is_none() {
83 return None;
84 }
85 Some(SlackMessage {
86 ts: m["ts"].as_str()?.to_string(),
87 channel_id: channel_id.to_string(),
88 user_id: m["user"].as_str()?.to_string(),
89 text,
90 audio_url: audio
91 .and_then(|f| f["url_private_download"].as_str())
92 .map(str::to_string),
93 audio_mime: audio
94 .and_then(|f| f["mimetype"].as_str())
95 .map(str::to_string),
96 })
97 })
98 .collect()
99 })
100 .unwrap_or_default();
101 // Slack timestamps are "<seconds>.<microseconds>" — compare
102 // numerically so a shorter second-part never sorts wrong.
103 out.sort_by(|a, b| {
104 ts_value(&a.ts)
105 .partial_cmp(&ts_value(&b.ts))
106 .unwrap_or(std::cmp::Ordering::Equal)
107 });
108 out
109}
110
111fn ts_value(ts: &str) -> f64 {
112 ts.parse::<f64>().unwrap_or(0.0)
113}
114
115fn http() -> reqwest::Client {
116 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
117 CLIENT
118 .get_or_init(|| {
119 reqwest::Client::builder()
120 .timeout(std::time::Duration::from_secs(300))
121 .build()
122 .unwrap_or_default()
123 })
124 .clone()
125}
126
127impl SlackBridge {
128 pub fn from_env(
129 gateway_url: String,
130 gateway_token: String,
131 bot_token: String,
132 bot_id: Option<String>,
133 ) -> Self {
134 let channel_ids = vak_config::get_var("SLACK_CHANNEL_IDS")
135 .unwrap_or_default()
136 .split(',')
137 .map(|s| s.trim().to_string())
138 .filter(|s| !s.is_empty())
139 .collect();
140 SlackBridge {
141 api_base: vak_config::get_var("SLACK_API_BASE")
142 .unwrap_or_else(|| API_BASE_DEFAULT.to_string()),
143 bot_token,
144 channel_ids,
145 gateway_url: gateway_url.trim_end_matches('/').to_string(),
146 gateway_token,
147 poll_secs: 3,
148 bot_id,
149 }
150 }
151
152 pub async fn tick(
153 &self,
154 cursors: &mut std::collections::HashMap<String, String>,
155 ) -> Result<(), String> {
156 for channel_id in &self.channel_ids {
157 let messages = self.fetch(channel_id, cursors.get(channel_id)).await?;
158 for message in messages {
159 let cold_start = !cursors.contains_key(channel_id);
160 cursors.insert(channel_id.clone(), message.ts.clone());
161 if cold_start {
162 continue;
163 }
164 let reply = self.process(&message).await;
165 if let Err(e) = self.send_message(channel_id, &reply.chunks).await {
166 eprintln!("[slack] send to {channel_id} failed: {e}");
167 }
168 if message.audio_url.is_some()
169 && let Err(e) = self
170 .send_voice(
171 channel_id,
172 &reply.chunks.join("\n"),
173 reply.session_id.as_deref(),
174 )
175 .await
176 {
177 eprintln!("[slack] voice reply unavailable: {e}");
178 }
179 }
180 }
181 Ok(())
182 }
183
184 async fn fetch(
185 &self,
186 channel_id: &str,
187 oldest: Option<&String>,
188 ) -> Result<Vec<SlackMessage>, String> {
189 let mut query: Vec<(&str, String)> = vec![
190 ("channel", channel_id.to_string()),
191 (
192 "limit",
193 if oldest.is_some() { "25" } else { "1" }.to_string(),
194 ),
195 ];
196 if let Some(ts) = oldest {
197 query.push(("oldest", ts.clone()));
198 // `oldest` is inclusive by default; excluding it means the
199 // cursor message is never re-routed.
200 query.push(("inclusive", "false".to_string()));
201 }
202 let resp = http()
203 .get(format!("{}/conversations.history", self.api_base))
204 .bearer_auth(&self.bot_token)
205 .query(&query)
206 .send()
207 .await
208 .map_err(|e| format!("slack conversations.history: {e}"))?;
209 if !resp.status().is_success() {
210 return Err(format!(
211 "slack conversations.history returned {}",
212 resp.status()
213 ));
214 }
215 let body: Value = resp
216 .json()
217 .await
218 .map_err(|e| format!("slack conversations.history body: {e}"))?;
219 // Slack answers 200 with `{"ok": false, "error": ...}`; treating
220 // that as success would silently poll forever against a bad token.
221 if body["ok"].as_bool() != Some(true) {
222 return Err(format!(
223 "slack conversations.history not ok: {}",
224 body["error"].as_str().unwrap_or("?")
225 ));
226 }
227 Ok(parse_history(channel_id, &body))
228 }
229
230 async fn process(&self, message: &SlackMessage) -> GatewayReply {
231 let text = message.text.clone();
232 let mut attachments = Vec::new();
233 if let Some(url) = &message.audio_url
234 && let Ok(response) = http().get(url).bearer_auth(&self.bot_token).send().await
235 && response.status().is_success()
236 && let Ok(bytes) = response.bytes().await
237 && bytes.len() <= 16 * 1024 * 1024
238 {
239 use base64::Engine as _;
240 let encoded = base64::engine::general_purpose::STANDARD.encode(bytes);
241 attachments.push(serde_json::json!({
242 "data": encoded,
243 "mime": message.audio_mime.as_deref().unwrap_or("audio/ogg"),
244 "kind": "audio",
245 "filename": "voice",
246 }));
247 }
248 // 0c-03: real per-user chat/sender, never a fixed placeholder.
249 let req = match InboundRequest::new(
250 self,
251 message.channel_id.clone(),
252 message.user_id.clone(),
253 text,
254 ) {
255 Ok(req) => req
256 .with_attachments(attachments)
257 .waiting()
258 .with_bot_id(self.bot_id.clone()),
259 Err(e) => {
260 return GatewayReply {
261 chunks: vec![format!("(bridge refused to send: {e})")],
262 ..Default::default()
263 };
264 }
265 };
266 let res = http()
267 .post(format!("{}/gateway/inbound", self.gateway_url))
268 .bearer_auth(&self.gateway_token)
269 .json(&req)
270 .send()
271 .await;
272 match res {
273 Ok(r) if r.status().as_u16() == 202 => GatewayReply {
274 chunks: vec!["(queued: I'm still working on your previous message)".into()],
275 ..Default::default()
276 },
277 Ok(r) if r.status().is_success() => match r.json::<Value>().await {
278 Ok(v) => {
279 let session_id = v["session_id"].as_str().map(String::from);
280 let chunks = super::prepared_chunks(v, "slack")
281 .unwrap_or_else(|e| vec![format!("(delivery failed: {e})")]);
282 GatewayReply { chunks, session_id }
283 }
284 Err(e) => GatewayReply {
285 chunks: vec![format!("(bad gateway reply: {e})")],
286 ..Default::default()
287 },
288 },
289 Ok(r) => GatewayReply {
290 chunks: vec![format!("(gateway error: {})", r.status())],
291 ..Default::default()
292 },
293 Err(e) => GatewayReply {
294 chunks: vec![format!("(gateway unreachable: {e})")],
295 ..Default::default()
296 },
297 }
298 }
299
300 async fn send_message(&self, channel_id: &str, chunks: &[String]) -> Result<(), String> {
301 for chunk in chunks {
302 let resp = http()
303 .post(format!("{}/chat.postMessage", self.api_base))
304 .bearer_auth(&self.bot_token)
305 .json(&serde_json::json!({ "channel": channel_id, "text": chunk }))
306 .send()
307 .await
308 .map_err(|e| format!("slack chat.postMessage: {e}"))?;
309 if !resp.status().is_success() {
310 return Err(format!("slack chat.postMessage returned {}", resp.status()));
311 }
312 let body: Value = resp
313 .json()
314 .await
315 .map_err(|e| format!("slack chat.postMessage body: {e}"))?;
316 if body["ok"].as_bool() != Some(true) {
317 return Err(format!(
318 "slack chat.postMessage not ok: {}",
319 body["error"].as_str().unwrap_or("?")
320 ));
321 }
322 }
323 Ok(())
324 }
325
326 async fn send_voice(
327 &self,
328 channel_id: &str,
329 text: &str,
330 session_id: Option<&str>,
331 ) -> Result<(), String> {
332 let response = http()
333 .post(format!("{}/voice/speak", self.gateway_url))
334 .bearer_auth(&self.gateway_token)
335 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
336 .send()
337 .await
338 .map_err(|e| e.to_string())?;
339 if !response.status().is_success() {
340 return Err(format!("voice speak returned {}", response.status()));
341 }
342 let part = reqwest::multipart::Part::bytes(
343 response.bytes().await.map_err(|e| e.to_string())?.to_vec(),
344 )
345 .file_name("reply.wav")
346 .mime_str("audio/wav")
347 .map_err(|e| e.to_string())?;
348 let sent = http()
349 .post(format!("{}/files.uploadV2", self.api_base))
350 .bearer_auth(&self.bot_token)
351 .multipart(
352 reqwest::multipart::Form::new()
353 .text("channel_id", channel_id.to_string())
354 .part("file", part),
355 )
356 .send()
357 .await
358 .map_err(|e| e.to_string())?;
359 if !sent.status().is_success() {
360 return Err(format!("slack voice upload returned {}", sent.status()));
361 }
362 Ok(())
363 }
364
365 pub async fn run(&self) -> Result<(), String> {
366 if self.channel_ids.is_empty() {
367 return Err(
368 "no channels to watch — set SLACK_CHANNEL_IDS to a comma-separated \
369 list of Slack channel or DM ids the bot has been invited to"
370 .into(),
371 );
372 }
373 let mut cursors = std::collections::HashMap::new();
374 let mut failures: u32 = 0;
375 loop {
376 match self.tick(&mut cursors).await {
377 Ok(()) => {
378 failures = 0;
379 tokio::time::sleep(std::time::Duration::from_secs(self.poll_secs)).await;
380 }
381 Err(e) => {
382 failures += 1;
383 if failures == 1 || failures.is_multiple_of(10) {
384 eprintln!("[slack] poll failed ({failures} consecutive): {e}");
385 }
386 tokio::time::sleep(std::time::Duration::from_secs(
387 crate::surfaces::discord::backoff_secs(failures),
388 ))
389 .await;
390 }
391 }
392 }
393 }
394}
395
396#[cfg(test)]
397#[allow(clippy::unwrap_used, clippy::expect_used)]
398mod tests {
399 use super::*;
400
401 fn bridge() -> SlackBridge {
402 SlackBridge {
403 api_base: "http://localhost".into(),
404 bot_token: "t".into(),
405 channel_ids: vec!["C1".into()],
406 gateway_url: "http://localhost".into(),
407 gateway_token: "g".into(),
408 poll_secs: 1,
409 bot_id: None,
410 }
411 }
412
413 #[test]
414 fn surface_is_the_allowlist_key_prefix() {
415 assert_eq!(bridge().surface(), "slack");
416 }
417
418 #[test]
419 fn identity_maps_channel_to_chat_and_user_to_sender() {
420 let req = InboundRequest::new(&bridge(), "C1", "U9", "hi").unwrap();
421 assert_eq!(req.surface, "slack");
422 assert_eq!(req.chat, "C1");
423 assert_eq!(req.sender, "U9");
424 }
425
426 #[test]
427 fn empty_or_placeholder_identity_is_refused() {
428 assert!(InboundRequest::new(&bridge(), "", "U9", "hi").is_err());
429 assert!(InboundRequest::new(&bridge(), "C1", "", "hi").is_err());
430 assert!(InboundRequest::new(&bridge(), "slack", "U9", "hi").is_err());
431 assert!(InboundRequest::new(&bridge(), "C1", "slack", "hi").is_err());
432 }
433
434 #[test]
435 fn parses_oldest_first_and_drops_bot_and_subtype_messages() {
436 let body = serde_json::json!({
437 "ok": true,
438 "messages": [
439 { "ts": "1000.000200", "text": "second", "user": "U9" },
440 { "ts": "1000.000300", "text": "beep", "user": "U1", "bot_id": "B1" },
441 { "ts": "1000.000100", "text": "first", "user": "U9" },
442 { "ts": "1000.000400", "text": "joined", "user": "U9", "subtype": "channel_join" },
443 ]
444 });
445 let parsed = parse_history("C1", &body);
446 assert_eq!(parsed.len(), 2);
447 assert_eq!(parsed[0].text, "first");
448 assert_eq!(parsed[0].user_id, "U9");
449 assert_eq!(parsed[0].channel_id, "C1");
450 assert_eq!(parsed[1].text, "second");
451 }
452
453 #[test]
454 fn parses_audio_files_without_text_for_governed_transcription() {
455 let body = serde_json::json!({"messages":[{"ts":"1.1","text":"","user":"U9","files":[{"mimetype":"audio/ogg","url_private_download":"https://files.example/voice"}]}]});
456 let parsed = parse_history("C1", &body);
457 assert_eq!(parsed.len(), 1);
458 assert_eq!(parsed[0].audio_mime.as_deref(), Some("audio/ogg"));
459 assert_eq!(
460 parsed[0].audio_url.as_deref(),
461 Some("https://files.example/voice")
462 );
463 }
464
465 #[test]
466 fn audio_file_without_download_url_is_not_routable_as_voice() {
467 let body = serde_json::json!({"messages":[{"ts":"1.2","text":"","user":"U9","files":[{"mimetype":"audio/ogg"}]}]});
468 let parsed = parse_history("C1", &body);
469 assert!(
470 parsed.is_empty(),
471 "unfetchable media must not enter the governed path"
472 );
473 }
474
475 #[test]
476 fn gateway_reply_session_id_is_preserved_for_playback() {
477 let value = serde_json::json!({"chunks":["ok"],"session_id":"slack-session"});
478 let reply = GatewayReply {
479 chunks: value["chunks"]
480 .as_array()
481 .unwrap()
482 .iter()
483 .filter_map(|v| v.as_str().map(String::from))
484 .collect(),
485 session_id: value["session_id"].as_str().map(String::from),
486 };
487 assert_eq!(reply.session_id.as_deref(), Some("slack-session"));
488 }
489
490 #[test]
491 fn timestamps_order_numerically_not_lexically() {
492 let body = serde_json::json!({
493 "ok": true,
494 "messages": [
495 { "ts": "1000.5", "text": "later", "user": "U9" },
496 { "ts": "999.9", "text": "earlier", "user": "U9" },
497 ]
498 });
499 let parsed = parse_history("C1", &body);
500 assert_eq!(parsed[0].text, "earlier");
501 assert_eq!(parsed[1].text, "later");
502 }
503}
504

Indexing the workspace…