From a60b00e12d0a70523d47a842321c07e0deae176c Mon Sep 17 00:00:00 2001 From: Jordan Hubbard Date: Thu, 16 Jul 2026 00:00:42 +0100 Subject: [PATCH] fix(slack): isolate workspace-local routing --- gateway/platforms/base.py | 1 + gateway/session.py | 236 +++++++++++++++++++++++------ plugins/platforms/slack/adapter.py | 229 +++++++++++++++++++++------- tests/gateway/test_session.py | 125 +++++++++++++++ tests/gateway/test_slack.py | 85 ++++++++++- 5 files changed, 574 insertions(+), 102 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 2a6623abd91b5..51c9f42405a05 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -5817,6 +5817,7 @@ class BasePlatformAdapter(ABC): user_id_alt=user_id_alt, chat_id_alt=chat_id_alt, is_bot=is_bot, + scope_id=str(scope_id) if scope_id else None, guild_id=str(guild_id) if guild_id else None, parent_chat_id=str(parent_chat_id) if parent_chat_id else None, message_id=str(message_id) if message_id else None, diff --git a/gateway/session.py b/gateway/session.py index 968c7f21dc136..31829cf5eee62 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -17,7 +17,7 @@ import threading import uuid from pathlib import Path from datetime import datetime, timedelta -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Dict, List, Optional, Any logger = logging.getLogger(__name__) @@ -1042,12 +1042,16 @@ def build_session_key( multiplexing gateway passes a non-default profile. DM rules: + - Slack ``scope_id`` identifies the workspace before chat/user ids. Other + platforms retain their existing key format; in particular, Discord + guild scope is intentionally not added here as a compatibility change. - DMs include chat_id when present, so each private conversation is isolated. - thread_id further differentiates threaded DMs within the same DM chat. - Without chat_id, thread_id is used as a best-effort fallback. - Without thread_id or chat_id, DMs share a single session. Group/channel rules: + - Slack ``scope_id`` identifies the workspace before chat/thread ids. - chat_id identifies the parent group/channel. - user_id/user_id_alt isolates participants within that parent chat when available when ``group_sessions_per_user`` is enabled. @@ -1062,15 +1066,24 @@ def build_session_key( """ ns = _session_key_namespace(profile) platform = source.platform.value + slack_scope_id = ( + str(source.scope_id) + if source.platform == Platform.SLACK and source.scope_id + else None + ) if source.chat_type == "dm": dm_chat_id = source.chat_id if source.platform == Platform.WHATSAPP: dm_chat_id = canonical_whatsapp_identifier(source.chat_id) + dm_parts = [ns, platform, "dm"] + if slack_scope_id: + dm_parts.append(slack_scope_id) if dm_chat_id: + dm_parts.append(dm_chat_id) if source.thread_id: - return f"{ns}:{platform}:dm:{dm_chat_id}:{source.thread_id}" - return f"{ns}:{platform}:dm:{dm_chat_id}" + dm_parts.append(source.thread_id) + return ":".join(str(part) for part in dm_parts) # No chat_id — fall back to the sender's own identifier before the # bare per-platform sink. Without this, every DM from every user that # arrives without a chat_id (non-standard adapters / synthetic sources) @@ -1084,12 +1097,13 @@ def build_session_key( or dm_participant_id ) if dm_participant_id: + dm_parts.append(str(dm_participant_id)) if source.thread_id: - return f"{ns}:{platform}:dm:{dm_participant_id}:{source.thread_id}" - return f"{ns}:{platform}:dm:{dm_participant_id}" + dm_parts.append(source.thread_id) + return ":".join(str(part) for part in dm_parts) if source.thread_id: - return f"{ns}:{platform}:dm:{source.thread_id}" - return f"{ns}:{platform}:dm" + dm_parts.append(source.thread_id) + return ":".join(str(part) for part in dm_parts) participant_id = source.user_id_alt or source.user_id if participant_id and source.platform == Platform.WHATSAPP: @@ -1099,6 +1113,8 @@ def build_session_key( participant_id = canonical_whatsapp_identifier(str(participant_id)) or participant_id key_parts = [ns, platform, source.chat_type] + if slack_scope_id: + key_parts.append(slack_scope_id) if source.chat_id: key_parts.append(source.chat_id) if source.thread_id: @@ -1114,7 +1130,7 @@ def build_session_key( if isolate_user and participant_id: key_parts.append(str(participant_id)) - return ":".join(key_parts) + return ":".join(str(part) for part in key_parts) class _SessionFlight: @@ -1164,6 +1180,11 @@ class SessionStore: self._persisted_routing_generation = 0 self._inflight_lock = threading.Lock() self._inflight_sessions: Dict[str, _SessionFlight] = {} + # An unscoped pre-migration Slack key can represent at most one + # workspace. Claim it once per process so simultaneous first messages + # from two workspaces cannot both revive the same legacy session. + self._legacy_slack_claim_lock = threading.Lock() + self._claimed_legacy_slack_keys: set[str] = set() self._transcript_retry_lock = threading.Lock() self._dirty_transcripts: Dict[str, List[Dict[str, Any]]] = {} self._transcript_append_failures: Dict[str, int] = {} @@ -1547,6 +1568,46 @@ class SessionStore: profile=self._resolve_profile_for_key(source), ) + def _legacy_slack_session_key(self, source: SessionSource) -> Optional[str]: + """Return the pre-workspace Slack key for an explicitly scoped source. + + The compatibility path is deliberately Slack-only. Discord and every + other platform keep byte-identical keys, and an unscoped Slack session + may be claimed by only one workspace because its old key contains no + information that could safely distinguish multiple teams. + """ + if source.platform != Platform.SLACK or not source.scope_id: + return None + legacy_source = replace(source, scope_id=None, guild_id=None) + return build_session_key( + legacy_source, + group_sessions_per_user=getattr( + self.config, "group_sessions_per_user", True + ), + thread_sessions_per_user=getattr( + self.config, "thread_sessions_per_user", False + ), + profile=self._resolve_profile_for_key(source), + ) + + def _claim_legacy_slack_key(self, legacy_key: Optional[str]) -> bool: + """Atomically reserve one ambiguous legacy Slack key for migration.""" + if not legacy_key: + return False + claim_lock = getattr(self, "_legacy_slack_claim_lock", None) + if claim_lock is None: + claim_lock = threading.Lock() + self._legacy_slack_claim_lock = claim_lock + with claim_lock: + claimed = getattr(self, "_claimed_legacy_slack_keys", None) + if claimed is None: + claimed = set() + self._claimed_legacy_slack_keys = claimed + if legacy_key in claimed: + return False + claimed.add(legacy_key) + return True + def _create_entry_from_recovered_row( self, *, @@ -1571,6 +1632,45 @@ class SessionStore: chat_type=source.chat_type, ) + def _find_gateway_session_row( + self, + *, + session_key: str, + source: SessionSource, + allow_peer_fallback: bool, + raise_on_lookup_error: bool = False, + ) -> Optional[Dict[str, Any]]: + """Query one durable gateway session row. + + Scoped Slack lookups disable SessionDB's platform/chat/user fallback: + that tuple does not contain a workspace id and could therefore revive + another team's session. The caller performs one explicit exact lookup + of the old unscoped key instead. + """ + if not self._db: + return None + finder = getattr(self._db, "find_latest_gateway_session_for_peer", None) + if not callable(finder): + return None + try: + return finder( + source=source.platform.value, + user_id=source.user_id, + session_key=session_key, + chat_id=source.chat_id if allow_peer_fallback else None, + chat_type=source.chat_type if allow_peer_fallback else None, + thread_id=source.thread_id, + ) + except Exception as exc: + logger.debug( + "Gateway session DB recovery failed for %s: %s", + session_key, + exc, + ) + if raise_on_lookup_error: + raise + return None + def _recover_session_from_db( self, *, @@ -1580,25 +1680,26 @@ class SessionStore: raise_on_lookup_error: bool = False, ) -> Optional[SessionEntry]: """Rebuild a missing session-key mapping from durable state.db data.""" - if not self._db: - return None - finder = getattr(self._db, "find_latest_gateway_session_for_peer", None) - if not callable(finder): - return None - try: - recovered = finder( - source=source.platform.value, - user_id=source.user_id, - session_key=session_key, - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, + legacy_key = self._legacy_slack_session_key(source) + recovered = self._find_gateway_session_row( + session_key=session_key, + source=source, + allow_peer_fallback=legacy_key is None, + raise_on_lookup_error=raise_on_lookup_error, + ) + migrated_legacy = False + if ( + not recovered + and legacy_key + and self._claim_legacy_slack_key(legacy_key) + ): + recovered = self._find_gateway_session_row( + session_key=legacy_key, + source=source, + allow_peer_fallback=False, + raise_on_lookup_error=raise_on_lookup_error, ) - except Exception as exc: - logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc) - if raise_on_lookup_error: - raise - return None + migrated_legacy = bool(recovered) if not recovered: return None if not self._recovered_row_allowed_for_active_profile( @@ -1617,36 +1718,44 @@ class SessionStore: self._db.reopen_session(str(recovered["id"])) except Exception as exc: logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc) - return self._create_entry_from_recovered_row( + entry = self._create_entry_from_recovered_row( row=recovered, session_key=session_key, source=source, now=now, ) + if migrated_legacy: + self._record_gateway_session_peer( + entry.session_id, + session_key, + source, + display_name=entry.display_name, + ) + return entry def _query_recoverable_session(self, *, session_key, source, now): """DB-only half of _recover_session_from_db (no lock needed). Returns a SessionEntry or None. Caller assigns _entries[key] under lock. """ - if not self._db: - return None - finder = getattr(self._db, "find_latest_gateway_session_for_peer", None) - if not callable(finder): - return None - try: - recovered = finder( - source=source.platform.value, - user_id=source.user_id, - session_key=session_key, - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, + legacy_key = self._legacy_slack_session_key(source) + recovered = self._find_gateway_session_row( + session_key=session_key, + source=source, + allow_peer_fallback=legacy_key is None, + ) + migrated_legacy = False + if ( + not recovered + and legacy_key + and self._claim_legacy_slack_key(legacy_key) + ): + recovered = self._find_gateway_session_row( + session_key=legacy_key, + source=source, + allow_peer_fallback=False, ) - except Exception as exc: - logger.debug("Gateway session DB recovery failed for %s: %s", - session_key, exc) - return None + migrated_legacy = bool(recovered) if not isinstance(recovered, dict): return None if not self._recovered_row_allowed_for_active_profile( @@ -1666,9 +1775,17 @@ class SessionStore: except Exception as exc: logger.debug("Gateway session DB reopen failed for %s: %s", session_key, exc) - return self._create_entry_from_recovered_row( + entry = self._create_entry_from_recovered_row( row=recovered, session_key=session_key, source=source, now=now, ) + if migrated_legacy: + self._record_gateway_session_peer( + entry.session_id, + session_key, + source, + display_name=entry.display_name, + ) + return entry def _record_gateway_session_peer( self, session_id: str, @@ -2032,6 +2149,35 @@ class SessionStore: session_key = self._generate_session_key(source) now = _now() + # One-time routing-index migration for Slack sessions created before + # workspace scope was part of the key. Move (rather than copy) the + # legacy entry so a second workspace with identical Slack ids cannot + # attach to the same transcript. + migrated_legacy_entry: Optional[SessionEntry] = None + legacy_key = self._legacy_slack_session_key(source) + if legacy_key and not force_new: + with self._lock: + self._ensure_loaded_locked() + if ( + session_key not in self._entries + and legacy_key in self._entries + and self._claim_legacy_slack_key(legacy_key) + ): + migrated_legacy_entry = self._entries.pop(legacy_key) + migrated_legacy_entry.session_key = session_key + migrated_legacy_entry.origin = source + migrated_legacy_entry.platform = source.platform + migrated_legacy_entry.chat_type = source.chat_type + self._entries[session_key] = migrated_legacy_entry + if migrated_legacy_entry is not None: + self._save_entries() + self._record_gateway_session_peer( + migrated_legacy_entry.session_id, + session_key, + source, + display_name=migrated_legacy_entry.display_name, + ) + db_end_session_id = None db_create_kwargs = None existing_session_id = None diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index f898d9a71c877..8004b06897124 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -905,9 +905,16 @@ class SlackAdapter(BasePlatformAdapter): # sees (DM channel IDs are per-user), so it must be bounded on busy # multi-workspace installs. Eviction is safe: entries are re-learned # from the next event on that channel, and _get_client falls back to - # the primary client meanwhile. + # the primary client meanwhile. Entries exist only while a channel id + # maps to exactly one workspace (see _remember_channel_team); + # explicit outbound metadata remains the authoritative route. self._channel_team: Dict[str, str] = {} self._CHANNEL_TEAM_MAX = 10000 + # channel_id → every team_id that has claimed it. Slack channel ids + # are workspace-local, so the same id CAN appear in two workspaces — + # when that happens the unqualified fallback is ambiguous and must be + # dropped rather than silently routed to whichever team wrote last. + self._channel_teams: Dict[str, set] = {} # user target (team_id:user_id) → opened DM conversation ID (D...) self._dm_conversation_cache: Dict[str, str] = {} self._DM_CONVERSATION_CACHE_MAX = 5000 @@ -925,12 +932,13 @@ class SlackAdapter(BasePlatformAdapter): self._PROCESSED_MESSAGE_TS_MAX = 5000 # Track pending approval message_ts → resolved flag to prevent # double-clicks on approval buttons. Bounded: an approval prompt the - # user never clicks would otherwise leak its entry forever. - self._approval_resolved: Dict[str, bool] = {} + # user never clicks would otherwise leak its entry forever. Keys may + # be workspace-scoped markers (team_id, ts) in multi-workspace mode. + self._approval_resolved: Dict[Any, bool] = {} self._APPROVAL_RESOLVED_MAX = 1000 # Same guard for clarify prompts (interactive multiple-choice # buttons); mirrors _approval_resolved. - self._clarify_resolved: Dict[str, bool] = {} + self._clarify_resolved: Dict[Any, bool] = {} self._CLARIFY_RESOLVED_MAX = 1000 # Track timestamps of messages sent by the bot so we can respond # to thread replies even without an explicit @mention. @@ -989,9 +997,11 @@ class SlackAdapter(BasePlatformAdapter): self._TITLED_ASSISTANT_THREADS_MAX = 5000 # Slash-command contexts: stash response_url + user_id so send() # can route the first reply ephemerally. Keyed by - # (channel_id, user_id) to avoid cross-user collisions. + # (team_id, channel_id, user_id) to avoid cross-workspace and + # cross-user collisions. The two-part form remains readable only for + # commands that arrived without a workspace id. # Each value: {"response_url": str, "ts": float} - self._slash_command_contexts: Dict[Tuple[str, str], Dict[str, Any]] = {} + self._slash_command_contexts: Dict[Tuple[str, ...], Dict[str, Any]] = {} # Socket Mode resilience: track runtime connection state so we can # self-heal when Slack silently drops the websocket. self._app_token: Optional[str] = None @@ -1036,8 +1046,15 @@ class SlackAdapter(BasePlatformAdapter): break @staticmethod - def _slack_timestamp_sort_key(ts: str) -> Tuple[int, int, str]: - """Return a chronological, deterministic sort key for Slack timestamps.""" + def _slack_timestamp_sort_key(ts: Any) -> Tuple[int, int, str]: + """Return a chronological, deterministic sort key for Slack timestamps. + + Accepts bare ``"seconds.fraction"`` strings and workspace-scoped + ``(team_id, ts)`` markers (see ``_workspace_message_marker``) — the + embedded ts drives the chronology in both cases. + """ + if isinstance(ts, tuple) and len(ts) == 2: + ts = ts[1] seconds, _, fraction = str(ts).partition(".") try: seconds_int = int(seconds) @@ -1109,11 +1126,27 @@ class SlackAdapter(BasePlatformAdapter): entries.discard(entry) def _remember_channel_team(self, channel_id: str, team_id: str) -> None: - """Record which workspace owns *channel_id*, bounded oldest-first.""" + """Record which workspace owns *channel_id*, bounded oldest-first. + + The unqualified fallback entry exists only while a channel id maps to + exactly one workspace: Slack channel ids are workspace-local, so the + same id CAN appear in two workspaces — when that happens the fallback + is ambiguous and is dropped rather than silently routed to whichever + team wrote last. Explicit outbound metadata (team_id) remains the + authoritative route. + """ if not channel_id or not team_id: return - self._channel_team[str(channel_id)] = str(team_id) + channel_id = str(channel_id) + team_id = str(team_id) + teams = self._channel_teams.setdefault(channel_id, set()) + teams.add(team_id) + if len(teams) == 1: + self._channel_team[channel_id] = team_id + else: + self._channel_team.pop(channel_id, None) self._trim_oldest_dict_entries(self._channel_team, self._CHANNEL_TEAM_MAX) + self._trim_oldest_dict_entries(self._channel_teams, self._CHANNEL_TEAM_MAX) def _start_socket_mode_handler(self) -> None: """Start the Slack Socket Mode background task.""" @@ -1429,17 +1462,20 @@ class SlackAdapter(BasePlatformAdapter): def _pop_slash_context( self, chat_id: str, + team_id: str = "", ) -> Optional[Dict[str, Any]]: """Return and remove the slash-command context for *chat_id*, if fresh. Contexts older than ``_SLASH_CTX_TTL`` seconds are silently discarded. Uses the ``_slash_user_id`` ContextVar (set in ``_handle_slash_command``) - to match the exact ``(channel_id, user_id)`` key. This prevents a - concurrent slash command from a different user on the same channel from - stealing another user's ephemeral context. When the ContextVar is - unset (e.g. send() called from a non-slash code path), do not match - anything — otherwise normal sends can steal a pending slash reply. + to match the exact ``(team_id, channel_id, user_id)`` key. This prevents + a concurrent slash command from another user or workspace with the same + Slack-local ids from stealing the ephemeral context. The legacy + two-part form is used only for commands that arrived without a + workspace id. When the ContextVar is unset (e.g. send() called from a + non-slash code path), do not match anything — otherwise normal sends + can steal a pending slash reply. """ now = time.monotonic() # Clean up stale entries on every lookup — dict is small. @@ -1451,10 +1487,13 @@ class SlackAdapter(BasePlatformAdapter): for k in stale_keys: self._slash_command_contexts.pop(k, None) - # Precise match: (channel_id, user_id) from ContextVar. + team_id = str(team_id or "") + + # Precise match from ContextVar. uid = _slash_user_id.get() if uid: - return self._slash_command_contexts.pop((chat_id, uid), None) + key = (team_id, chat_id, uid) if team_id else (chat_id, uid) + return self._slash_command_contexts.pop(key, None) return None @@ -2234,12 +2273,40 @@ class SlackAdapter(BasePlatformAdapter): """Return Slack workspace id from generic or Slack-specific metadata.""" if not metadata: return "" - return str( - metadata.get("team_id") - or metadata.get("team") - or metadata.get("slack_team_id") - or "" - ) + for key in ( + "scope_id", + "slack_team_id", + "team_id", + "team", + "guild_id", + "workspace_id", + ): + value = metadata.get(key) + if value: + return str(value) + source = metadata.get("source") + if isinstance(source, dict): + for key in ("scope_id", "slack_team_id", "team_id", "guild_id"): + value = source.get(key) + if value: + return str(value) + elif source is not None: + value = getattr(source, "scope_id", None) or getattr( + source, "guild_id", None + ) + if value: + return str(value) + return "" + + @staticmethod + def _workspace_event_id(team_id: str, event_id: str) -> str: + """Scope Slack's workspace-local event/message ids for deduplication.""" + return f"{team_id}:{event_id}" if team_id else str(event_id) + + @staticmethod + def _workspace_message_marker(team_id: str, message_id: str) -> Any: + """Return an in-memory routing marker without changing legacy no-team tests.""" + return (str(team_id), str(message_id)) if team_id else str(message_id) def _get_client(self, chat_id: str, team_id: Optional[str] = None) -> Any: """Return the workspace-specific WebClient for a channel.""" @@ -2332,12 +2399,13 @@ class SlackAdapter(BasePlatformAdapter): ) thread_ts = None try: + team_id = self._metadata_team_id(metadata) # Check for a pending slash-command context. When the user ran a # native slash command (e.g. /q, /stop, /model), the initial ack # already showed an ephemeral "Running /cmd…" message. If we have # a stashed response_url for this channel, replace that ack with # the actual command reply ephemerally instead of posting publicly. - slash_ctx = self._pop_slash_context(chat_id) + slash_ctx = self._pop_slash_context(chat_id, team_id) if slash_ctx: ephemeral_result = await self._send_slash_ephemeral( slash_ctx, @@ -2427,7 +2495,7 @@ class SlackAdapter(BasePlatformAdapter): try: last_result = await self._get_client( - chat_id, team_id=self._metadata_team_id(metadata) + chat_id, team_id=team_id ).chat_postMessage(**kwargs) except Exception as e: if kwargs.get("blocks") and self._is_block_payload_rejection(e): @@ -2438,7 +2506,7 @@ class SlackAdapter(BasePlatformAdapter): e, ) last_result = await self._get_client( - chat_id, team_id=self._metadata_team_id(metadata) + chat_id, team_id=team_id ).chat_postMessage(**retry_kwargs) else: raise @@ -2451,10 +2519,14 @@ class SlackAdapter(BasePlatformAdapter): # replies without requiring @mention. sent_ts = last_result.get("ts") if last_result else None if sent_ts: - self._bot_message_ts.add(sent_ts) + self._bot_message_ts.add( + self._workspace_message_marker(team_id, sent_ts) + ) # Also register the thread root so replies-to-my-replies work if thread_ts: - self._bot_message_ts.add(thread_ts) + self._bot_message_ts.add( + self._workspace_message_marker(team_id, thread_ts) + ) self._trim_bot_message_timestamps() return SendResult( @@ -3039,7 +3111,7 @@ class SlackAdapter(BasePlatformAdapter): initial_comment=caption or "", thread_ts=thread_ts, ) - self._record_uploaded_file_thread(chat_id, thread_ts) + self._record_uploaded_file_thread(chat_id, thread_ts, metadata) return SendResult(success=True, raw_response=result) except Exception as exc: last_exc = exc @@ -3175,7 +3247,7 @@ class SlackAdapter(BasePlatformAdapter): initial_comment=initial_comment, thread_ts=thread_ts, ) - self._record_uploaded_file_thread(chat_id, thread_ts) + self._record_uploaded_file_thread(chat_id, thread_ts, metadata) _ = result except Exception as e: logger.warning( @@ -3190,12 +3262,18 @@ class SlackAdapter(BasePlatformAdapter): ) def _record_uploaded_file_thread( - self, chat_id: str, thread_ts: Optional[str] + self, + chat_id: str, + thread_ts: Optional[str], + metadata: Optional[Dict[str, Any]] = None, ) -> None: """Treat successful file uploads as bot participation in a thread.""" if not thread_ts: return - self._bot_message_ts.add(thread_ts) + team_id = self._metadata_team_id(metadata) + self._bot_message_ts.add( + self._workspace_message_marker(team_id, thread_ts) + ) self._trim_bot_message_timestamps() def _is_retryable_upload_error(self, exc: Exception) -> bool: @@ -3580,13 +3658,13 @@ class SlackAdapter(BasePlatformAdapter): if not self._reactions_enabled(): return ts = getattr(event, "message_id", None) - if not ts or ts not in self._reacting_message_ids: + team_id = str(getattr(event.source, "scope_id", "") or "") + marker = self._workspace_message_marker(team_id, ts) if ts else None + if not ts or marker not in self._reacting_message_ids: return channel_id = getattr(event.source, "chat_id", None) if channel_id: - await self._add_reaction( - channel_id, ts, "eyes", str(getattr(event.source, "scope_id", "") or "") - ) + await self._add_reaction(channel_id, ts, "eyes", team_id) async def on_processing_complete( self, event: MessageEvent, outcome: ProcessingOutcome @@ -3595,13 +3673,14 @@ class SlackAdapter(BasePlatformAdapter): if not self._reactions_enabled(): return ts = getattr(event, "message_id", None) - if not ts or ts not in self._reacting_message_ids: + team_id = str(getattr(event.source, "scope_id", "") or "") + marker = self._workspace_message_marker(team_id, ts) if ts else None + if not ts or marker not in self._reacting_message_ids: return - self._reacting_message_ids.discard(ts) + self._reacting_message_ids.discard(marker) channel_id = getattr(event.source, "chat_id", None) if not channel_id: return - team_id = str(getattr(event.source, "scope_id", "") or "") await self._remove_reaction(channel_id, ts, "eyes", team_id) if outcome == ProcessingOutcome.SUCCESS: await self._add_reaction(channel_id, ts, "white_check_mark", team_id) @@ -3864,7 +3943,7 @@ class SlackAdapter(BasePlatformAdapter): initial_comment=caption or "", thread_ts=thread_ts, ) - self._record_uploaded_file_thread(chat_id, thread_ts) + self._record_uploaded_file_thread(chat_id, thread_ts, metadata) return SendResult(success=True, raw_response=result) @@ -3945,7 +4024,7 @@ class SlackAdapter(BasePlatformAdapter): initial_comment=caption or "", thread_ts=thread_ts, ) - self._record_uploaded_file_thread(chat_id, thread_ts) + self._record_uploaded_file_thread(chat_id, thread_ts, metadata) return SendResult(success=True, raw_response=result) except Exception as exc: last_exc = exc @@ -4010,7 +4089,7 @@ class SlackAdapter(BasePlatformAdapter): initial_comment=caption or "", thread_ts=thread_ts, ) - self._record_uploaded_file_thread(chat_id, thread_ts) + self._record_uploaded_file_thread(chat_id, thread_ts, metadata) return SendResult(success=True, raw_response=result) except Exception as exc: last_exc = exc @@ -4836,7 +4915,9 @@ class SlackAdapter(BasePlatformAdapter): # If it does, _handle_slack_message records the same share ts and this # fallback skips instead of duplicating the user turn. await asyncio.sleep(0.75) - if ts and self._dedup.is_duplicate(ts): + if ts and self._dedup.is_duplicate( + self._workspace_event_id(team_id, ts) + ): return fallback_event = { @@ -4854,15 +4935,19 @@ class SlackAdapter(BasePlatformAdapter): fallback_event["thread_ts"] = thread_ts await self._handle_slack_message(fallback_event) - def _register_mentioned_thread(self, thread_ts: str) -> None: + def _register_mentioned_thread(self, thread_ts: str, team_id: str = "") -> None: """Record a thread as bot-mentioned so future replies auto-trigger. Centralizes the bounded-set eviction previously inlined at the - mention branch of _handle_slack_message. + mention branch of _handle_slack_message. Markers are workspace-scoped + (``(team_id, ts)``) when a team id is known so identical thread ts + values in two workspaces never wake each other's bot. """ if not thread_ts: return - self._mentioned_threads.add(thread_ts) + self._mentioned_threads.add( + self._workspace_message_marker(team_id, thread_ts) + ) self._trim_mentioned_threads() async def _bot_authored_thread_root( @@ -4943,9 +5028,19 @@ class SlackAdapter(BasePlatformAdapter): """ if not event_thread_ts: return False - if is_thread_reply and event_thread_ts in self._bot_message_ts: + thread_marker = self._workspace_message_marker(team_id, event_thread_ts) + # Check both the workspace-scoped marker and the bare ts: entries + # recorded before a team id was learned (or by legacy paths) are bare + # strings, and a scoped-vs-bare mismatch must not silence the bot. + if is_thread_reply and ( + thread_marker in self._bot_message_ts + or event_thread_ts in self._bot_message_ts + ): return True - if event_thread_ts in self._mentioned_threads: + if ( + thread_marker in self._mentioned_threads + or event_thread_ts in self._mentioned_threads + ): return True if is_thread_reply and self._has_active_session_for_thread( channel_id=channel_id, @@ -5044,8 +5139,14 @@ class SlackAdapter(BasePlatformAdapter): event = normalized_event # Dedup: Slack Socket Mode can redeliver events after reconnects (#4777) + # Scope the dedup id by workspace: Slack event ts values are only + # unique within one workspace, so two teams' events with the same ts + # must not suppress each other. event_ts = event.get("_slack_changed_event_ts") or event.get("ts", "") - if event_ts and self._dedup.is_duplicate(event_ts): + dedup_team_id = self._event_team_id(event, payload) + if event_ts and self._dedup.is_duplicate( + self._workspace_event_id(dedup_team_id, event_ts) + ): return # Bot/app-authored message filtering (SLACK_ALLOW_BOTS / config @@ -5494,7 +5595,7 @@ class SlackAdapter(BasePlatformAdapter): and not self._slack_strict_mention() and not self._slack_thread_require_mention() ): - self._register_mentioned_thread(thread_ts) + self._register_mentioned_thread(thread_ts, team_id=team_id) # Thread context rules: # - First message in a thread session (cold start): hydrate full @@ -6040,9 +6141,12 @@ class SlackAdapter(BasePlatformAdapter): # be @mentioned to earn a reaction — same as any channel. _should_react = (is_one_to_one_dm or is_mentioned) and self._reactions_enabled() if _should_react: - self._reacting_message_ids.add(ts) + self._reacting_message_ids.add( + self._workspace_message_marker(team_id, ts) + ) if len(self._reacting_message_ids) > self._REACTING_MESSAGE_IDS_MAX: - # Entries are bare Slack message ts values — evict oldest first. + # Entries embed a Slack message ts (bare or workspace-scoped + # tuple) — evict oldest first by the embedded ts. self._discard_oldest_slack_timestamps( self._reacting_message_ids, len(self._reacting_message_ids) @@ -6171,7 +6275,10 @@ class SlackAdapter(BasePlatformAdapter): ).chat_postMessage(**kwargs) msg_ts = result.get("ts", "") if msg_ts: - self._approval_resolved[msg_ts] = False + team_id = self._metadata_team_id(metadata) + self._approval_resolved[ + self._workspace_message_marker(team_id, msg_ts) + ] = False self._trim_oldest_dict_entries( self._approval_resolved, self._APPROVAL_RESOLVED_MAX ) @@ -6624,7 +6731,14 @@ class SlackAdapter(BasePlatformAdapter): choice = choice_map.get(action_id, "deny") # Prevent double-clicks — atomic pop; first caller gets False, others get True (default) - if self._approval_resolved.pop(msg_ts, True): + # Check both the workspace-scoped marker and the bare ts: the approval + # may have been stored without a team id (metadata-poor send path) + # while the click event carries one, and that mismatch must not + # swallow a legitimate first click. + approval_key = self._workspace_message_marker(team_id, msg_ts) + if msg_ts in self._approval_resolved: + approval_key = msg_ts + if self._approval_resolved.pop(approval_key, True): return # Resolve the approval FIRST — this unblocks the agent thread. Render @@ -7462,7 +7576,12 @@ class SlackAdapter(BasePlatformAdapter): # the whole channel can see the agent's answer. response_url = command.get("response_url", "") if response_url and user_id and channel_id and text.startswith("/"): - self._slash_command_contexts[(channel_id, user_id)] = { + context_key = ( + (str(team_id), str(channel_id), str(user_id)) + if team_id + else (str(channel_id), str(user_id)) + ) + self._slash_command_contexts[context_key] = { "response_url": response_url, # Kept for the chat.postEphemeral fallback when response_url # delivery fails — postEphemeral needs an explicit user. diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index f927bc00b8552..86d4d2eeac7ce 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -1,6 +1,7 @@ """Tests for gateway session management.""" import json import pytest +from dataclasses import replace from pathlib import Path from unittest.mock import patch, MagicMock from gateway.config import Platform, HomeChannel, GatewayConfig, PlatformConfig @@ -946,6 +947,130 @@ class TestSessionStoreLookupBySessionId: assert store.lookup_by_session_id("") is None +class TestSlackWorkspaceSessionIsolation: + @pytest.fixture() + def store(self, tmp_path): + config = GatewayConfig() + with patch("gateway.session.SessionStore._ensure_loaded"): + session_store = SessionStore(sessions_dir=tmp_path, config=config) + session_store._db = None + session_store._loaded = True + return session_store + + def test_dm_keys_include_only_slack_workspace_scope(self): + first = SessionSource( + platform=Platform.SLACK, + scope_id="T111", + chat_id="D123", + chat_type="dm", + ) + second = SessionSource( + platform=Platform.SLACK, + scope_id="T222", + chat_id="D123", + chat_type="dm", + ) + + assert build_session_key(first) == "agent:main:slack:dm:T111:D123" + assert build_session_key(second) == "agent:main:slack:dm:T222:D123" + assert build_session_key(first) != build_session_key(second) + + discord = SessionSource( + platform=Platform.DISCORD, + scope_id="G111", + chat_id="D123", + chat_type="dm", + ) + assert build_session_key(discord) == "agent:main:discord:dm:D123" + + def test_channel_keys_include_workspace_scope(self): + first = SessionSource( + platform=Platform.SLACK, + scope_id="T111", + chat_id="C123", + chat_type="group", + user_id="U1", + thread_id="1700000000.000100", + ) + second = SessionSource( + platform=Platform.SLACK, + scope_id="T222", + chat_id="C123", + chat_type="group", + user_id="U1", + thread_id="1700000000.000100", + ) + + expected_suffix = "C123:1700000000.000100" + assert build_session_key(first) == f"agent:main:slack:group:T111:{expected_suffix}" + assert build_session_key(second) == f"agent:main:slack:group:T222:{expected_suffix}" + assert build_session_key(first) != build_session_key(second) + + def test_legacy_routing_entry_moves_to_first_workspace_only(self, store): + legacy_source = SessionSource( + platform=Platform.SLACK, + chat_id="D_SHARED", + chat_type="dm", + user_id="U_SHARED", + ) + legacy_entry = store.get_or_create_session(legacy_source) + legacy_key = legacy_entry.session_key + + team_one_source = SessionSource( + platform=Platform.SLACK, + scope_id="T_ONE", + chat_id="D_SHARED", + chat_type="dm", + user_id="U_SHARED", + ) + team_one_entry = store.get_or_create_session(team_one_source) + + assert team_one_entry.session_id == legacy_entry.session_id + assert team_one_entry.session_key == "agent:main:slack:dm:T_ONE:D_SHARED" + assert legacy_key not in store._entries + + team_two_source = replace(team_one_source, scope_id="T_TWO", guild_id="T_TWO") + team_two_entry = store.get_or_create_session(team_two_source) + assert team_two_entry.session_id != team_one_entry.session_id + assert team_two_entry.session_key == "agent:main:slack:dm:T_TWO:D_SHARED" + + def test_legacy_db_fallback_is_exact_and_rewrites_peer_key(self, store): + source = SessionSource( + platform=Platform.SLACK, + scope_id="T_ONE", + chat_id="D_SHARED", + chat_type="dm", + user_id="U_SHARED", + ) + scoped_key = build_session_key(source) + legacy_key = build_session_key(replace(source, scope_id=None, guild_id=None)) + store._db = MagicMock() + store._db.find_latest_gateway_session_for_peer.side_effect = [ + None, + { + "id": "legacy-session", + "session_key": legacy_key, + "started_at": 1.0, + }, + ] + + entry = store.get_or_create_session(source) + + assert entry.session_id == "legacy-session" + assert entry.session_key == scoped_key + calls = store._db.find_latest_gateway_session_for_peer.call_args_list + assert [call.kwargs["session_key"] for call in calls] == [ + scoped_key, + legacy_key, + ] + assert all(call.kwargs["chat_id"] is None for call in calls) + assert all(call.kwargs["chat_type"] is None for call in calls) + assert ( + store._db.record_gateway_session_peer.call_args.kwargs["session_key"] + == scoped_key + ) + + class TestWhatsAppSessionKeyConsistency: """Regression: WhatsApp session keys must collapse JID/LID aliases to a single stable identity for both DM chat_ids and group participant_ids.""" diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index cd35c001c946c..d59f15e98d826 100644 --- a/tests/gateway/test_slack.py +++ b/tests/gateway/test_slack.py @@ -287,6 +287,87 @@ class TestSlashCommandSessionIsolation: assert event.source.thread_id == "1700000000.123456" +class TestSlackWorkspaceCollisionIsolation: + @pytest.mark.asyncio + async def test_same_ids_in_two_workspaces_are_both_delivered(self, adapter): + from gateway.session import build_session_key + + team_one, team_two = AsyncMock(), AsyncMock() + team_one.users_info = AsyncMock( + return_value={"user": {"profile": {"display_name": "Alice"}}} + ) + team_two.users_info = AsyncMock( + return_value={"user": {"profile": {"display_name": "Bob"}}} + ) + adapter._team_clients.update({"T_ONE": team_one, "T_TWO": team_two}) + + event = { + "text": "same Slack-local ids", + "user": "U_SHARED", + "channel": "D_SHARED", + "channel_type": "im", + "ts": "171.000", + } + await adapter._handle_slack_message(event, {"team_id": "T_ONE"}) + await adapter._handle_slack_message(event, {"team_id": "T_TWO"}) + + assert adapter.handle_message.await_count == 2 + first = adapter.handle_message.await_args_list[0].args[0] + second = adapter.handle_message.await_args_list[1].args[0] + assert first.source.scope_id == "T_ONE" + assert second.source.scope_id == "T_TWO" + assert build_session_key(first.source) != build_session_key(second.source) + assert adapter._channel_teams["D_SHARED"] == {"T_ONE", "T_TWO"} + assert "D_SHARED" not in adapter._channel_team + + @pytest.mark.asyncio + async def test_same_ids_route_outbound_through_each_workspace_client(self, adapter): + one, two = AsyncMock(), AsyncMock() + one.chat_postMessage = AsyncMock(return_value={"ts": "171.000"}) + two.chat_postMessage = AsyncMock(return_value={"ts": "171.000"}) + adapter._team_clients.update({"T_ONE": one, "T_TWO": two}) + + await adapter.send( + "D_SHARED", "one", metadata={"scope_id": "T_ONE"} + ) + await adapter.send( + "D_SHARED", "two", metadata={"slack_team_id": "T_TWO"} + ) + + one.chat_postMessage.assert_awaited_once_with( + channel="D_SHARED", text="one", mrkdwn=True + ) + two.chat_postMessage.assert_awaited_once_with( + channel="D_SHARED", text="two", mrkdwn=True + ) + assert ("T_ONE", "171.000") in adapter._bot_message_ts + assert ("T_TWO", "171.000") in adapter._bot_message_ts + + @pytest.mark.asyncio + async def test_same_ids_keep_slash_contexts_workspace_scoped(self, adapter): + import time + from plugins.platforms.slack.adapter import _slash_user_id + + for team_id in ("T_ONE", "T_TWO"): + adapter._slash_command_contexts[ + (team_id, "C_SHARED", "U_SHARED") + ] = { + "response_url": f"https://hooks.slack.com/{team_id}", + "ts": time.monotonic(), + } + + token = _slash_user_id.set("U_SHARED") + try: + first = adapter._pop_slash_context("C_SHARED", "T_ONE") + second = adapter._pop_slash_context("C_SHARED", "T_TWO") + finally: + _slash_user_id.reset(token) + + assert first["response_url"].endswith("T_ONE") + assert second["response_url"].endswith("T_TWO") + assert adapter._slash_command_contexts == {} + + # --------------------------------------------------------------------------- # TestAppMentionHandler # --------------------------------------------------------------------------- @@ -4797,7 +4878,7 @@ class TestThreadReplyHandling: ): """Thread replies without mention should be processed if there's an active session.""" # Simulate an active session for this thread - session_key = "agent:main:slack:group:C123:123.000:U_USER" + session_key = "agent:main:slack:group:T_TEAM:C123:123.000:U_USER" mock_session_store._entries = {session_key: MagicMock()} event = { @@ -4906,7 +4987,7 @@ class TestThreadReplyHandling: ): """Thread replies with @mention should still strip the bot ID.""" # Even with a session, mentions should be stripped - session_key = "agent:main:slack:group:C123:123.000:U_USER" + session_key = "agent:main:slack:group:T_TEAM:C123:123.000:U_USER" mock_session_store._entries = {session_key: MagicMock()} event = {