diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index 4ca7e135e0c2..419ceccf9064 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -84,6 +84,12 @@ class _ThreadContextCache: fetched_at: float = field(default_factory=time.monotonic) message_count: int = 0 parent_text: str = "" # Raw text of the thread parent (for reply_to_text injection) + # The Slack user_id of the thread parent message author. Used by + # _bot_authored_thread_root (#63530) to detect threads whose root was + # posted by the bot via direct chat.postMessage (outside the gateway's + # send() path). Empty string when the parent could not be fetched or + # did not have a user_id field. + parent_user_id: str = "" # Raw Slack reply payloads from conversations.replies. Kept so context can # be re-formatted with a different watermark (``after_ts``) without an # extra API call (#23918). @@ -3526,6 +3532,103 @@ class SlackAdapter(BasePlatformAdapter): fallback_event["thread_ts"] = thread_ts await self._handle_slack_message(fallback_event) + async def _bot_authored_thread_root( + self, channel_id: str, thread_ts: str, team_id: str = "" + ) -> bool: + """Return True when the thread root was authored by this bot. + + Used by the wake-decision to detect threads where the bot posted + the root via direct chat.postMessage (outside the gateway's + send() path) — see #63530. Without this, human replies in + bot-initiated threads were silently dropped when there was no + active session and no @mention. Root-authorship is derived from + the Slack API, so unlike the in-memory _bot_message_ts set it + also survives gateway restarts. + + Implementation: check the in-memory _thread_context_cache first + (cheap; populated whenever thread context is fetched). On a miss, + fetch thread context — the fetch is bounded by the TTL cache in + _fetch_thread_context, so the API-call overhead is paid only on + the miss path. + """ + if not thread_ts: + return False + + bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id) or "" + if not bot_uid: + return False + + def _cached_parent_matches() -> Optional[bool]: + # Cache keys are "{channel_id}:{thread_ts}:{team_id}"; team_id may + # be empty at some call sites, so match on the channel+thread + # prefix rather than guessing the exact key. + for cached_key, cached_entry in self._thread_context_cache.items(): + if cached_key.startswith(f"{channel_id}:{thread_ts}:"): + return bool( + cached_entry.parent_user_id + and cached_entry.parent_user_id == bot_uid + ) + return None + + cached = _cached_parent_matches() + if cached is not None: + return cached + + # Miss path: fetch thread context (its own TTL cache applies) and + # re-check — a successful fetch populates parent_user_id. + await self._fetch_thread_context( + channel_id=channel_id, + thread_ts=thread_ts, + current_ts="", + team_id=team_id, + ) + cached = _cached_parent_matches() + return bool(cached) + + async def _should_wake_on_unmentioned_message( + self, + event_thread_ts, + channel_id: str, + user_id: str, + is_thread_reply: bool, + team_id: str = "", + ) -> bool: + """Return True if the bot should wake on an un-mentioned message. + + Combines the four wake checks: + 1. _bot_message_ts (thread root was sent by us via send()) + 2. _mentioned_threads (someone @-mentioned us earlier) + 3. _has_active_session... (there's already an agent session) + 4. _bot_authored_thread_root (#63530: the bot posted the thread root + via direct chat.postMessage, outside the gateway send() path — + derived from the Slack API, so it also survives restarts). + + Extracted from the inline branch in _handle_slack_message so it + can be unit-tested without spinning up Slack or a real adapter + lifecycle. + """ + if not event_thread_ts: + return False + if is_thread_reply and event_thread_ts in self._bot_message_ts: + return True + if event_thread_ts in self._mentioned_threads: + return True + if is_thread_reply and self._has_active_session_for_thread( + channel_id=channel_id, + thread_ts=event_thread_ts, + user_id=user_id, + team_id=team_id, + ): + return True + # 4th check: bot-initiated thread via direct chat.postMessage. + if is_thread_reply and await self._bot_authored_thread_root( + channel_id=channel_id, + thread_ts=event_thread_ts, + team_id=team_id, + ): + return True + return False + async def _handle_slack_message( self, event: dict, payload: Optional[dict] = None ) -> None: @@ -3799,23 +3902,12 @@ class SlackAdapter(BasePlatformAdapter): elif self._slack_strict_mention() and not is_mentioned: return # Strict mode: ignore until @-mentioned again elif not is_mentioned: - reply_to_bot_thread = ( - is_thread_reply and event_thread_ts in self._bot_message_ts - ) - in_mentioned_thread = ( - event_thread_ts is not None - and event_thread_ts in self._mentioned_threads - ) - has_session = is_thread_reply and self._has_active_session_for_thread( + if not await self._should_wake_on_unmentioned_message( + event_thread_ts=event_thread_ts, channel_id=channel_id, - thread_ts=event_thread_ts, user_id=user_id, team_id=team_id, - ) - if ( - not reply_to_bot_thread - and not in_mentioned_thread - and not has_session + is_thread_reply=is_thread_reply, ): return @@ -5220,11 +5312,26 @@ class SlackAdapter(BasePlatformAdapter): team_id=team_id, channel_id=channel_id, ) + # Capture the parent message's user_id so _bot_authored_thread_root + # can detect threads whose root was posted by us via direct + # chat.postMessage (outside the gateway's send() path). #63530: + # bot-initiated threads with no active session were silently + # dropping human replies because _bot_message_ts only records + # gateway-routed sends. + parent_user_id = next( + ( + m.get("user", "") or "" + for m in messages + if m.get("ts", "") == thread_ts + ), + "", + ) self._thread_context_cache[cache_key] = _ThreadContextCache( content=content, fetched_at=now, message_count=len(messages), parent_text=parent_text, + parent_user_id=parent_user_id, messages=list(messages), ) if after_ts: diff --git a/tests/gateway/test_slack_wake_external_bot_messages.py b/tests/gateway/test_slack_wake_external_bot_messages.py new file mode 100644 index 000000000000..2f1b6d2ab62d --- /dev/null +++ b/tests/gateway/test_slack_wake_external_bot_messages.py @@ -0,0 +1,329 @@ +"""Regression tests for #63530 — Slack adapter drops human replies in +threads whose root was posted by the bot via direct chat.postMessage +(outside the gateway's send() path). + +Background: the adapter's wake-decision at the un-mentioned branch in +_handle_slack_message uses three checks: + + 1. thread_ts ∈ _bot_message_ts (only populated by send() / files_upload_v2) + 2. thread_ts ∈ _mentioned_threads (only populated on @mention) + 3. _has_active_session_for_thread(...) (survives restarts) + +When a skill posts a triage message into a Slack thread via the Web API +directly (chat.postMessage, no gateway run), the bot's own ts is NOT +recorded in _bot_message_ts. A human reply in that thread, without an +@-mention and without an existing session, falls through all three +checks and is silently dropped. The same gap opens after a gateway +restart: _bot_message_ts is process memory, so threads the bot started +before the restart no longer wake it. + +Fix: a 4th check — was the thread root authored by the bot? Root +authorship is derived from the Slack API (conversations.replies), so it +survives restarts, unlike the in-memory ts set. The wake decision is +extracted into _should_wake_on_unmentioned_message so it's directly +testable without spinning up Slack. +""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +# Mock slack-bolt / slack-sdk the same way test_slack_mention.py does. +def _ensure_slack_mock(): + if "slack_bolt" in sys.modules and hasattr(sys.modules["slack_bolt"], "__file__"): + return + slack_bolt = MagicMock() + slack_bolt.async_app.AsyncApp = MagicMock + slack_bolt.adapter.socket_mode.async_handler.AsyncSocketModeHandler = MagicMock + slack_sdk = MagicMock() + slack_sdk.web.async_client.AsyncWebClient = MagicMock + for name, mod in [ + ("slack_bolt", slack_bolt), + ("slack_bolt.async_app", slack_bolt.async_app), + ("slack_bolt.adapter", slack_bolt.adapter), + ("slack_bolt.adapter.socket_mode", slack_bolt.adapter.socket_mode), + ( + "slack_bolt.adapter.socket_mode.async_handler", + slack_bolt.adapter.socket_mode.async_handler, + ), + ("slack_sdk", slack_sdk), + ("slack_sdk.web", slack_sdk.web), + ("slack_sdk.web.async_client", slack_sdk.web.async_client), + ]: + sys.modules.setdefault(name, mod) + sys.modules.setdefault("aiohttp", MagicMock()) + + +_ensure_slack_mock() + +import plugins.platforms.slack.adapter as _slack_mod # noqa: E402 + +_slack_mod.SLACK_AVAILABLE = True + +from plugins.platforms.slack.adapter import ( # noqa: E402 + SlackAdapter, + _ThreadContextCache, +) + +from gateway.config import Platform, PlatformConfig # noqa: E402 + + +BOT_USER_ID = "U_BOT_OWN" +CHANNEL_ID = "C_incident" +USER_ID = "U_engineer" +THREAD_TS = "1700000000.000100" + + +def _make_adapter(bot_authored_root: bool = False): + """Build a bare SlackAdapter with the wake-decision state controlled. + + None of the 3 legacy in-memory checks pass by default: the bot didn't + send via gateway, the thread wasn't @-mentioned, and there is no active + session — exactly the post-restart / outside-send state. + """ + adapter = object.__new__(SlackAdapter) + adapter.platform = Platform.SLACK + adapter.config = PlatformConfig( + enabled=True, + extra={"require_mention": True, "strict_mention": False}, + ) + adapter._bot_user_id = BOT_USER_ID + adapter._team_bot_user_ids = {} + adapter._bot_message_ts = set() + adapter._mentioned_threads = set() + adapter._thread_context_cache = {} + + adapter._has_active_session_for_thread = lambda **kw: False + # Mock _fetch_thread_context so the miss-path doesn't make a real + # Slack API call. Tests that need a populated cache pre-populate + # _thread_context_cache directly. + adapter._fetch_thread_context = AsyncMock(return_value="") + # The 4th-check helper is mocked so wake-decision tests can control + # its result without setting up the full cache path. Helper-specific + # tests call the real method via the class instead. + adapter._bot_authored_thread_root = AsyncMock(return_value=bot_authored_root) + + return adapter + + +# --------------------------------------------------------------------------- +# _should_wake_on_unmentioned_message — composes all 4 checks +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_wake_decision_returns_false_when_not_thread_reply(): + """A top-level channel message (no thread_ts) should never wake the bot + when require_mention is true — unchanged by this fix.""" + adapter = _make_adapter(bot_authored_root=True) + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=None, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=False, + ) + assert wake is False + + +@pytest.mark.asyncio +async def test_wake_decision_returns_false_when_all_four_checks_miss(): + """All four checks miss (no bot-message, no mention, no session, no + bot-authored root) → wake decision is False.""" + adapter = _make_adapter(bot_authored_root=False) + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=THREAD_TS, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=True, + ) + assert wake is False + + +@pytest.mark.asyncio +async def test_wake_decision_returns_true_when_bot_authored_thread_root(): + """The new behavior (#63530): a human reply in a thread whose root was + authored by the bot via direct chat.postMessage (outside gateway send) + wakes the bot even though none of the legacy 3 checks pass — including + after a restart, when _bot_message_ts is empty.""" + adapter = _make_adapter(bot_authored_root=True) + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=THREAD_TS, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=True, + ) + assert wake is True, ( + "human reply in a thread whose root was bot-posted (not via gateway " + "send) should wake the bot — #63530" + ) + + +@pytest.mark.asyncio +async def test_wake_decision_returns_true_when_legacy_check_1_hits(): + """Regression guard: _bot_message_ts hit still wakes (additive check).""" + adapter = _make_adapter(bot_authored_root=False) + adapter._bot_message_ts = {THREAD_TS} + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=THREAD_TS, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=True, + ) + assert wake is True + + +@pytest.mark.asyncio +async def test_wake_decision_returns_true_when_legacy_check_2_hits(): + """Regression guard: _mentioned_threads hit still wakes (additive).""" + adapter = _make_adapter(bot_authored_root=False) + adapter._mentioned_threads = {THREAD_TS} + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=THREAD_TS, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=True, + ) + assert wake is True + + +@pytest.mark.asyncio +async def test_wake_decision_returns_true_when_legacy_check_3_hits(): + """Regression guard: an active session still wakes (additive).""" + adapter = _make_adapter(bot_authored_root=False) + adapter._has_active_session_for_thread = lambda **kw: True + wake = await adapter._should_wake_on_unmentioned_message( + event_thread_ts=THREAD_TS, + channel_id=CHANNEL_ID, + user_id=USER_ID, + is_thread_reply=True, + ) + assert wake is True + + +# --------------------------------------------------------------------------- +# _bot_authored_thread_root — the API-derived, restart-surviving check +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_true_from_cache(): + """Cache hit whose parent_user_id matches the bot's user_id → True.""" + adapter = _make_adapter() + adapter._thread_context_cache = { + f"{CHANNEL_ID}:{THREAD_TS}:": _ThreadContextCache( + content="[Thread context — prior messages...]", + fetched_at=0, + message_count=1, + parent_text="triage analysis", + parent_user_id=BOT_USER_ID, + ), + } + + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS + ) + assert result is True + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_false_for_human_authored_root(): + """A human-authored root must return False even on a cache hit — guards + against waking on any thread reply just because the cache is warm.""" + adapter = _make_adapter() + adapter._thread_context_cache = { + f"{CHANNEL_ID}:{THREAD_TS}:": _ThreadContextCache( + content="[Thread context — prior messages...]", + fetched_at=0, + message_count=1, + parent_text="someone else's message", + parent_user_id="U_other_user", + ), + } + + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS + ) + assert result is False + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_false_on_empty_thread_ts(): + """Defensive: empty thread_ts short-circuits to False without any + cache lookup or network call.""" + adapter = _make_adapter() + result = await SlackAdapter._bot_authored_thread_root(adapter, CHANNEL_ID, "") + assert result is False + adapter._fetch_thread_context.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_fetches_on_cache_miss(): + """Cache miss → _fetch_thread_context runs; a successful fetch that + populates parent_user_id with the bot's id yields True. This is the + restart path: fresh process, empty caches, root authorship recovered + from the Slack API.""" + adapter = _make_adapter() + + async def _fake_fetch(channel_id, thread_ts, current_ts, team_id=""): + adapter._thread_context_cache[f"{channel_id}:{thread_ts}:{team_id}"] = ( + _ThreadContextCache( + content="ctx", + fetched_at=0, + message_count=1, + parent_text="bot-posted root", + parent_user_id=BOT_USER_ID, + ) + ) + return "ctx" + + adapter._fetch_thread_context = AsyncMock(side_effect=_fake_fetch) + + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS + ) + assert result is True + adapter._fetch_thread_context.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_false_when_fetch_fails(): + """Fetch failure (empty result, nothing cached) → False, no wake.""" + adapter = _make_adapter() + adapter._fetch_thread_context = AsyncMock(return_value="") + + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS + ) + assert result is False + + +@pytest.mark.asyncio +async def test_bot_authored_thread_root_uses_per_team_bot_id(): + """Multi-workspace: the comparison must use the team's bot user id, + not the primary workspace's.""" + adapter = _make_adapter() + adapter._team_bot_user_ids = {"T2": "U_BOT_T2"} + adapter._thread_context_cache = { + f"{CHANNEL_ID}:{THREAD_TS}:T2": _ThreadContextCache( + content="ctx", + fetched_at=0, + message_count=1, + parent_text="root", + parent_user_id="U_BOT_T2", + ), + } + + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS, team_id="T2" + ) + assert result is True + # And the primary bot id must NOT match in that workspace. + adapter._thread_context_cache[f"{CHANNEL_ID}:{THREAD_TS}:T2"].parent_user_id = ( + BOT_USER_ID + ) + result = await SlackAdapter._bot_authored_thread_root( + adapter, CHANNEL_ID, THREAD_TS, team_id="T2" + ) + assert result is False