feat: Fetch Slack thread history for agent context

gateway/platforms/base.py

@@ -519,6 +519,16 @@ def set_message_handler(self, handler: MessageHandler) -> None:
         """
         self._message_handler = handler
 
+    def set_session_store(self, session_store: Any) -> None:
+        """
+        Set the session store for checking active sessions.
+
+        Used by adapters that need to check if a thread/conversation
+        has an active session before processing messages (e.g., Slack
+        thread replies without explicit mentions).
+        """
+        self._session_store = session_store
+
     @abstractmethod
     async def connect(self) -> bool:
         """

gateway/platforms/slack.py

@@ -84,6 +84,11 @@ def __init__(self, config: PlatformConfig):
         self._seen_messages: Dict[str, float] = {}
         self._SEEN_TTL = 300   # 5 minutes
         self._SEEN_MAX = 2000  # prune threshold
+        # Thread history cache: (channel_id, thread_ts) → list of messages
+        # Caches recent thread history to avoid repeated API calls
+        self._thread_history_cache: Dict[tuple, list] = {}
+        self._THREAD_HISTORY_TTL = 60  # 1 minute cache TTL
+        self._THREAD_HISTORY_MAX = 100  # max cached threads
 
     async def connect(self) -> bool:
         """Connect to Slack via Socket Mode."""
@@ -276,10 +281,13 @@ async def edit_message(
         if not self._app:
             return SendResult(success=False, error="Not connected")
         try:
+            # Convert standard markdown → Slack mrkdwn
+            formatted = self.format_message(content)
+
             await self._get_client(chat_id).chat_update(
                 channel=chat_id,
                 ts=message_id,
-                text=content,
+                text=formatted,
             )
             return SendResult(success=True, message_id=message_id)
         except Exception as e:  # pragma: no cover - defensive logging
@@ -763,11 +771,28 @@ async def _handle_slack_message(self, event: dict) -> None:
         else:
             thread_ts = event.get("thread_ts") or ts  # ts fallback for channels
 
-        # In channels, only respond if bot is mentioned
+        # In channels, only respond if bot is mentioned OR if this is a
+        # reply in a thread where the bot has an active session.
         bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id)
-        if not is_dm and bot_uid:
-            if f"<@{bot_uid}>" not in text:
+        is_mentioned = bot_uid and f"<@{bot_uid}>" in text
+
+        if not is_dm and bot_uid and not is_mentioned:
+            # Check if this is a thread reply (thread_ts exists and differs from ts)
+            event_thread_ts = event.get("thread_ts")
+            is_thread_reply = event_thread_ts and event_thread_ts != ts
+
+            if is_thread_reply and self._has_active_session_for_thread(
+                channel_id=channel_id,
+                thread_ts=event_thread_ts,
+                user_id=user_id,
+            ):
+                # Allow thread replies without mention if there's an active session
+                pass
+            else:
+                # Not a thread reply or no active session - ignore
                 return
+
+        if is_mentioned:
             # Strip the bot mention from the text
             text = text.replace(f"<@{bot_uid}>", "").strip()
 
@@ -862,6 +887,18 @@ async def _handle_slack_message(self, event: dict) -> None:
         # Resolve user display name (cached after first lookup)
         user_name = await self._resolve_user_name(user_id, chat_id=channel_id)
 
