diff --git a/docs/PRD-slack-enhancements.md b/docs/PRD-slack-enhancements.md new file mode 100644 index 000000000..80373c5fd --- /dev/null +++ b/docs/PRD-slack-enhancements.md @@ -0,0 +1,275 @@ +# PRD: Slack Connector Enhancements + +**Status:** Draft +**Author:** sookochoff +**Repo:** spacedriveapp/spacebot +**Branch target:** main + +--- + +## Background + +The Slack connector in `src/messaging/slack.rs` is functional but uses a narrow slice of what the Slack platform and the existing `slack-morphism` library offer. For a company running Spacebot as a team agent, the current adapter covers basic conversation but leaves significant productivity and workflow value on the table. + +This document enumerates the gaps, scores them by value and effort, and proposes a phased delivery plan. + +--- + +## Current State Audit + +### What the connector already does + +| Feature | Status | +|---|---| +| Receives plain text messages via Socket Mode | ✅ | +| Receives file attachments | ✅ | +| Sends plain text replies (single channel or thread) | ✅ | +| Message splitting at 4,000 chars | ✅ | +| Emoji reactions (add) | ✅ | +| Streaming via message edit (`chat.update`) | ✅ | +| File upload (v2 flow: get URL → upload → complete) | ✅ | +| DM broadcasting via `conversations.open` | ✅ | +| History backfill (`conversations.history` + `conversations.replies`) | ✅ | +| User display name resolution on message receipt | ✅ | +| Permission filtering (workspace, channel, DM allowlist) | ✅ | +| Health check (`api.test`) | ✅ | + +### What the connector explicitly skips + +| Feature | Notes in code | +|---|---| +| Typing indicator | Comment: `// no-op, Slack has no native typing indicator API` — **incorrect**, `assistant.threads.setStatus` is the modern equivalent | +| Message subtypes (edits, deletes) | Silently dropped | +| `app_mention` events | Not wired in `handle_push_event` — agent won't respond to `@bot` in channels where it isn't already present | +| Slash commands | `command_callback` in slack-morphism is available but unused | +| Block Kit (rich messages) | `SlackBlock` et al. exist in the library, never used in outbound | +| Interactive components (button clicks, select menus) | `SlackInteractionEvent` / `SlackInteractionBlockActionsEvent` available, unused | +| Ephemeral messages | `chat_post_ephemeral` in library, unused | +| Scheduled messages | `chat_schedule_message` in library, unused | +| `app_home` tab | `SlackAppHomeOpenedEvent` in library, unused | +| Thread awareness in history | Thread replies fetched separately from channel history — not unified | +| Reaction-received events | `SlackReactionAddedEvent` available, unused | +| Unfurl / link previews | `chat_unfurl` in library, unused | +| User presence | `users_get_presence` in library, unused | +| Message pinning by agent | `pins_add` in library, unused | +| User groups (`@here`, `@channel`, group handles) | `usergroups_list` in library, unused | +| `SlackApiAssistantThreadsSetStatus` | Available — correct typing-indicator mechanism, unused | + +--- + +## Opportunity Analysis + +The gaps above are not all equal. Evaluated from the perspective of **a company team using Slack day-to-day with a Spacebot agent:** + +### High value, low effort + +1. **`app_mention` support** — Users in a channel naturally @mention the bot. Currently those events are ignored. One-line fix in the push event handler. + +2. **Typing indicator via `assistant.threads.setStatus`** — The agent goes dark while thinking. On Discord the bot shows a typing indicator; on Slack it's a dead interface. The Slack Assistants API provides exactly this. Low code change; meaningful UX. + +3. **Block Kit for structured outbound messages** — A new `OutboundResponse::Blocks` variant (or `RichMessage`) lets the agent send formatted cards, section headers, dividers, and inline code. Slack's renderer makes these significantly more readable than walls of markdown. Library support is full; gap is only in the `OutboundResponse` enum and adapter glue. + +4. **Ephemeral messages** — Agent can whisper a confirmation or warning to only the requesting user. Useful for admin-type responses in shared channels. + +### High value, medium effort + +5. **Slash commands** — `/ask`, `/summarize`, `/task` etc. Slack's command UX is familiar to every team user. The `command_callback` is available in `SlackSocketModeListenerCallbacks` and just needs wiring up + a config schema for command→agent routing. + +6. **Interactive components (buttons/select menus)** — The agent posts a decision card; a user clicks "Approve" or "Reject". The `SlackInteractionBlockActionsEvent` comes back through `interaction_callback`. Needs: (a) `OutboundResponse` variant to express buttons, (b) inbound `MessageContent` variant for interaction events, (c) interaction callback wired in the socket mode listener. + +7. **Scheduled messages** — The agent can post to a channel at a specific future time (`chat.schedule_message`). This is a natural output for cron-triggered workflows, e.g. "post standup prompt at 9am Monday". Low Rust effort; zero Spacebot architecture change needed — it's purely an outbound response variant. + +### Medium value, medium effort + +8. **Message edit/delete awareness** — Currently silently dropped. If a user corrects a message that the agent already acted on, the agent should be aware. Needs a new `MessageContent` variant and some channel-level state to correlate `message_changed` subtype events back to prior turns. + +9. **Reaction-received events** — User reacts with ✅ or ❌ to acknowledge or reject a proposal. Arriving as `SlackReactionAddedEvent`. Useful for lightweight approvals without typing. + +10. **`app_home` tab** — The agent's home tab in the Slack sidebar can surface a custom view (memory summary, recent tasks, status). Entirely cosmetic from the agent's perspective but gives it a professional presence. + +### Lower priority / out of scope for now + +- User presence (`users.getPresence`) — niche +- Link unfurling — requires Events API subscription changes +- Message pinning — useful but not urgent +- User groups resolution — useful for `@channel` type broadcasts but edge case + +--- + +## Proposed Phases + +### Phase 1: Foundational UX (Recommended first PR) + +**Scope:** Four targeted changes that require no new architecture. + +| Item | Change required | +|---|---| +| **`app_mention` events** | Add `AppMention` arm to `handle_push_event` match | +| **Typing indicator** | Implement `send_status()` using `SlackApiAssistantThreadsSetStatusRequest` | +| **Ephemeral messages** | Add `OutboundResponse::Ephemeral { text, user_id }` variant; handle in Slack adapter with `chat_post_ephemeral` | +| **Reaction removal** | Add `OutboundResponse::RemoveReaction(String)` for completeness (currently only add) | + +**Effort:** Small — 1–2 days of focused Rust work +**Risk:** Low — isolated changes, no new dependencies +**PRs:** Likely one PR with four commits + +--- + +### Phase 2: Block Kit + Interactive Components + +**Scope:** Rich outbound messages and inbound interaction events. + +#### 2a — Block Kit outbound + +Add a new `OutboundResponse` variant: + +```rust +RichMessage { + /// Plain text fallback (always required — used by non-Slack adapters and notifications). + text: String, + /// Block Kit blocks. Slack-only; other adapters fall back to `text`. + blocks: Vec, +} +``` + +- Slack adapter: build `SlackMessageContent` with `blocks` +- Discord adapter: falls back to `text` +- Webhook/Telegram: falls back to `text` + +This is platform-agnostic at the type level. The LLM would request a structured response via a new `reply_with_blocks` tool or through a structured tool output schema. + +#### 2b — Interactive components inbound + +Add a new `MessageContent` variant: + +```rust +Interaction { + /// action_id of the block element that was acted on. + action_id: String, + /// block_id for context. + block_id: Option, + /// Selected value(s), if applicable (button value or select menu option). + value: Option, + /// Human-readable label of the selected option. + label: Option, + /// The original message ts so the agent can correlate back. + message_ts: Option, +} +``` + +Wire `interaction_callback` in the socket mode setup to receive `SlackInteractionBlockActionsEvent` and convert to an `InboundMessage` with this content. + +**Effort:** Medium — 3–4 days +**Risk:** Medium — touches `OutboundResponse` and `MessageContent` enums (shared types), all adapters need an audit pass to ensure the new variants are handled (even if as no-ops) +**PRs:** 2a and 2b can be separate PRs + +--- + +### Phase 3: Slash Commands + +**Scope:** Allow users to invoke the agent via `/command` in any channel. + +Config extension: + +```toml +[messaging.slack.commands] +"/ask" = { agent_id = "main", description = "Ask the agent a question" } +"/task" = { agent_id = "main", description = "Kick off a background task" } +``` + +Implementation: +- Wire `command_callback` in socket mode listener setup +- Parse `SlackCommandEvent` into an `InboundMessage` (using the command text as content) +- Route via the existing binding resolution +- Respond to the command's `response_url` or via `chat.post_message` to the channel + +Slash commands have a 3-second acknowledge requirement from Slack. The adapter must acknowledge immediately (200 OK with empty body or with a deferral message) and post the real reply asynchronously. + +**Effort:** Medium — 2–3 days +**Risk:** Medium — requires config schema extension and a deferred response pattern +**PRs:** Single focused PR + +--- + +### Phase 4: Scheduled Messages + +**Scope:** Let the agent post messages at a future time. + +New `OutboundResponse` variant: + +```rust +ScheduledMessage { + text: String, + /// Unix timestamp when the message should be delivered. + post_at: i64, + /// Optional Block Kit content. Falls back to `text` on non-Slack adapters. + blocks: Option>, +} +``` + +This maps cleanly to `chat_schedule_message` in slack-morphism and requires no new infrastructure. Practically, it makes cron workflows more polished: instead of sending a message immediately on job completion, the agent can time-shift delivery to normal working hours. + +**Effort:** Low — 1 day +**Risk:** Low — self-contained outbound variant +**PRs:** Single small PR, could be bundled with Phase 1 + +--- + +## What Does NOT Need to Change + +- **Socket Mode** — already the right transport for a self-hosted bot. No switch to Events API needed. +- **slack-morphism version** — 2.17 already has full Block Kit and interaction model support. No dependency bump needed. +- **Agent architecture** — Channels, branches, workers, cortex are all unaffected. This is entirely connector-layer work. +- **Existing permissions model** — `SlackPermissions` is sufficient for all phases. + +--- + +## Recommended Delivery Order + +| Phase | Deliverable | Effort | Impact | +|---|---|---|---| +| 1 | `app_mention`, typing indicator, ephemeral messages | Small | High | +| 2a | Block Kit outbound (`RichMessage`) | Medium | High | +| 4 | Scheduled messages | Small | Medium | +| 2b | Interactive components inbound | Medium | High | +| 3 | Slash commands | Medium | Medium | + +Phases 1 and 4 are natural first PRs — they're self-contained, low risk, and address the most visible UX gaps (agent ignores @mentions, no typing indicator, no rich formatting). Phase 2 is where the real workflow power is. + +--- + +## Open Questions + +1. **Tool schema for Block Kit**: Should the LLM specify blocks explicitly (raw Block Kit JSON in tool args), or should there be a higher-level tool interface (e.g., `reply_with_sections([{header, body}])`) that the adapter renders into blocks? The latter is safer and more portable but requires maintaining a thin DSL. + +2. **Interaction routing**: When a button click arrives, should it be routed to the same channel that originally sent the message with the buttons, or treated as a new conversation turn? The former is simpler; the latter is more correct for multi-step flows. + +3. **Slash command permissions**: Should slash commands go through the same binding/channel filter as regular messages, or have their own allowlist? Consider that `/ask` from an unbound workspace should probably be rejected. + +4. **Typing indicator scope**: `assistant.threads.setStatus` works in Slack Assistant threads (AI-mode threads). For regular channel messages, the canonical signal is `typing.setStatus` from the RTM API, which Socket Mode doesn't expose. We may need to accept that typing indicators only work in Slack Assistant thread contexts and document this clearly. + +--- + +## Files Affected (by phase) + +**Phase 1:** +- `src/messaging/slack.rs` — `handle_push_event`, new `send_status` impl, new response arms +- `src/lib.rs` — add `Ephemeral` to `OutboundResponse`, `RemoveReaction` optional +- Other adapters — handle new variants as no-ops + +**Phase 2:** +- `src/messaging/slack.rs` — interaction callback, `RichMessage` response arm +- `src/lib.rs` — `RichMessage` to `OutboundResponse`, `Interaction` to `MessageContent` +- `src/messaging/discord.rs` — handle `RichMessage` (fallback to text) +- `src/messaging/telegram.rs` — handle `RichMessage` (fallback to text) +- `src/messaging/webhook.rs` — handle `RichMessage` (passthrough or fallback) + +**Phase 3:** +- `src/config.rs` — `SlackCommandConfig` struct, extend `SlackConfig` +- `src/messaging/slack.rs` — command callback, deferred response pattern +- `src/lib.rs` — no change (command maps to existing `InboundMessage`) + +**Phase 4:** +- `src/messaging/slack.rs` — `ScheduledMessage` response arm +- `src/lib.rs` — add `ScheduledMessage` to `OutboundResponse` +- Other adapters — no-op or error for the new variant diff --git a/src/api/bindings.rs b/src/api/bindings.rs index 046e31860..c1df1c7d6 100644 --- a/src/api/bindings.rs +++ b/src/api/bindings.rs @@ -368,9 +368,15 @@ pub(super) async fn create_binding( } } }; - let adapter = crate::messaging::slack::SlackAdapter::new(&bot_token, &app_token, slack_perms); - if let Err(error) = manager.register_and_start(adapter).await { - tracing::error!(%error, "failed to hot-start slack adapter"); + match crate::messaging::slack::SlackAdapter::new(&bot_token, &app_token, slack_perms) { + Ok(adapter) => { + if let Err(error) = manager.register_and_start(adapter).await { + tracing::error!(%error, "failed to hot-start slack adapter"); + } + } + Err(error) => { + tracing::error!(%error, "failed to build slack adapter"); + } } } diff --git a/src/api/messaging.rs b/src/api/messaging.rs index 2c2cd7802..5fa88d089 100644 --- a/src/api/messaging.rs +++ b/src/api/messaging.rs @@ -307,13 +307,19 @@ pub(super) async fn toggle_platform( } } }; - let adapter = crate::messaging::slack::SlackAdapter::new( + match crate::messaging::slack::SlackAdapter::new( &slack_config.bot_token, &slack_config.app_token, perms, - ); - if let Err(error) = manager.register_and_start(adapter).await { - tracing::error!(%error, "failed to start slack adapter on toggle"); + ) { + Ok(adapter) => { + if let Err(error) = manager.register_and_start(adapter).await { + tracing::error!(%error, "failed to start slack adapter on toggle"); + } + } + Err(error) => { + tracing::error!(%error, "failed to build slack adapter on toggle"); + } } } } diff --git a/src/config.rs b/src/config.rs index 29e22ab00..bcb6fcf55 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2455,13 +2455,19 @@ pub fn spawn_file_watcher( Arc::new(arc_swap::ArcSwap::from_pointee(perms)) } }; - let adapter = crate::messaging::slack::SlackAdapter::new( + match crate::messaging::slack::SlackAdapter::new( &slack_config.bot_token, &slack_config.app_token, perms, - ); - if let Err(error) = manager.register_and_start(adapter).await { - tracing::error!(%error, "failed to hot-start slack adapter from config change"); + ) { + Ok(adapter) => { + if let Err(error) = manager.register_and_start(adapter).await { + tracing::error!(%error, "failed to hot-start slack adapter from config change"); + } + } + Err(error) => { + tracing::error!(%error, "failed to build slack adapter from config change"); + } } } } diff --git a/src/lib.rs b/src/lib.rs index c677a1e1c..23e3be8c6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,10 +4,10 @@ pub mod agent; pub mod api; pub mod config; pub mod conversation; +pub mod cron; pub mod daemon; pub mod db; pub mod error; -pub mod cron; pub mod hooks; pub mod identity; pub mod llm; @@ -18,9 +18,9 @@ pub mod prompts; pub mod secrets; pub mod settings; pub mod skills; -pub mod tools; #[cfg(feature = "metrics")] pub mod telemetry; +pub mod tools; pub mod update; pub use error::{Error, Result}; @@ -183,8 +183,12 @@ pub struct AgentDeps { } impl AgentDeps { - pub fn memory_search(&self) -> &Arc { &self.memory_search } - pub fn llm_manager(&self) -> &Arc { &self.llm_manager } + pub fn memory_search(&self) -> &Arc { + &self.memory_search + } + pub fn llm_manager(&self) -> &Arc { + &self.llm_manager + } /// Load the current routing config snapshot. pub fn routing(&self) -> arc_swap::Guard> { @@ -273,6 +277,33 @@ pub enum OutboundResponse { }, /// Add a reaction emoji to the triggering message. Reaction(String), + /// Remove a reaction emoji from the triggering message. + /// No-op on platforms that don't support reaction removal. + RemoveReaction(String), + /// Send a message visible only to the triggering user (ephemeral). + /// Falls back to a regular `Text` message on platforms that don't support it. + Ephemeral { + /// The message text (mrkdwn on Slack, plain text on others). + text: String, + /// The user ID who should see the message. Required on Slack; ignored elsewhere. + user_id: String, + }, + /// Send a message with Block Kit rich formatting (Slack) or plain text fallback. + /// Other adapters use `text` as-is. + RichMessage { + /// Plain-text fallback — always present, used for notifications and non-Slack adapters. + text: String, + /// Slack Block Kit blocks. Serialised as raw JSON so callers don't need to depend on + /// slack-morphism types. The Slack adapter deserialises these at send time. + blocks: Vec, + }, + /// Schedule a message to be posted at a future Unix timestamp (Slack only). + /// Other adapters send immediately as a regular `Text` message. + ScheduledMessage { + text: String, + /// Unix epoch seconds when the message should be delivered. + post_at: i64, + }, StreamStart, StreamChunk(String), StreamEnd, @@ -303,9 +334,21 @@ pub enum StatusUpdate { Thinking, /// Cancel the typing indicator (e.g. when the skip tool fires). StopTyping, - ToolStarted { tool_name: String }, - ToolCompleted { tool_name: String }, - BranchStarted { branch_id: BranchId }, - WorkerStarted { worker_id: WorkerId, task: String }, - WorkerCompleted { worker_id: WorkerId, result: String }, + ToolStarted { + tool_name: String, + }, + ToolCompleted { + tool_name: String, + }, + BranchStarted { + branch_id: BranchId, + }, + WorkerStarted { + worker_id: WorkerId, + task: String, + }, + WorkerCompleted { + worker_id: WorkerId, + result: String, + }, } diff --git a/src/main.rs b/src/main.rs index c7496b750..3bf9cc27a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1257,14 +1257,16 @@ async fn initialize_agents( if let Some(slack_config) = &config.messaging.slack { if slack_config.enabled { - let adapter = spacebot::messaging::slack::SlackAdapter::new( + match spacebot::messaging::slack::SlackAdapter::new( &slack_config.bot_token, &slack_config.app_token, slack_permissions .clone() .expect("slack permissions initialized when slack is enabled"), - ); - new_messaging_manager.register(adapter).await; + ) { + Ok(adapter) => { new_messaging_manager.register(adapter).await; } + Err(error) => { tracing::error!(%error, "failed to build slack adapter"); } + } } } diff --git a/src/messaging/discord.rs b/src/messaging/discord.rs index 4b0b5e1a5..a966a5c45 100644 --- a/src/messaging/discord.rs +++ b/src/messaging/discord.rs @@ -250,6 +250,40 @@ impl Messaging for DiscordAdapter { OutboundResponse::Status(status) => { self.send_status(message, status).await?; } + // Slack-specific variants — graceful fallbacks for Discord + OutboundResponse::RemoveReaction(_) => {} // no-op + OutboundResponse::Ephemeral { text, .. } => { + // Discord has no ephemeral equivalent here; send as regular text + if let Ok(channel_id) = self.extract_channel_id(message) { + let http = self.get_http().await?; + channel_id + .say(&*http, &text) + .await + .context("failed to send ephemeral fallback on discord")?; + } + } + OutboundResponse::RichMessage { text, .. } => { + // No Block Kit on Discord — plain text fallback + if let Ok(channel_id) = self.extract_channel_id(message) { + let http = self.get_http().await?; + for chunk in split_message(&text, 2000) { + channel_id + .say(&*http, &chunk) + .await + .context("failed to send rich message fallback on discord")?; + } + } + } + OutboundResponse::ScheduledMessage { text, .. } => { + // Discord has no native scheduled messages — send immediately + if let Ok(channel_id) = self.extract_channel_id(message) { + let http = self.get_http().await?; + channel_id + .say(&*http, &text) + .await + .context("failed to send scheduled message fallback on discord")?; + } + } } Ok(()) diff --git a/src/messaging/slack.rs b/src/messaging/slack.rs index 76a3a8d99..1f5a1f345 100644 --- a/src/messaging/slack.rs +++ b/src/messaging/slack.rs @@ -1,8 +1,29 @@ //! Slack messaging adapter using slack-morphism. +//! +//! ## Features +//! +//! **Inbound** +//! - Plain text and file-attachment messages (Socket Mode) +//! - `app_mention` events — agent responds when @-mentioned in any channel +//! - Message subtype filtering (edits/deletes ignored) +//! - Per-workspace / per-channel / DM permission filtering (hot-reloadable) +//! - Full user identity resolution (display name, mention tag) +//! +//! **Outbound** +//! - Plain text with smart UTF-8-safe chunking +//! - Thread replies +//! - File uploads (v2 flow) +//! - Emoji reactions (add + remove) +//! - Ephemeral messages (visible only to the triggering user) +//! - Block Kit rich messages with plain-text fallback +//! - Scheduled messages (`chat.scheduleMessage`) +//! - Streaming via `chat.update` edits +//! - Typing indicator via `assistant.threads.setStatus` +//! - DM broadcast via `conversations.open` use crate::config::SlackPermissions; use crate::messaging::traits::{HistoryMessage, InboundStream, Messaging}; -use crate::{InboundMessage, MessageContent, OutboundResponse}; +use crate::{InboundMessage, MessageContent, OutboundResponse, StatusUpdate}; use anyhow::Context as _; use arc_swap::ArcSwap; @@ -25,12 +46,18 @@ struct SlackUserIdentity { username: Option, } -/// Slack adapter state. +/// Slack adapter. pub struct SlackAdapter { bot_token: String, app_token: String, permissions: Arc>, - /// Maps InboundMessage.id to the Slack message timestamp (ts) for editing during streaming. + /// Shared HTTP client — constructed once, reused across all API calls. + /// Holds a hyper connection pool internally; allocating one per call would + /// discard that pool on every status update / respond / broadcast. + client: Arc, + /// Pre-built API token wrapping `bot_token`. Created once alongside `client`. + token: SlackApiToken, + /// Maps InboundMessage.id → Slack ts for streaming edits. active_messages: Arc>>, shutdown_tx: Arc>>>, } @@ -40,39 +67,58 @@ impl SlackAdapter { bot_token: impl Into, app_token: impl Into, permissions: Arc>, - ) -> Self { - Self { - bot_token: bot_token.into(), + ) -> anyhow::Result { + let bot_token = bot_token.into(); + let client = Arc::new(SlackClient::new( + SlackClientHyperConnector::new().context("failed to create slack HTTP connector")?, + )); + let token = SlackApiToken::new(SlackApiTokenValue(bot_token.clone())); + Ok(Self { + bot_token, app_token: app_token.into(), permissions, + client, + token, active_messages: Arc::new(RwLock::new(HashMap::new())), shutdown_tx: Arc::new(RwLock::new(None)), - } + }) } - /// Create a session for making API calls. - fn create_session(&self) -> anyhow::Result<(Arc, SlackApiToken)> { - let client = Arc::new(SlackClient::new( - SlackClientHyperConnector::new().context("failed to create slack connector")?, - )); - let token = SlackApiToken::new(SlackApiTokenValue(self.bot_token.clone())); - Ok((client, token)) + /// Open a session against the cached client using the cached bot token. + fn session(&self) -> SlackClientSession<'_, SlackClientHyperHttpsConnector> { + self.client.open_session(&self.token) } } -/// Socket mode push event handler. Must be a `fn` pointer (not a closure) -/// because slack-morphism requires `UserCallbackFunction` which is a fn pointer type. +// --------------------------------------------------------------------------- +// Inbound event handlers (fn pointers — slack-morphism requirement) +// --------------------------------------------------------------------------- + +/// Handle regular channel/DM messages. async fn handle_push_event( event: SlackPushEventCallback, client: Arc, states: SlackClientEventsUserState, ) -> UserCallbackResult<()> { - let msg_event = match event.event { - SlackEventCallbackBody::Message(msg) => msg, - _ => return Ok(()), - }; + match event.event { + SlackEventCallbackBody::Message(msg) => { + handle_message_event(msg, &event.team_id, client, states).await + } + SlackEventCallbackBody::AppMention(mention) => { + handle_app_mention_event(mention, &event.team_id, client, states).await + } + _ => Ok(()), + } +} - // Skip message edits/deletes +/// Core logic shared by Message and AppMention handlers. +async fn handle_message_event( + msg_event: SlackMessageEvent, + team_id: &SlackTeamId, + client: Arc, + states: SlackClientEventsUserState, +) -> UserCallbackResult<()> { + // Skip message edits / deletes / bot_message subtypes if msg_event.subtype.is_some() { return Ok(()); } @@ -84,16 +130,14 @@ async fn handle_push_event( let user_id = msg_event.sender.user.as_ref().map(|u| u.0.clone()); - // Skip messages from the bot itself if user_id.as_deref() == Some(&adapter_state.bot_user_id) { - return Ok(()); + return Ok(()); // ignore self } - - // Skip messages with no user (system messages) if user_id.is_none() { - return Ok(()); + return Ok(()); // system message } - let team_id = event.team_id.0.clone(); + + let team_id_str = team_id.0.clone(); let channel_id = msg_event .origin .channel @@ -102,10 +146,9 @@ async fn handle_push_event( .unwrap_or_default(); let ts = msg_event.origin.ts.0.clone(); - // Load permissions snapshot let perms = adapter_state.permissions.load(); - // DM filter: Slack DM channel IDs start with "D" + // DM filter if channel_id.starts_with('D') { if perms.dm_allowed_users.is_empty() { return Ok(()); @@ -119,137 +162,128 @@ async fn handle_push_event( // Workspace filter if let Some(ref filter) = perms.workspace_filter { - if !filter.contains(&team_id) { + if !filter.contains(&team_id_str) { return Ok(()); } } // Channel filter - if let Some(allowed_channels) = perms.channel_filter.get(&team_id) { - if !allowed_channels.is_empty() && !allowed_channels.contains(&channel_id) { + if let Some(allowed) = perms.channel_filter.get(&team_id_str) { + if !allowed.is_empty() && !allowed.contains(&channel_id) { return Ok(()); } } - // Build conversation ID (threaded conversations get their own channel) let conversation_id = if let Some(ref thread_ts) = msg_event.origin.thread_ts { - format!("slack:{}:{}:{}", team_id, channel_id, thread_ts.0) + format!("slack:{}:{}:{}", team_id_str, channel_id, thread_ts.0) } else { - format!("slack:{}:{}", team_id, channel_id) + format!("slack:{}:{}", team_id_str, channel_id) }; - // Extract content - let content = if let Some(ref msg_content) = msg_event.content { - if let Some(ref files) = msg_content.files { - let attachments: Vec = files - .iter() - .filter_map(|f| { - let url = f.url_private.as_ref()?; - Some(crate::Attachment { - filename: f.name.clone().unwrap_or_else(|| "unnamed".into()), - mime_type: f.mimetype.as_ref().map(|m| m.0.clone()).unwrap_or_default(), - url: url.to_string(), - size_bytes: None, - }) - }) - .collect(); - - if attachments.is_empty() { - MessageContent::Text(msg_content.text.clone().unwrap_or_default()) - } else { - MessageContent::Media { - text: msg_content.text.clone(), - attachments, - } - } - } else { - MessageContent::Text(msg_content.text.clone().unwrap_or_default()) - } - } else { - MessageContent::Text(String::new()) - }; + let content = extract_message_content(&msg_event.content); + + let (metadata, formatted_author) = build_metadata_and_author( + &team_id_str, + &channel_id, + &ts, + msg_event.origin.thread_ts.as_ref().map(|t| t.0.as_str()), + user_id.as_deref(), + msg_event.sender.user.as_ref(), + &client, + &adapter_state.bot_token, + ) + .await; + + send_inbound( + &adapter_state.inbound_tx, + ts, + conversation_id, + user_id.unwrap_or_default(), + content, + metadata, + formatted_author, + ) + .await; - // Build metadata - let mut metadata = HashMap::new(); - metadata.insert( - "slack_workspace_id".into(), - serde_json::Value::String(team_id), - ); - metadata.insert( - "slack_channel_id".into(), - serde_json::Value::String(channel_id), - ); - metadata.insert( - "slack_message_ts".into(), - serde_json::Value::String(ts.clone()), - ); - if let Some(ref thread_ts) = msg_event.origin.thread_ts { - metadata.insert( - "slack_thread_ts".into(), - serde_json::Value::String(thread_ts.0.clone()), - ); + Ok(()) +} + +/// Handle `app_mention` events — fired when the bot is @-mentioned in a channel +/// it may not be a primary member of. +/// +/// `SlackAppMentionEvent` has a flat `user: SlackUserId` field (not a `sender` sub-struct) +/// and a flat `channel: SlackChannelId` field (not nested in `origin`). +async fn handle_app_mention_event( + mention: SlackAppMentionEvent, + team_id: &SlackTeamId, + client: Arc, + states: SlackClientEventsUserState, +) -> UserCallbackResult<()> { + let state_guard = states.read().await; + let adapter_state = state_guard + .get_user_state::>() + .expect("SlackAdapterState must be in user_state"); + + let user_id = mention.user.0.clone(); + + if user_id == adapter_state.bot_user_id { + return Ok(()); } - if let Some(ref uid) = user_id { - metadata.insert( - "slack_user_id".into(), - serde_json::Value::String(uid.clone()), - ); - metadata.insert("sender_id".into(), serde_json::Value::String(uid.clone())); - metadata.insert( - "slack_user_mention".into(), - serde_json::Value::String(format!("<@{uid}>")), - ); + + let team_id_str = team_id.0.clone(); + let channel_id = mention.channel.0.clone(); + let ts = mention.origin.ts.0.clone(); + + let perms = adapter_state.permissions.load(); + + // Workspace filter applies to mentions too + if let Some(ref filter) = perms.workspace_filter { + if !filter.contains(&team_id_str) { + return Ok(()); + } } - // Resolve username/display name for better attribution and mention support. - if let Some(ref uid) = msg_event.sender.user { - let token = SlackApiToken::new(SlackApiTokenValue(adapter_state.bot_token.clone())); - let session = client.open_session(&token); - if let Ok(user_info) = session - .users_info(&SlackApiUsersInfoRequest::new(uid.clone())) - .await - { - let identity = resolve_slack_user_identity(&user_info.user, &uid.0); - let display_with_mention = format!("{} (<@{}>)", identity.display_name, uid.0); - metadata.insert( - "sender_display_name".into(), - serde_json::Value::String(display_with_mention.clone()), - ); - if let Some(name) = identity.username { - metadata.insert( - "sender_username".into(), - serde_json::Value::String(name), - ); - } + // Channel filter — same logic as handle_message_event + if let Some(allowed) = perms.channel_filter.get(&team_id_str) { + if !allowed.is_empty() && !allowed.contains(&channel_id) { + return Ok(()); } } - // For Slack, formatted_author is the display_with_mention if user info was resolved - let formatted_author = if let Some(uid) = &user_id { - metadata - .get("sender_display_name") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()) - .or_else(|| Some(uid.clone())) + let conversation_id = if let Some(ref thread_ts) = mention.origin.thread_ts { + format!("slack:{}:{}:{}", team_id_str, channel_id, thread_ts.0) } else { - None + format!("slack:{}:{}", team_id_str, channel_id) }; - let inbound = InboundMessage { - id: ts, - source: "slack".into(), + // Strip the leading @-mention from the text so the agent sees clean input + let raw_text = mention.content.text.clone().unwrap_or_default(); + let text = strip_bot_mention(&raw_text, &adapter_state.bot_user_id); + let content = MessageContent::Text(text); + + let slack_uid = SlackUserId(user_id.clone()); + let (metadata, formatted_author) = build_metadata_and_author( + &team_id_str, + &channel_id, + &ts, + mention.origin.thread_ts.as_ref().map(|t| t.0.as_str()), + Some(&user_id), + Some(&slack_uid), + &client, + &adapter_state.bot_token, + ) + .await; + + send_inbound( + &adapter_state.inbound_tx, + ts, conversation_id, - sender_id: user_id.unwrap_or_default(), - agent_id: None, + user_id, content, - timestamp: chrono::Utc::now(), metadata, formatted_author, - }; - - if let Err(error) = adapter_state.inbound_tx.send(inbound).await { - tracing::warn!(%error, "failed to send inbound message from Slack"); - } + ) + .await; Ok(()) } @@ -263,6 +297,10 @@ fn slack_error_handler( HttpStatusCode::OK } +// --------------------------------------------------------------------------- +// Messaging trait impl +// --------------------------------------------------------------------------- + impl Messaging for SlackAdapter { fn name(&self) -> &str { "slack" @@ -274,14 +312,9 @@ impl Messaging for SlackAdapter { *self.shutdown_tx.write().await = Some(shutdown_tx); - let client = Arc::new(SlackClient::new( - SlackClientHyperConnector::new().context("failed to create slack connector")?, - )); - - // Fetch bot's own user ID so we can filter out self-messages - let bot_token = SlackApiToken::new(SlackApiTokenValue(self.bot_token.clone())); - let session = client.open_session(&bot_token); - let auth_response = session + // Reuse the shared client for auth.test; no new allocation needed. + let auth_response = self + .session() .auth_test() .await .context("failed to call auth.test for bot user ID")?; @@ -295,11 +328,16 @@ impl Messaging for SlackAdapter { bot_user_id, }); - let callbacks = SlackSocketModeListenerCallbacks::new() - .with_push_events(handle_push_event); + let callbacks = SlackSocketModeListenerCallbacks::new().with_push_events(handle_push_event); + + // The socket mode listener needs its own client — it owns a persistent + // WebSocket connection. The shared self.client is for REST calls only. + let listener_client = Arc::new(SlackClient::new( + SlackClientHyperConnector::new().context("failed to create slack socket mode connector")?, + )); let listener_environment = Arc::new( - SlackClientEventsListenerEnvironment::new(client.clone()) + SlackClientEventsListenerEnvironment::new(listener_client.clone()) .with_error_handler(slack_error_handler) .with_user_state(adapter_state), ); @@ -331,8 +369,65 @@ impl Messaging for SlackAdapter { } }); - let stream = tokio_stream::wrappers::ReceiverStream::new(inbound_rx); - Ok(Box::pin(stream)) + Ok(Box::pin(tokio_stream::wrappers::ReceiverStream::new( + inbound_rx, + ))) + } + + /// Show a typing-style status in Slack Assistant threads while the agent is thinking. + /// + /// Uses `assistant.threads.setStatus` — the correct API for Socket Mode bots running + /// in Slack Assistant thread contexts. + /// + /// **Scope limitation (Slack API):** this API only works inside Slack Assistant threads + /// (i.e. messages that carry a `thread_ts`). Regular channel messages and DMs do not + /// support `setStatus`; this function no-ops for them rather than erroring. If you are + /// not seeing typing indicators, verify the conversation is inside an Assistant thread. + /// + /// Pass an empty string to clear the status (e.g. on `StopTyping`). + async fn send_status( + &self, + message: &InboundMessage, + status: StatusUpdate, + ) -> crate::Result<()> { + let thread_ts = match extract_thread_ts(message) { + Some(ts) => ts, + None => { + tracing::debug!( + message_id = %message.id, + "skipping assistant.threads.setStatus — message has no thread_ts \ + (typing indicators only work in Slack Assistant threads)" + ); + return Ok(()); + } + }; + let channel_id = match extract_channel_id(message) { + Ok(id) => id, + Err(_) => return Ok(()), + }; + + let status_text = match &status { + StatusUpdate::Thinking => "Thinking…".to_string(), + StatusUpdate::StopTyping => String::new(), // empty string clears the status + StatusUpdate::ToolStarted { .. } => "Working…".to_string(), + StatusUpdate::ToolCompleted { .. } => "Working…".to_string(), + _ => "Working…".to_string(), + }; + + let session = self.session(); + + let req = SlackApiAssistantThreadsSetStatusRequest { + channel_id, + thread_ts, + status: status_text, + }; + + // Best-effort — don't propagate status errors into the main response pipeline. + if let Err(err) = session.assistant_threads_set_status(&req).await { + tracing::debug!(error = %err, "failed to set slack assistant thread status (non-fatal)"); + } + + Ok(()) } async fn respond( @@ -340,9 +435,7 @@ impl Messaging for SlackAdapter { message: &InboundMessage, response: OutboundResponse, ) -> crate::Result<()> { - let (client, token) = self.create_session()?; - let session = client.open_session(&token); - + let session = self.session(); let channel_id = extract_channel_id(message)?; match response { @@ -355,18 +448,17 @@ impl Messaging for SlackAdapter { markdown_content(chunk), ); req = req.opt_thread_ts(thread_ts.clone()); - session .chat_post_message(&req) .await .context("failed to send slack message")?; } } + OutboundResponse::ThreadReply { thread_name: _, text, } => { - // Use existing thread_ts, or create a thread from the source message let thread_ts = extract_thread_ts(message).or_else(|| extract_message_ts(message)); for chunk in split_message(&text, 12_000) { @@ -375,19 +467,24 @@ impl Messaging for SlackAdapter { markdown_content(chunk), ); req = req.opt_thread_ts(thread_ts.clone()); - session .chat_post_message(&req) .await .context("failed to send slack thread reply")?; } } - OutboundResponse::File { filename, data, mime_type, caption } => { - // Slack's v2 upload flow: get upload URL, upload bytes, complete + + OutboundResponse::File { + filename, + data, + mime_type, + caption, + } => { let upload_url_response = session - .get_upload_url_external( - &SlackApiFilesGetUploadUrlExternalRequest::new(filename.clone(), data.len()), - ) + .get_upload_url_external(&SlackApiFilesGetUploadUrlExternalRequest::new( + filename.clone(), + data.len(), + )) .await .context("failed to get slack upload URL")?; @@ -401,55 +498,129 @@ impl Messaging for SlackAdapter { .context("failed to upload file to slack")?; let thread_ts = extract_thread_ts(message); - let file_complete = SlackApiFilesComplete::new(upload_url_response.file_id) - .with_title(filename); - - let mut complete_request = SlackApiFilesCompleteUploadExternalRequest::new( - vec![file_complete], - ) - .with_channel_id(channel_id.clone()); - + let file_complete = + SlackApiFilesComplete::new(upload_url_response.file_id).with_title(filename); + let mut complete_request = + SlackApiFilesCompleteUploadExternalRequest::new(vec![file_complete]) + .with_channel_id(channel_id.clone()); complete_request = complete_request.opt_initial_comment(caption); complete_request = complete_request.opt_thread_ts(thread_ts); - session .files_complete_upload_external(&complete_request) .await .context("failed to complete slack file upload")?; } - OutboundResponse::Reaction(emoji) => { - let ts = extract_message_ts(message) - .context("missing slack_message_ts for reaction")?; - - let reaction_name = sanitize_reaction_name(&emoji); + OutboundResponse::Reaction(emoji) => { + let ts = + extract_message_ts(message).context("missing slack_message_ts for reaction")?; let req = SlackApiReactionsAddRequest::new( channel_id.clone(), - SlackReactionName(reaction_name), + SlackReactionName(sanitize_reaction_name(&emoji)), ts, ); - session .reactions_add(&req) .await .context("failed to add slack reaction")?; } + + OutboundResponse::RemoveReaction(emoji) => { + let ts = extract_message_ts(message) + .context("missing slack_message_ts for reaction removal")?; + // channel and timestamp are Optional on the remove request + let req = SlackApiReactionsRemoveRequest::new(SlackReactionName( + sanitize_reaction_name(&emoji), + )) + .with_channel(channel_id.clone()) + .with_timestamp(ts); + session + .reactions_remove(&req) + .await + .context("failed to remove slack reaction")?; + } + + OutboundResponse::Ephemeral { text, user_id } => { + let thread_ts = extract_thread_ts(message); + let req = SlackApiChatPostEphemeralRequest::new( + channel_id.clone(), + SlackUserId(user_id), + SlackMessageContent::new().with_text(text), + ) + .opt_thread_ts(thread_ts); + session + .chat_post_ephemeral(&req) + .await + .context("failed to send slack ephemeral message")?; + } + + OutboundResponse::RichMessage { text, blocks } => { + let thread_ts = extract_thread_ts(message); + let attempted = blocks.len(); + let slack_blocks = deserialize_blocks(&blocks); + let dropped = attempted - slack_blocks.len(); + let content = if slack_blocks.is_empty() { + if attempted > 0 { + tracing::warn!( + attempted, + "all {} block(s) failed to deserialise — sending plain text fallback", + attempted + ); + } + SlackMessageContent::new().with_text(text) + } else { + if dropped > 0 { + tracing::warn!( + dropped, + attempted, + "{} of {} block(s) dropped due to deserialisation errors", + dropped, + attempted + ); + } + SlackMessageContent::new() + .with_text(text) + .with_blocks(slack_blocks) + }; + let mut req = SlackApiChatPostMessageRequest::new(channel_id.clone(), content); + req = req.opt_thread_ts(thread_ts); + session + .chat_post_message(&req) + .await + .context("failed to send slack rich message")?; + } + + OutboundResponse::ScheduledMessage { text, post_at } => { + let thread_ts = extract_thread_ts(message); + let post_at_dt = chrono::DateTime::::from_timestamp(post_at, 0) + .context("invalid post_at unix timestamp for scheduled message")?; + let req = SlackApiChatScheduleMessageRequest::new( + channel_id.clone(), + SlackMessageContent::new().with_text(text), + SlackDateTime(post_at_dt), + ) + .opt_thread_ts(thread_ts); + session + .chat_schedule_message(&req) + .await + .context("failed to schedule slack message")?; + } + OutboundResponse::StreamStart => { let req = SlackApiChatPostMessageRequest::new( channel_id.clone(), SlackMessageContent::new().with_text("\u{200B}".into()), ); - let resp = session .chat_post_message(&req) .await .context("failed to send stream placeholder")?; - self.active_messages .write() .await .insert(message.id.clone(), resp.ts.0); } + OutboundResponse::StreamChunk(text) => { let active = self.active_messages.read().await; if let Some(ts) = active.get(&message.id) { @@ -459,43 +630,35 @@ impl Messaging for SlackAdapter { } else { text }; - let req = SlackApiChatUpdateRequest::new( channel_id.clone(), markdown_content(display_text), SlackTs(ts.clone()), ); - if let Err(error) = session.chat_update(&req).await { tracing::warn!(%error, "failed to edit streaming message"); } } } + OutboundResponse::StreamEnd => { self.active_messages.write().await.remove(&message.id); } - OutboundResponse::Status(_) => {} // no-op, Slack has no native typing indicator + + OutboundResponse::Status(_) => { + // Status updates are handled via send_status(); ignored here. + } } Ok(()) } - // send_status: uses the default no-op from the Messaging trait. - // Slack has no native typing indicator API. + async fn broadcast(&self, target: &str, response: OutboundResponse) -> crate::Result<()> { + let session = self.session(); - async fn broadcast( - &self, - target: &str, - response: OutboundResponse, - ) -> crate::Result<()> { - let (client, token) = self.create_session()?; - let session = client.open_session(&token); - - // Support "dm:{user_id}" targets by opening a DM conversation first let channel_id = if let Some(user_id_str) = target.strip_prefix("dm:") { - let user_id = SlackUserId(user_id_str.to_string()); let open_req = SlackApiConversationsOpenRequest::new() - .with_users(vec![user_id]); + .with_users(vec![SlackUserId(user_id_str.to_string())]); let open_resp = session .conversations_open(&open_req) .await @@ -505,17 +668,43 @@ impl Messaging for SlackAdapter { SlackChannelId(target.to_string()) }; - if let OutboundResponse::Text(text) = response { - for chunk in split_message(&text, 12_000) { - let req = SlackApiChatPostMessageRequest::new( - channel_id.clone(), - markdown_content(chunk), - ); - + match response { + OutboundResponse::Text(text) => { + for chunk in split_message(&text, 12_000) { + let req = SlackApiChatPostMessageRequest::new( + channel_id.clone(), + markdown_content(chunk), + ); + session + .chat_post_message(&req) + .await + .context("failed to broadcast slack message")?; + } + } + OutboundResponse::RichMessage { text, blocks } => { + let slack_blocks = deserialize_blocks(&blocks); + let content = if slack_blocks.is_empty() { + SlackMessageContent::new().with_text(text) + } else { + SlackMessageContent::new() + .with_text(text) + .with_blocks(slack_blocks) + }; + let req = SlackApiChatPostMessageRequest::new(channel_id.clone(), content); session .chat_post_message(&req) .await - .context("failed to broadcast slack message")?; + .context("failed to broadcast slack rich message")?; + } + // Other variants are not meaningful for broadcast (e.g. Ephemeral requires a + // specific user_id from a live conversation, Reaction requires an existing ts, + // Scheduled/Stream are respond()-only flows). + other => { + tracing::warn!( + variant = %variant_name(&other), + target = %target, + "broadcast() received a variant that is not supported for broadcast — ignoring" + ); } } @@ -527,8 +716,7 @@ impl Messaging for SlackAdapter { message: &InboundMessage, limit: usize, ) -> crate::Result> { - let (client, token) = self.create_session()?; - let session = client.open_session(&token); + let session = self.session(); let channel_id = extract_channel_id(message)?; let thread_ts = extract_thread_ts(message); let capped_limit = limit.min(100) as u16; @@ -536,7 +724,6 @@ impl Messaging for SlackAdapter { let messages = if let Some(ts) = thread_ts { let req = SlackApiConversationsRepliesRequest::new(channel_id.clone(), ts) .with_limit(capped_limit); - session .conversations_replies(&req) .await @@ -546,7 +733,6 @@ impl Messaging for SlackAdapter { let req = SlackApiConversationsHistoryRequest::new() .with_channel(channel_id.clone()) .with_limit(capped_limit); - session .conversations_history(&req) .await @@ -557,14 +743,13 @@ impl Messaging for SlackAdapter { let mut user_identity_by_id = HashMap::new(); for user_id in messages .iter() - .filter_map(|msg| msg.sender.user.as_ref().map(|user| user.0.clone())) + .filter_map(|msg| msg.sender.user.as_ref().map(|u| u.0.clone())) { if user_identity_by_id.contains_key(&user_id) { continue; } - let slack_user_id = SlackUserId(user_id.clone()); if let Ok(user_info) = session - .users_info(&SlackApiUsersInfoRequest::new(slack_user_id)) + .users_info(&SlackApiUsersInfoRequest::new(SlackUserId(user_id.clone()))) .await { let identity = resolve_slack_user_identity(&user_info.user, &user_id); @@ -572,7 +757,7 @@ impl Messaging for SlackAdapter { } } - // Slack returns newest-first, reverse to chronological. + // Slack returns newest-first; reverse to chronological. let result: Vec = messages .into_iter() .rev() @@ -581,23 +766,18 @@ impl Messaging for SlackAdapter { let is_bot = user_id.is_none() || msg.sender.bot_id.is_some(); let author = if is_bot { "bot".to_string() - } else if let Some(user_id) = user_id { + } else if let Some(uid) = user_id { let display_name = user_identity_by_id - .get(&user_id) - .map(|identity| identity.display_name.clone()) - .unwrap_or_else(|| user_id.clone()); - format!("{display_name} (<@{user_id}>)") + .get(&uid) + .map(|i| i.display_name.clone()) + .unwrap_or_else(|| uid.clone()); + format!("{display_name} (<@{uid}>)") } else { "unknown".to_string() }; - HistoryMessage { author, - content: msg - .content - .text - .clone() - .unwrap_or_default(), + content: msg.content.text.clone().unwrap_or_default(), is_bot, } }) @@ -613,30 +793,27 @@ impl Messaging for SlackAdapter { } async fn health_check(&self) -> crate::Result<()> { - let (client, token) = self.create_session()?; - let session = client.open_session(&token); - + let session = self.session(); session .api_test(&SlackApiTestRequest::new()) .await .context("slack health check failed")?; - Ok(()) } async fn shutdown(&self) -> crate::Result<()> { self.active_messages.write().await.clear(); - if let Some(tx) = self.shutdown_tx.write().await.take() { let _ = tx.send(()).await; } - tracing::info!("slack adapter shut down"); Ok(()) } } -// -- Helper functions -- +// --------------------------------------------------------------------------- +// Helper functions +// --------------------------------------------------------------------------- fn extract_channel_id(message: &InboundMessage) -> anyhow::Result { message @@ -685,8 +862,180 @@ fn markdown_content(text: impl Into) -> SlackMessageContent { } } -/// Split a message into chunks that fit within Slack's character limit. -/// Tries to split at newlines, then spaces, then hard-cuts. +/// Extract `MessageContent` from an optional `SlackMessageContent`. +fn extract_message_content(content: &Option) -> MessageContent { + let Some(msg_content) = content else { + return MessageContent::Text(String::new()); + }; + + if let Some(ref files) = msg_content.files { + let attachments: Vec = files + .iter() + .filter_map(|f| { + let url = f.url_private.as_ref()?; + Some(crate::Attachment { + filename: f.name.clone().unwrap_or_else(|| "unnamed".into()), + mime_type: f.mimetype.as_ref().map(|m| m.0.clone()).unwrap_or_default(), + url: url.to_string(), + size_bytes: None, + }) + }) + .collect(); + + if !attachments.is_empty() { + return MessageContent::Media { + text: msg_content.text.clone(), + attachments, + }; + } + } + + MessageContent::Text(msg_content.text.clone().unwrap_or_default()) +} + +/// Build the metadata map and formatted author string shared by all inbound paths. +#[allow(clippy::too_many_arguments)] +async fn build_metadata_and_author( + team_id: &str, + channel_id: &str, + ts: &str, + thread_ts: Option<&str>, + user_id: Option<&str>, + slack_user_id: Option<&SlackUserId>, + client: &Arc, + bot_token: &str, +) -> (HashMap, Option) { + let mut metadata = HashMap::new(); + + metadata.insert( + "slack_workspace_id".into(), + serde_json::Value::String(team_id.into()), + ); + metadata.insert( + "slack_channel_id".into(), + serde_json::Value::String(channel_id.into()), + ); + metadata.insert( + "slack_message_ts".into(), + serde_json::Value::String(ts.into()), + ); + + if let Some(tts) = thread_ts { + metadata.insert( + "slack_thread_ts".into(), + serde_json::Value::String(tts.into()), + ); + } + + if let Some(uid) = user_id { + metadata.insert( + "slack_user_id".into(), + serde_json::Value::String(uid.into()), + ); + metadata.insert("sender_id".into(), serde_json::Value::String(uid.into())); + metadata.insert( + "slack_user_mention".into(), + serde_json::Value::String(format!("<@{uid}>")), + ); + } + + // Resolve display name and username + let mut formatted_author = user_id.map(|u| u.to_string()); + + if let Some(uid) = slack_user_id { + let token = SlackApiToken::new(SlackApiTokenValue(bot_token.to_string())); + let session = client.open_session(&token); + if let Ok(user_info) = session + .users_info(&SlackApiUsersInfoRequest::new(uid.clone())) + .await + { + let identity = resolve_slack_user_identity(&user_info.user, &uid.0); + let display_with_mention = format!("{} (<@{}>)", identity.display_name, uid.0); + metadata.insert( + "sender_display_name".into(), + serde_json::Value::String(display_with_mention.clone()), + ); + if let Some(name) = identity.username { + metadata.insert("sender_username".into(), serde_json::Value::String(name)); + } + formatted_author = Some(display_with_mention); + } + } + + (metadata, formatted_author) +} + +/// Dispatch a fully-constructed `InboundMessage` to the inbound channel. +async fn send_inbound( + tx: &mpsc::Sender, + ts: String, + conversation_id: String, + sender_id: String, + content: MessageContent, + metadata: HashMap, + formatted_author: Option, +) { + let inbound = InboundMessage { + id: ts, + source: "slack".into(), + conversation_id, + sender_id, + agent_id: None, + content, + timestamp: chrono::Utc::now(), + metadata, + formatted_author, + }; + if let Err(error) = tx.send(inbound).await { + tracing::warn!(%error, "failed to send inbound message from Slack"); + } +} + +/// Deserialise a `Vec` into `Vec`. +/// +/// Blocks that fail to deserialise are silently skipped with a warning so a +/// single bad block doesn't kill the whole message. +fn deserialize_blocks(values: &[serde_json::Value]) -> Vec { + values + .iter() + .filter_map(|v| match serde_json::from_value::(v.clone()) { + Ok(block) => Some(block), + Err(err) => { + tracing::warn!(error = %err, "failed to deserialise slack block, skipping"); + None + } + }) + .collect() +} + +/// Strip the leading `<@BOT_USER_ID>` mention from an `app_mention` event text. +/// +/// Slack always formats user IDs in uppercase (e.g. `<@U012AB3CD>`), so a +/// simple prefix strip is sufficient — no case-folding is needed. +fn strip_bot_mention(text: &str, bot_user_id: &str) -> String { + let mention = format!("<@{}>", bot_user_id); + text.trim_start_matches(mention.as_str()).trim_start().to_string() +} + +/// Return a short human-readable name for an `OutboundResponse` variant for log messages. +fn variant_name(response: &OutboundResponse) -> &'static str { + match response { + OutboundResponse::Text(_) => "Text", + OutboundResponse::ThreadReply { .. } => "ThreadReply", + OutboundResponse::File { .. } => "File", + OutboundResponse::Reaction(_) => "Reaction", + OutboundResponse::RemoveReaction(_) => "RemoveReaction", + OutboundResponse::Ephemeral { .. } => "Ephemeral", + OutboundResponse::RichMessage { .. } => "RichMessage", + OutboundResponse::ScheduledMessage { .. } => "ScheduledMessage", + OutboundResponse::StreamStart => "StreamStart", + OutboundResponse::StreamChunk(_) => "StreamChunk", + OutboundResponse::StreamEnd => "StreamEnd", + OutboundResponse::Status(_) => "Status", + } +} + +/// Split a message into UTF-8-safe chunks at line/word boundaries. fn split_message(text: &str, max_len: usize) -> Vec { if text.len() <= max_len { return vec![text.to_string()]; @@ -701,10 +1050,16 @@ fn split_message(text: &str, max_len: usize) -> Vec { break; } - let split_at = remaining[..max_len] + // Walk back to a valid char boundary before slicing + let mut safe_max = max_len.min(remaining.len()); + while !remaining.is_char_boundary(safe_max) { + safe_max -= 1; + } + + let split_at = remaining[..safe_max] .rfind('\n') - .or_else(|| remaining[..max_len].rfind(' ')) - .unwrap_or(max_len); + .or_else(|| remaining[..safe_max].rfind(' ')) + .unwrap_or(safe_max); chunks.push(remaining[..split_at].to_string()); remaining = remaining[split_at..].trim_start(); @@ -713,7 +1068,7 @@ fn split_message(text: &str, max_len: usize) -> Vec { chunks } -/// Sanitize an emoji string for Slack reactions (remove colons, lowercase). +/// Sanitize an emoji name for Slack reactions (strip colons, lowercase). fn sanitize_reaction_name(emoji: &str) -> String { emoji .trim() @@ -723,21 +1078,14 @@ fn sanitize_reaction_name(emoji: &str) -> String { } fn resolve_slack_user_identity(user: &SlackUser, user_id: &str) -> SlackUserIdentity { - let username = user.name.clone().filter(|name| !name.trim().is_empty()); - + let username = user.name.clone().filter(|n| !n.trim().is_empty()); let display_name = user .profile .as_ref() - .and_then(|profile| { - profile - .display_name - .clone() - .or(profile.real_name.clone()) - }) - .filter(|name| !name.trim().is_empty()) + .and_then(|p| p.display_name.clone().or_else(|| p.real_name.clone())) + .filter(|n| !n.trim().is_empty()) .or_else(|| username.clone()) .unwrap_or_else(|| user_id.to_string()); - SlackUserIdentity { display_name, username, diff --git a/src/messaging/telegram.rs b/src/messaging/telegram.rs index 7b8895abb..631f71572 100644 --- a/src/messaging/telegram.rs +++ b/src/messaging/telegram.rs @@ -359,6 +359,26 @@ impl Messaging for TelegramAdapter { OutboundResponse::Status(status) => { self.send_status(message, status).await?; } + // Slack-specific variants — graceful fallbacks for Telegram + OutboundResponse::RemoveReaction(_) => {} // no-op + OutboundResponse::Ephemeral { text, .. } => { + // Telegram has no ephemeral messages — send as regular text + let chat_id = self.extract_chat_id(message)?; + self.bot.send_message(chat_id, text).await + .context("failed to send ephemeral fallback on telegram")?; + } + OutboundResponse::RichMessage { text, .. } => { + // No Block Kit on Telegram — plain text fallback + let chat_id = self.extract_chat_id(message)?; + self.bot.send_message(chat_id, text).await + .context("failed to send rich message fallback on telegram")?; + } + OutboundResponse::ScheduledMessage { text, .. } => { + // Telegram has no scheduled messages — send immediately + let chat_id = self.extract_chat_id(message)?; + self.bot.send_message(chat_id, text).await + .context("failed to send scheduled message fallback on telegram")?; + } } Ok(()) diff --git a/src/messaging/twitch.rs b/src/messaging/twitch.rs index 51ccae169..d1dea0f0b 100644 --- a/src/messaging/twitch.rs +++ b/src/messaging/twitch.rs @@ -270,8 +270,31 @@ impl Messaging for TwitchAdapter { // which sends a Text response after StreamEnd. } OutboundResponse::StreamEnd => {} - // Reactions and status updates aren't meaningful in Twitch chat - OutboundResponse::Reaction(_) | OutboundResponse::Status(_) => {} + // Reactions, status updates, and Slack-specific variants aren't meaningful in Twitch chat + OutboundResponse::Reaction(_) + | OutboundResponse::RemoveReaction(_) + | OutboundResponse::Status(_) => {} + OutboundResponse::Ephemeral { text, .. } => { + // No ephemeral concept in Twitch — send as regular chat message + client + .say(channel.to_owned(), text) + .await + .context("failed to send ephemeral fallback on twitch")?; + } + OutboundResponse::RichMessage { text, .. } => { + // No Block Kit on Twitch — plain text fallback + client + .say(channel.to_owned(), text) + .await + .context("failed to send rich message fallback on twitch")?; + } + OutboundResponse::ScheduledMessage { text, .. } => { + // No scheduled messages on Twitch — send immediately + client + .say(channel.to_owned(), text) + .await + .context("failed to send scheduled message fallback on twitch")?; + } } Ok(()) diff --git a/src/messaging/webhook.rs b/src/messaging/webhook.rs index f555fee98..e3e8c1d67 100644 --- a/src/messaging/webhook.rs +++ b/src/messaging/webhook.rs @@ -9,9 +9,9 @@ use crate::messaging::traits::{InboundStream, Messaging}; use crate::{InboundMessage, MessageContent, OutboundResponse}; use anyhow::Context as _; +use axum::Router; use axum::extract::{Json, State}; use axum::http::StatusCode; -use axum::Router; use axum::routing::{get, post}; use serde::{Deserialize, Serialize}; @@ -175,8 +175,29 @@ impl Messaging for WebhookAdapter { filename: None, caption: None, }, - // Reactions and status updates aren't meaningful over webhook - OutboundResponse::Reaction(_) | OutboundResponse::Status(_) => return Ok(()), + // Reactions, status updates, and remove-reaction aren't meaningful over webhook + OutboundResponse::Reaction(_) + | OutboundResponse::RemoveReaction(_) + | OutboundResponse::Status(_) => return Ok(()), + // Slack-specific rich variants — fall back to plain text + OutboundResponse::Ephemeral { text, .. } => WebhookResponse { + response_type: "text".into(), + content: Some(text), + filename: None, + caption: None, + }, + OutboundResponse::RichMessage { text, .. } => WebhookResponse { + response_type: "text".into(), + content: Some(text), + filename: None, + caption: None, + }, + OutboundResponse::ScheduledMessage { text, .. } => WebhookResponse { + response_type: "text".into(), + content: Some(text), + filename: None, + caption: None, + }, }; self.response_buffers