fix(slack): isolate workspace-local routing
This commit is contained in:
Jordan Hubbard
Teknium
parent
f50c3d904c
commit
a60b00e12d
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
Loading…
Reference in New Issue