+        # Fetch and inject thread context if enabled and in a thread
+        include_thread_context = self.config.extra.get("include_thread_context", True)
+        if include_thread_context and thread_ts:
+            try:
+                thread_messages = await self._fetch_thread_history(channel_id, thread_ts)
+                if thread_messages:
+                    context_str = self._format_thread_context(thread_messages, ts)
+                    if context_str:
+                        text = f"{context_str}\n\n{text}"
+            except Exception as e:
+                logger.debug("[Slack] Failed to inject thread context: %s", e)
+
         # Build source
         source = self.build_source(
             chat_id=channel_id,
@@ -933,6 +970,194 @@ async def _handle_slash_command(self, command: dict) -> None:
 
         await self.handle_message(event)
 
+    async def _fetch_thread_history(
+        self,
+        channel_id: str,
+        thread_ts: str,
+        limit: int = 50,
+    ) -> list:
+        """Fetch message history from a Slack thread via conversations.replies.
+
+        Args:
+            channel_id: The Slack channel ID
+            thread_ts: The parent message timestamp (thread identifier)
+            limit: Maximum number of messages to fetch (excluding the parent)
+
+        Returns:
+            List of message dicts with user, text, and ts fields
+        """
+        if not self._app or not channel_id or not thread_ts:
+            return []
+
+        cache_key = (channel_id, thread_ts)
+        now = time.time()
+
+        # Check cache first
+        if cache_key in self._thread_history_cache:
+            cached = self._thread_history_cache[cache_key]
+            if cached and (now - cached[0].get("_cached_at", 0)) < self._THREAD_HISTORY_TTL:
+                logger.debug("[Slack] Using cached thread history for %s", thread_ts)
+                return cached[1:]  # Skip the cache metadata entry
+
+        try:
+            client = self._get_client(channel_id)
+            result = await client.conversations_replies(
+                channel=channel_id,
+                ts=thread_ts,
+                limit=limit + 1,  # +1 for the parent message
+            )
+
+            messages = result.get("messages", [])
+            if not messages:
+                return []
+
+            # Filter out bot messages and the parent message (first in list)
+            # Only include user messages
+            filtered = []
+            for msg in messages[1:]:  # Skip parent message
+                if msg.get("bot_id") or msg.get("subtype") == "bot_message":
+                    continue
+                # Skip message changes/deletions
+                if msg.get("subtype") in ("message_changed", "message_deleted"):
+                    continue
+                filtered.append({
+                    "user": msg.get("user", ""),
+                    "text": msg.get("text", ""),
+                    "ts": msg.get("ts", ""),
+                })
+
+            # Cache the result with metadata
+            cache_entry = [{"_cached_at": now}] + filtered
+            self._thread_history_cache[cache_key] = cache_entry
+
+            # Prune cache if needed
+            if len(self._thread_history_cache) > self._THREAD_HISTORY_MAX:
+                # Remove oldest entries
+                sorted_keys = sorted(
+                    self._thread_history_cache.keys(),
+                    key=lambda k: -(self._thread_history_cache[k][0].get("_cached_at", 0))
+                )
+                for old_key in sorted_keys[self._THREAD_HISTORY_MAX:]:
+                    del self._thread_history_cache[old_key]
+
+            logger.debug(
+                "[Slack] Fetched %d messages from thread %s",
+                len(filtered), thread_ts
+            )
+            return filtered
+
+        except Exception as e:
+            logger.warning("[Slack] Failed to fetch thread history: %s", e)
+            return []
+
+    def _format_thread_context(
+        self,
+        messages: list,
+        current_ts: str,
+        max_messages: int = 20,
+    ) -> str:
+        """Format thread history messages as context string.
+
+        Args:
+            messages: List of message dicts from _fetch_thread_history
+            current_ts: Timestamp of the current message to exclude
+            max_messages: Maximum messages to include in context
+
+        Returns:
+            Formatted context string or empty string if no context
+        """
+        if not messages:
+            return ""
+
+        # Filter out the current message and take most recent N
+        filtered = [m for m in messages if m.get("ts") != current_ts]
+        if not filtered:
+            return ""
+
+        # Take last N messages (most recent)
+        recent = filtered[-max_messages:]
+
+        lines = ["--- Thread Context ---"]
+        for msg in recent:
+            user_id = msg.get("user", "")
+            text = msg.get("text", "")
+            if not text:
+                continue
+
+            # Resolve user name
+            user_name = self._user_name_cache.get(user_id, user_id)
+
+            # Strip bot mentions from text
+            for bot_uid in self._team_bot_user_ids.values():
+                text = text.replace(f"<@{bot_uid}>", "").strip()
+
+            lines.append(f"[{user_name}]: {text}")
+
+        lines.append("--- End Context ---\n")
+        return "\n".join(lines)
+
+    def _has_active_session_for_thread(
+        self,
+        channel_id: str,
+        thread_ts: str,
+        user_id: str,
+    ) -> bool:
+        """Check if there's an active session for a thread.
+
+        Used to determine if thread replies without @mentions should be
+        processed (they should if there's an active session).
+
+        Args:
+            channel_id: The Slack channel ID
+            thread_ts: The thread timestamp (parent message ts)
+            user_id: The user ID of the sender
+
+        Returns:
+            True if there's an active session for this thread
+        """
+        session_store = getattr(self, "_session_store", None)
+        if not session_store:
+            return False
+
+        try:
+            # Build a SessionSource for this thread
+            from gateway.session import SessionSource
+            from gateway.config import Platform
+
+            source = SessionSource(
+                platform=Platform.SLACK,
+                chat_id=channel_id,
+                chat_type="group",
+                user_id=user_id,
+                thread_id=thread_ts,
+            )
+
+            # Generate the session key using the same logic as SessionStore
+            # This mirrors the logic in build_session_key for group sessions
+            key_parts = ["agent:main", "slack", "group", channel_id, thread_ts]
+
+            # Include user_id if group_sessions_per_user is enabled
+            # We check the session store config if available
+            group_sessions_per_user = getattr(
+                session_store, "config", {}
+            )
+            if hasattr(group_sessions_per_user, "group_sessions_per_user"):
+                group_sessions_per_user = group_sessions_per_user.group_sessions_per_user
+            else:
+                group_sessions_per_user = True  # Default
+
+            if group_sessions_per_user and user_id:
+                key_parts.append(str(user_id))
+
+            session_key = ":".join(key_parts)
+
+            # Check if the session exists in the store
+            session_store._ensure_loaded()
+            return session_key in session_store._entries
+        except Exception:
+            # If anything goes wrong, default to False (require mention)
+            return False
+
     async def _download_slack_file(self, url: str, ext: str, audio: bool = False, team_id: str = "") -> str:
         """Download a Slack file using the bot token for auth, with retry."""
         import asyncio

gateway/run.py

@@ -1127,6 +1127,7 @@ async def start(self) -> bool:
             # Set up message + fatal error handlers
             adapter.set_message_handler(self._handle_message)
             adapter.set_fatal_error_handler(self._handle_adapter_fatal_error)
+            adapter.set_session_store(self.session_store)
 
             # Try to connect
             logger.info("Connecting to %s...", platform.value)
@@ -1424,6 +1425,7 @@ async def _platform_reconnect_watcher(self) -> None:
 
                     adapter.set_message_handler(self._handle_message)
                     adapter.set_fatal_error_handler(self._handle_adapter_fatal_error)
+                    adapter.set_session_store(self.session_store)
 
                     success = await adapter.connect()
                     if success: