feat(gateway): unified thread context hook in BasePlatformAdapter

gateway/platforms/base.py

@@ -1778,13 +1778,76 @@ def set_busy_session_handler(self, handler: Optional[Callable[[MessageEvent, str
     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
-
+
+    def has_active_session_for_event(self, event: "MessageEvent") -> bool:
+        """Check if a persistent session already exists for this event's thread/chat.
+
+        Uses ``build_session_key()`` as the single source of truth for key
+        construction, reading session-isolation flags from the adapter config.
+        Also evaluates the store's reset policy — a session that would be
+        auto-reset (idle timeout, daily reset) on the next interaction is
+        treated as non-existent so that thread context is still seeded.
+
+        Subclasses can override to customise the ``SessionSource`` (e.g. force
+        ``chat_type="group"``), but the default implementation covers most
+        platforms.
+        """
+        session_store = getattr(self, "_session_store", None)
+        if not session_store:
+            return False
+
+        try:
+            store_cfg = getattr(session_store, "config", None)
+            gspu = getattr(store_cfg, "group_sessions_per_user", True) if store_cfg else True
+            tspu = getattr(store_cfg, "thread_sessions_per_user", False) if store_cfg else False
+
+            session_key = build_session_key(
+                event.source,
+                group_sessions_per_user=gspu,
+                thread_sessions_per_user=tspu,
+            )
+            session_store._ensure_loaded()
+            entry = session_store._entries.get(session_key)
+            if entry is None:
+                return False
+
+            # Check whether the session would be auto-reset (idle/daily
+            # policy).  If so, the gateway will create a fresh session on
+            # the next interaction — treat as non-existent so thread context
+            # is still seeded into the new session.
+            _should_reset = getattr(session_store, "_should_reset", None)
+            if _should_reset and event.source:
+                if _should_reset(entry, event.source):
+                    return False
+
+            return True
+        except Exception:
+            return False
+
+    async def fetch_thread_context(self, event: "MessageEvent") -> Optional[str]:
+        """Fetch platform-specific thread/conversation context for first-time
+        thread entry.
+
+        Override in subclasses to fetch prior messages from the platform API
+        when the bot is first mentioned in an existing thread.  Return a
+        formatted context string to prepend to ``event.text``, or ``None``.
+
+        Implementations should:
+        1. Determine whether this event warrants context fetching (e.g. it is
+           a thread reply, not a brand-new thread the bot just created).
+        2. Call ``self.has_active_session_for_event(event)`` — return ``None``
+           if a session already exists (the session transcript already holds
+           prior messages).
+        3. Fetch and format prior messages via the platform's native API.
+        """
+        return None
+
     @abstractmethod
     async def connect(self) -> bool:
         """
@@ -3538,6 +3601,16 @@ async def _stop_typing_task() -> None:
         try:
             await self._run_processing_hook("on_processing_start", event)
 
+            # Fetch platform-specific thread context for first-time thread
+            # entry.  Each adapter overrides fetch_thread_context() to call
+            # its native API; the default returns None (no-op).
+            # Skip for commands — prepending context to "/reset" etc. would
+            # break command parsing in the gateway runner.
+            if not event.is_command():
+                thread_context = await self.fetch_thread_context(event)
+                if thread_context:
+                    event.text = thread_context + event.text
+
             # Call the handler (this can take a while with tool calls)
             response = await self._message_handler(event)
 

gateway/platforms/slack.py

@@ -2008,21 +2008,8 @@ async def _handle_slack_message(self, event: dict) -> None:
                     for t in to_remove:
                         self._mentioned_threads.discard(t)
 
-        # When entering a thread for the first time (no existing session),
-        # fetch thread context so the agent understands the conversation.
-        if is_thread_reply and not self._has_active_session_for_thread(
-            channel_id=channel_id,
-            thread_ts=event_thread_ts,
-            user_id=user_id,
-        ):
-            thread_context = await self._fetch_thread_context(
-                channel_id=channel_id,
-                thread_ts=event_thread_ts,
-                current_ts=ts,
-                team_id=team_id,
-            )
-            if thread_context:
-                text = thread_context + text
+        # Thread context fetching is now handled by the base class via
+        # fetch_thread_context() called in _process_message_background().
 
         # Determine message type
         msg_type = MessageType.TEXT
@@ -2754,6 +2741,36 @@ async def _fetch_thread_parent_text(
             logger.debug("[Slack] Failed to fetch thread parent text: %s", exc)
             return ""
 
+    async def fetch_thread_context(self, event: "MessageEvent") -> Optional[str]:
+        """Fetch Slack thread context on first-time thread entry.
+
+        Overrides ``BasePlatformAdapter.fetch_thread_context()`` so that the
+        base class call site in ``_process_message_background()`` automatically
+        prepends thread history for Slack threads.
+        """
+        raw = event.raw_message
+        if not isinstance(raw, dict):
+            return None
+
+        ts = raw.get("ts", "")
+        event_thread_ts = raw.get("thread_ts")
+        is_thread_reply = bool(event_thread_ts and event_thread_ts != ts)
+        if not is_thread_reply:
+            return None
+
+        if self.has_active_session_for_event(event):
+            return None
+
+        team_id = raw.get("team", "")
+        channel_id = raw.get("channel", event.source.chat_id if event.source else "")
+        context = await self._fetch_thread_context(
+            channel_id=channel_id,
+            thread_ts=event_thread_ts,
+            current_ts=ts,
+            team_id=team_id,
+        )
+        return context or None
+
     async def _handle_slash_command(self, command: dict) -> None:
         """Handle Slack slash commands.
 
@@ -2846,46 +2863,22 @@ def _has_active_session_for_thread(
         thread_ts: str,
         user_id: str,
     ) -> bool:
-        """Check if there's an active session for a thread.
+        """Check if there's a live (non-expired) session for a thread.
 
         Used to determine if thread replies without @mentions should be
         processed (they should if there's an active session).
 
-        Uses ``build_session_key()`` as the single source of truth for key
-        construction — avoids the bug where manual key building didn't
-        respect ``thread_sessions_per_user`` and ``group_sessions_per_user``
-        settings correctly.
+        Delegates to ``has_active_session_for_event()`` which checks both
+        key presence AND the store's reset policy (idle/daily expiry).
         """
-        session_store = getattr(self, "_session_store", None)
-        if not session_store:
-            return False
-
-        try:
-            from gateway.session import SessionSource, build_session_key
-
-            source = SessionSource(
-                platform=Platform.SLACK,
-                chat_id=channel_id,
-                chat_type="group",
-                user_id=user_id,
-                thread_id=thread_ts,
-            )
-
-            # Read session isolation settings from the store's config
-            store_cfg = getattr(session_store, "config", None)
-            gspu = getattr(store_cfg, "group_sessions_per_user", True) if store_cfg else True
-            tspu = getattr(store_cfg, "thread_sessions_per_user", False) if store_cfg else False
-
-            session_key = build_session_key(
-                source,
-                group_sessions_per_user=gspu,
-                thread_sessions_per_user=tspu,
-            )
-
-            session_store._ensure_loaded()
-            return session_key in session_store._entries
-        except Exception:
-            return False
+        source = self.build_source(
+            chat_id=channel_id,
+            chat_type="group",
+            user_id=user_id,
+            thread_id=thread_ts,
+        )
+        event = MessageEvent(text="", source=source)
+        return self.has_active_session_for_event(event)
 
     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."""

plugins/platforms/discord/adapter.py

@@ -4450,6 +4450,62 @@ async def _cache_discord_document(self, att, ext: str) -> bytes:
                     raise Exception(f"HTTP {resp.status}")
                 return await resp.read()
 
+    async def fetch_thread_context(self, event: "MessageEvent") -> Optional[str]:
+        """Fetch Discord thread history on first-time thread entry.
+
+        Overrides ``BasePlatformAdapter.fetch_thread_context()`` so that the
+        base class call site in ``_process_message_background()`` automatically
+        prepends thread history for Discord threads.
+
+        Skips auto-created threads (no history to fetch) and DMs.
+        """
+        message = event.raw_message
+        if message is None:
+            return None
+
+        # Only fetch for pre-existing threads — not DMs or auto-created ones.
+        channel = getattr(message, "channel", None)
+        if channel is None or not isinstance(channel, discord.Thread):
+            return None
+
+        if self.has_active_session_for_event(event):
+            return None
+
+        try:
+            context_parts: list[str] = []
+            bot_user = self._client.user
+            # oldest_first=True gives chronological order
+            async for msg in channel.history(limit=30, oldest_first=True):
+                # Skip the triggering message itself
+                if msg.id == message.id:
+                    continue
+                # Skip bot's own messages to avoid circular context
+                if bot_user and msg.author.id == bot_user.id:
+                    continue
+                msg_text = (msg.content or "").strip()
+                if not msg_text:
+                    continue
+                # Strip bot @mentions from context messages
+                if bot_user:
+                    msg_text = msg_text.replace(f"<@{bot_user.id}>", "").strip()
+                    msg_text = msg_text.replace(f"<@!{bot_user.id}>", "").strip()
+                if not msg_text:
+                    continue
+                name = getattr(msg.author, "display_name", None) or msg.author.name
+                context_parts.append(f"{name}: {msg_text}")
+
+            if not context_parts:
+                return None
+
+            return (
+                "[Thread context \u2014 prior messages in this thread (not yet in conversation history):]\n"
+                + "\n".join(context_parts)
+                + "\n[End of thread context]\n\n"
+            )
+        except Exception as e:
+            logger.warning("[Discord] Failed to fetch thread context: %s", e)
+            return None
+
     async def _handle_message(self, message: DiscordMessage) -> None:
         """Handle incoming Discord messages."""
         # In server channels (not DMs), require the bot to be @mentioned

tests/gateway/test_slack_approval_buttons.py

@@ -518,6 +518,7 @@ def test_uses_build_session_key(self):
         mock_store.config = MagicMock()
         mock_store.config.group_sessions_per_user = False  # threads don't include user_id
         mock_store.config.thread_sessions_per_user = False
+        mock_store._should_reset = MagicMock(return_value=None)  # session is live
         adapter._session_store = mock_store
 
         # With the fix, build_session_key should be called which respects
@@ -538,6 +539,7 @@ def test_no_session_returns_false(self):
         mock_store.config = MagicMock()
         mock_store.config.group_sessions_per_user = True
         mock_store.config.thread_sessions_per_user = False
+        mock_store._should_reset = MagicMock(return_value=None)
         adapter._session_store = mock_store
 
         result = adapter._has_active_session_for_thread(