| 1 | """Slack channel implementation using Socket Mode.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import re |
| 5 | from typing import Any |
| 6 | |
| 7 | from loguru import logger |
| 8 | from pydantic import Field |
| 9 | from slack_sdk.socket_mode.request import SocketModeRequest |
| 10 | from slack_sdk.socket_mode.response import SocketModeResponse |
| 11 | from slack_sdk.socket_mode.websockets import SocketModeClient |
| 12 | from slack_sdk.web.async_client import AsyncWebClient |
| 13 | from slackify_markdown import slackify_markdown |
| 14 | |
| 15 | from nanobot.bus.events import OutboundMessage |
| 16 | from nanobot.bus.queue import MessageBus |
| 17 | from nanobot.channels.base import BaseChannel |
| 18 | from nanobot.config.schema import Base |
| 19 | |
| 20 | |
| 21 | class SlackDMConfig(Base): |
| 22 | """Slack DM policy configuration.""" |
| 23 | |
| 24 | enabled: bool = True |
| 25 | policy: str = "open" |
| 26 | allow_from: list[str] = Field(default_factory=list) |
| 27 | |
| 28 | |
| 29 | class SlackConfig(Base): |
| 30 | """Slack channel configuration.""" |
| 31 | |
| 32 | enabled: bool = False |
| 33 | mode: str = "socket" |
| 34 | webhook_path: str = "/slack/events" |
| 35 | bot_token: str = "" |
| 36 | app_token: str = "" |
| 37 | user_token_read_only: bool = True |
| 38 | reply_in_thread: bool = True |
| 39 | react_emoji: str = "eyes" |
| 40 | done_emoji: str = "white_check_mark" |
| 41 | allow_from: list[str] = Field(default_factory=list) |
| 42 | group_policy: str = "mention" |
| 43 | group_allow_from: list[str] = Field(default_factory=list) |
| 44 | dm: SlackDMConfig = Field(default_factory=SlackDMConfig) |
| 45 | |
| 46 | |
| 47 | class SlackChannel(BaseChannel): |
| 48 | """Slack channel using Socket Mode.""" |
| 49 | |
| 50 | name = "slack" |
| 51 | display_name = "Slack" |
| 52 | _SLACK_ID_RE = re.compile(r"^[CDGUW][A-Z0-9]{2,}$") |
| 53 | _SLACK_CHANNEL_REF_RE = re.compile(r"^<#([A-Z0-9]+)(?:\|[^>]+)?>$") |
| 54 | _SLACK_USER_REF_RE = re.compile(r"^<@([A-Z0-9]+)(?:\|[^>]+)?>$") |
| 55 | |
| 56 | @classmethod |
| 57 | def default_config(cls) -> dict[str, Any]: |
| 58 | return SlackConfig().model_dump(by_alias=True) |
| 59 | |
| 60 | def __init__(self, config: Any, bus: MessageBus): |
| 61 | if isinstance(config, dict): |
| 62 | config = SlackConfig.model_validate(config) |
| 63 | super().__init__(config, bus) |
| 64 | self.config: SlackConfig = config |
| 65 | self._web_client: AsyncWebClient | None = None |
| 66 | self._socket_client: SocketModeClient | None = None |
| 67 | self._bot_user_id: str | None = None |
| 68 | self._target_cache: dict[str, str] = {} |
| 69 | |
| 70 | async def start(self) -> None: |
| 71 | """Start the Slack Socket Mode client.""" |
| 72 | if not self.config.bot_token or not self.config.app_token: |
| 73 | logger.error("Slack bot/app token not configured") |
| 74 | return |
| 75 | if self.config.mode != "socket": |
| 76 | logger.error("Unsupported Slack mode: {}", self.config.mode) |
| 77 | return |
| 78 | |
| 79 | self._running = True |
| 80 | |
| 81 | self._web_client = AsyncWebClient(token=self.config.bot_token) |
| 82 | self._socket_client = SocketModeClient( |
| 83 | app_token=self.config.app_token, |
| 84 | web_client=self._web_client, |
| 85 | ) |
| 86 | |
| 87 | self._socket_client.socket_mode_request_listeners.append(self._on_socket_request) |
| 88 | |
| 89 | # Resolve bot user ID for mention handling |
| 90 | try: |
| 91 | auth = await self._web_client.auth_test() |
| 92 | self._bot_user_id = auth.get("user_id") |
| 93 | logger.info("Slack bot connected as {}", self._bot_user_id) |
| 94 | except Exception as e: |
| 95 | logger.warning("Slack auth_test failed: {}", e) |
| 96 | |
| 97 | logger.info("Starting Slack Socket Mode client...") |
| 98 | await self._socket_client.connect() |
| 99 | |
| 100 | while self._running: |
| 101 | await asyncio.sleep(1) |
| 102 | |
| 103 | async def stop(self) -> None: |
| 104 | """Stop the Slack client.""" |
| 105 | self._running = False |
| 106 | if self._socket_client: |
| 107 | try: |
| 108 | await self._socket_client.close() |
| 109 | except Exception as e: |
| 110 | logger.warning("Slack socket close failed: {}", e) |
| 111 | self._socket_client = None |
| 112 | |
| 113 | async def send(self, msg: OutboundMessage) -> None: |
| 114 | """Send a message through Slack.""" |
| 115 | if not self._web_client: |
| 116 | logger.warning("Slack client not running") |
| 117 | return |
| 118 | try: |
| 119 | target_chat_id = await self._resolve_target_chat_id(msg.chat_id) |
| 120 | slack_meta = msg.metadata.get("slack", {}) if msg.metadata else {} |
| 121 | thread_ts = slack_meta.get("thread_ts") |
| 122 | channel_type = slack_meta.get("channel_type") |
| 123 | origin_chat_id = str((slack_meta.get("event", {}) or {}).get("channel") or msg.chat_id) |
| 124 | # Slack DMs don't use threads; channel/group replies may keep thread_ts. |
| 125 | thread_ts_param = ( |
| 126 | thread_ts |
| 127 | if thread_ts and channel_type != "im" and target_chat_id == origin_chat_id |
| 128 | else None |
| 129 | ) |
| 130 | |
| 131 | # Slack rejects empty text payloads. Keep media-only messages media-only, |
| 132 | # but send a single blank message when the bot has no text or files to send. |
| 133 | if msg.content or not (msg.media or []): |
| 134 | await self._web_client.chat_postMessage( |
| 135 | channel=target_chat_id, |
| 136 | text=self._to_mrkdwn(msg.content) if msg.content else " ", |
| 137 | thread_ts=thread_ts_param, |
| 138 | ) |
| 139 | |
| 140 | for media_path in msg.media or []: |
| 141 | try: |
| 142 | await self._web_client.files_upload_v2( |
| 143 | channel=target_chat_id, |
| 144 | file=media_path, |
| 145 | thread_ts=thread_ts_param, |
| 146 | ) |
| 147 | except Exception as e: |
| 148 | logger.error("Failed to upload file {}: {}", media_path, e) |
| 149 | |
| 150 | # Update reaction emoji when the final (non-progress) response is sent |
| 151 | if not (msg.metadata or {}).get("_progress"): |
| 152 | event = slack_meta.get("event", {}) |
| 153 | await self._update_react_emoji(origin_chat_id, event.get("ts")) |
| 154 | |
| 155 | except Exception as e: |
| 156 | logger.error("Error sending Slack message: {}", e) |
| 157 | raise |
| 158 | |
| 159 | async def _resolve_target_chat_id(self, target: str) -> str: |
| 160 | """Resolve human-friendly Slack targets to concrete IDs when needed.""" |
| 161 | if not self._web_client: |
| 162 | return target |
| 163 | |
| 164 | target = target.strip() |
| 165 | if not target: |
| 166 | return target |
| 167 | |
| 168 | if match := self._SLACK_CHANNEL_REF_RE.fullmatch(target): |
| 169 | return match.group(1) |
| 170 | if match := self._SLACK_USER_REF_RE.fullmatch(target): |
| 171 | return await self._open_dm_for_user(match.group(1)) |
| 172 | if self._SLACK_ID_RE.fullmatch(target): |
| 173 | if target.startswith(("U", "W")): |
| 174 | return await self._open_dm_for_user(target) |
| 175 | return target |
| 176 | |
| 177 | if target.startswith("#"): |
| 178 | return await self._resolve_channel_name(target[1:]) |
| 179 | if target.startswith("@"): |
| 180 | return await self._resolve_user_handle(target[1:]) |
| 181 | |
| 182 | try: |
| 183 | return await self._resolve_channel_name(target) |
| 184 | except ValueError: |
| 185 | return await self._resolve_user_handle(target) |
| 186 | |
| 187 | async def _resolve_channel_name(self, name: str) -> str: |
| 188 | normalized = self._normalize_target_name(name) |
| 189 | if not normalized: |
| 190 | raise ValueError("Slack target channel name is empty") |
| 191 | |
| 192 | cache_key = f"channel:{normalized}" |
| 193 | if cache_key in self._target_cache: |
| 194 | return self._target_cache[cache_key] |
| 195 | |
| 196 | cursor: str | None = None |
| 197 | while True: |
| 198 | response = await self._web_client.conversations_list( |
| 199 | types="public_channel,private_channel", |
| 200 | exclude_archived=True, |
| 201 | limit=200, |
| 202 | cursor=cursor, |
| 203 | ) |
| 204 | for channel in response.get("channels", []): |
| 205 | if self._normalize_target_name(str(channel.get("name") or "")) == normalized: |
| 206 | channel_id = str(channel.get("id") or "") |
| 207 | if channel_id: |
| 208 | self._target_cache[cache_key] = channel_id |
| 209 | return channel_id |
| 210 | cursor = ((response.get("response_metadata") or {}).get("next_cursor") or "").strip() |
| 211 | if not cursor: |
| 212 | break |
| 213 | |
| 214 | raise ValueError( |
| 215 | f"Slack channel '{name}' was not found. Use a joined channel name like " |
| 216 | f"'#general' or a concrete channel ID." |
| 217 | ) |
| 218 | |
| 219 | async def _resolve_user_handle(self, handle: str) -> str: |
| 220 | normalized = self._normalize_target_name(handle) |
| 221 | if not normalized: |
| 222 | raise ValueError("Slack target user handle is empty") |
| 223 | |
| 224 | cache_key = f"user:{normalized}" |
| 225 | if cache_key in self._target_cache: |
| 226 | return self._target_cache[cache_key] |
| 227 | |
| 228 | cursor: str | None = None |
| 229 | while True: |
| 230 | response = await self._web_client.users_list(limit=200, cursor=cursor) |
| 231 | for member in response.get("members", []): |
| 232 | if self._member_matches_handle(member, normalized): |
| 233 | user_id = str(member.get("id") or "") |
| 234 | if not user_id: |
| 235 | continue |
| 236 | dm_id = await self._open_dm_for_user(user_id) |
| 237 | self._target_cache[cache_key] = dm_id |
| 238 | return dm_id |
| 239 | cursor = ((response.get("response_metadata") or {}).get("next_cursor") or "").strip() |
| 240 | if not cursor: |
| 241 | break |
| 242 | |
| 243 | raise ValueError( |
| 244 | f"Slack user '{handle}' was not found. Use '@name' or a concrete DM/channel ID." |
| 245 | ) |
| 246 | |
| 247 | async def _open_dm_for_user(self, user_id: str) -> str: |
| 248 | response = await self._web_client.conversations_open(users=user_id) |
| 249 | channel_id = str(((response.get("channel") or {}).get("id")) or "") |
| 250 | if not channel_id: |
| 251 | raise ValueError(f"Slack DM target for user '{user_id}' could not be opened.") |
| 252 | return channel_id |
| 253 | |
| 254 | @staticmethod |
| 255 | def _normalize_target_name(value: str) -> str: |
| 256 | return value.strip().lstrip("#@").lower() |
| 257 | |
| 258 | @classmethod |
| 259 | def _member_matches_handle(cls, member: dict[str, Any], normalized: str) -> bool: |
| 260 | profile = member.get("profile") or {} |
| 261 | candidates = { |
| 262 | str(member.get("name") or ""), |
| 263 | str(profile.get("display_name") or ""), |
| 264 | str(profile.get("display_name_normalized") or ""), |
| 265 | str(profile.get("real_name") or ""), |
| 266 | str(profile.get("real_name_normalized") or ""), |
| 267 | } |
| 268 | return normalized in {cls._normalize_target_name(candidate) for candidate in candidates if candidate} |
| 269 | |
| 270 | async def _on_socket_request( |
| 271 | self, |
| 272 | client: SocketModeClient, |
| 273 | req: SocketModeRequest, |
| 274 | ) -> None: |
| 275 | """Handle incoming Socket Mode requests.""" |
| 276 | if req.type != "events_api": |
| 277 | return |
| 278 | |
| 279 | # Acknowledge right away |
| 280 | await client.send_socket_mode_response( |
| 281 | SocketModeResponse(envelope_id=req.envelope_id) |
| 282 | ) |
| 283 | |
| 284 | payload = req.payload or {} |
| 285 | event = payload.get("event") or {} |
| 286 | event_type = event.get("type") |
| 287 | |
| 288 | # Handle app mentions or plain messages |
| 289 | if event_type not in ("message", "app_mention"): |
| 290 | return |
| 291 | |
| 292 | sender_id = event.get("user") |
| 293 | chat_id = event.get("channel") |
| 294 | |
| 295 | # Ignore bot/system messages (any subtype = not a normal user message) |
| 296 | if event.get("subtype"): |
| 297 | return |
| 298 | if self._bot_user_id and sender_id == self._bot_user_id: |
| 299 | return |
| 300 | |
| 301 | # Avoid double-processing: Slack sends both `message` and `app_mention` |
| 302 | # for mentions in channels. Prefer `app_mention`. |
| 303 | text = event.get("text") or "" |
| 304 | if event_type == "message" and self._bot_user_id and f"<@{self._bot_user_id}>" in text: |
| 305 | return |
| 306 | |
| 307 | # Debug: log basic event shape |
| 308 | logger.debug( |
| 309 | "Slack event: type={} subtype={} user={} channel={} channel_type={} text={}", |
| 310 | event_type, |
| 311 | event.get("subtype"), |
| 312 | sender_id, |
| 313 | chat_id, |
| 314 | event.get("channel_type"), |
| 315 | text[:80], |
| 316 | ) |
| 317 | if not sender_id or not chat_id: |
| 318 | return |
| 319 | |
| 320 | channel_type = event.get("channel_type") or "" |
| 321 | |
| 322 | if not self._is_allowed(sender_id, chat_id, channel_type): |
| 323 | return |
| 324 | |
| 325 | if channel_type != "im" and not self._should_respond_in_channel(event_type, text, chat_id): |
| 326 | return |
| 327 | |
| 328 | text = self._strip_bot_mention(text) |
| 329 | |
| 330 | thread_ts = event.get("thread_ts") |
| 331 | if self.config.reply_in_thread and not thread_ts: |
| 332 | thread_ts = event.get("ts") |
| 333 | # Add :eyes: reaction to the triggering message (best-effort) |
| 334 | try: |
| 335 | if self._web_client and event.get("ts"): |
| 336 | await self._web_client.reactions_add( |
| 337 | channel=chat_id, |
| 338 | name=self.config.react_emoji, |
| 339 | timestamp=event.get("ts"), |
| 340 | ) |
| 341 | except Exception as e: |
| 342 | logger.debug("Slack reactions_add failed: {}", e) |
| 343 | |
| 344 | # Thread-scoped session key for channel/group messages |
| 345 | session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts and channel_type != "im" else None |
| 346 | |
| 347 | try: |
| 348 | await self._handle_message( |
| 349 | sender_id=sender_id, |
| 350 | chat_id=chat_id, |
| 351 | content=text, |
| 352 | metadata={ |
| 353 | "slack": { |
| 354 | "event": event, |
| 355 | "thread_ts": thread_ts, |
| 356 | "channel_type": channel_type, |
| 357 | }, |
| 358 | }, |
| 359 | session_key=session_key, |
| 360 | ) |
| 361 | except Exception: |
| 362 | logger.exception("Error handling Slack message from {}", sender_id) |
| 363 | |
| 364 | async def _update_react_emoji(self, chat_id: str, ts: str | None) -> None: |
| 365 | """Remove the in-progress reaction and optionally add a done reaction.""" |
| 366 | if not self._web_client or not ts: |
| 367 | return |
| 368 | try: |
| 369 | await self._web_client.reactions_remove( |
| 370 | channel=chat_id, |
| 371 | name=self.config.react_emoji, |
| 372 | timestamp=ts, |
| 373 | ) |
| 374 | except Exception as e: |
| 375 | logger.debug("Slack reactions_remove failed: {}", e) |
| 376 | if self.config.done_emoji: |
| 377 | try: |
| 378 | await self._web_client.reactions_add( |
| 379 | channel=chat_id, |
| 380 | name=self.config.done_emoji, |
| 381 | timestamp=ts, |
| 382 | ) |
| 383 | except Exception as e: |
| 384 | logger.debug("Slack done reaction failed: {}", e) |
| 385 | |
| 386 | def _is_allowed(self, sender_id: str, chat_id: str, channel_type: str) -> bool: |
| 387 | if channel_type == "im": |
| 388 | if not self.config.dm.enabled: |
| 389 | return False |
| 390 | if self.config.dm.policy == "allowlist": |
| 391 | return sender_id in self.config.dm.allow_from |
| 392 | return True |
| 393 | |
| 394 | # Group / channel messages |
| 395 | if self.config.group_policy == "allowlist": |
| 396 | return chat_id in self.config.group_allow_from |
| 397 | return True |
| 398 | |
| 399 | def _should_respond_in_channel(self, event_type: str, text: str, chat_id: str) -> bool: |
| 400 | if self.config.group_policy == "open": |
| 401 | return True |
| 402 | if self.config.group_policy == "mention": |
| 403 | if event_type == "app_mention": |
| 404 | return True |
| 405 | return self._bot_user_id is not None and f"<@{self._bot_user_id}>" in text |
| 406 | if self.config.group_policy == "allowlist": |
| 407 | return chat_id in self.config.group_allow_from |
| 408 | return False |
| 409 | |
| 410 | def _strip_bot_mention(self, text: str) -> str: |
| 411 | if not text or not self._bot_user_id: |
| 412 | return text |
| 413 | return re.sub(rf"<@{re.escape(self._bot_user_id)}>\s*", "", text).strip() |
| 414 | |
| 415 | _TABLE_RE = re.compile(r"(?m)^\|.*\|$(?:\n\|[\s:|-]*\|$)(?:\n\|.*\|$)*") |
| 416 | _CODE_FENCE_RE = re.compile(r"```[\s\S]*?```") |
| 417 | _INLINE_CODE_RE = re.compile(r"`[^`]+`") |
| 418 | _LEFTOVER_BOLD_RE = re.compile(r"\*\*(.+?)\*\*") |
| 419 | _LEFTOVER_HEADER_RE = re.compile(r"^#{1,6}\s+(.+)$", re.MULTILINE) |
| 420 | _BARE_URL_RE = re.compile(r"(?<![|<])(https?://\S+)") |
| 421 | |
| 422 | @classmethod |
| 423 | def _to_mrkdwn(cls, text: str) -> str: |
| 424 | """Convert Markdown to Slack mrkdwn, including tables.""" |
| 425 | if not text: |
| 426 | return "" |
| 427 | text = cls._TABLE_RE.sub(cls._convert_table, text) |
| 428 | return cls._fixup_mrkdwn(slackify_markdown(text)) |
| 429 | |
| 430 | @classmethod |
| 431 | def _fixup_mrkdwn(cls, text: str) -> str: |
| 432 | """Fix markdown artifacts that slackify_markdown misses.""" |
| 433 | code_blocks: list[str] = [] |
| 434 | |
| 435 | def _save_code(m: re.Match) -> str: |
| 436 | code_blocks.append(m.group(0)) |
| 437 | return f"\x00CB{len(code_blocks) - 1}\x00" |
| 438 | |
| 439 | text = cls._CODE_FENCE_RE.sub(_save_code, text) |
| 440 | text = cls._INLINE_CODE_RE.sub(_save_code, text) |
| 441 | text = cls._LEFTOVER_BOLD_RE.sub(r"*\1*", text) |
| 442 | text = cls._LEFTOVER_HEADER_RE.sub(r"*\1*", text) |
| 443 | text = cls._BARE_URL_RE.sub(lambda m: m.group(0).replace("&", "&"), text) |
| 444 | |
| 445 | for i, block in enumerate(code_blocks): |
| 446 | text = text.replace(f"\x00CB{i}\x00", block) |
| 447 | return text |
| 448 | |
| 449 | @staticmethod |
| 450 | def _convert_table(match: re.Match) -> str: |
| 451 | """Convert a Markdown table to a Slack-readable list.""" |
| 452 | lines = [ln.strip() for ln in match.group(0).strip().splitlines() if ln.strip()] |
| 453 | if len(lines) < 2: |
| 454 | return match.group(0) |
| 455 | headers = [h.strip() for h in lines[0].strip("|").split("|")] |
| 456 | start = 2 if re.fullmatch(r"[|\s:\-]+", lines[1]) else 1 |
| 457 | rows: list[str] = [] |
| 458 | for line in lines[start:]: |
| 459 | cells = [c.strip() for c in line.strip("|").split("|")] |
| 460 | cells = (cells + [""] * len(headers))[: len(headers)] |
| 461 | parts = [f"**{headers[i]}**: {cells[i]}" for i in range(len(headers)) if cells[i]] |
| 462 | if parts: |
| 463 | rows.append(" · ".join(parts)) |
| 464 | return "\n".join(rows) |
| 465 |