From 06be0e69b6caa29a5bd6d8effdc64d682fd9d591 Mon Sep 17 00:00:00 2001 From: Bob Date: Tue, 21 Jul 2026 17:09:47 -0400 Subject: [PATCH] fix(gateway): namespace Slack sessions by workspace --- gateway/session.py | 91 ++++++++++-- tests/gateway/test_session.py | 264 ++++++++++++++++++++++++++++++++++ 2 files changed, 343 insertions(+), 12 deletions(-) diff --git a/gateway/session.py b/gateway/session.py index 31829cf5eee6..fefd7346a767 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1608,6 +1608,34 @@ class SessionStore: claimed.add(legacy_key) return True + @staticmethod + def _recovered_row_matches_source_scope( + recovered: Dict[str, Any], source: SessionSource + ) -> bool: + """Reject recovered rows whose recorded origin belongs to another workspace. + + Slack group/channel rows recorded with an origin_json carry the + workspace (scope_id) they were created under. A workspace-scoped + lookup must not adopt a row another team recorded — even via the + legacy-key fallback — unless the recorded origin names the same + workspace. Rows without a parseable origin are rejected for scoped + sources: an unattributable transcript is precisely the ambiguity + this guard exists to avoid. + """ + if ( + source.platform != Platform.SLACK + or source.chat_type == "dm" + or not source.scope_id + ): + return True + try: + origin = json.loads(recovered.get("origin_json") or "") + except (TypeError, ValueError): + return False + if not isinstance(origin, dict): + return False + return origin.get("scope_id", origin.get("guild_id")) == source.scope_id + def _create_entry_from_recovered_row( self, *, @@ -1702,6 +1730,8 @@ class SessionStore: migrated_legacy = bool(recovered) if not recovered: return None + if not self._recovered_row_matches_source_scope(recovered, source): + return None if not self._recovered_row_allowed_for_active_profile( requested_session_key=session_key, recovered=recovered, @@ -1733,7 +1763,9 @@ class SessionStore: ) return entry - def _query_recoverable_session(self, *, session_key, source, now): + def _query_recoverable_session( + self, *, session_key, source, now, lookup_session_key=None + ): """DB-only half of _recover_session_from_db (no lock needed). Returns a SessionEntry or None. Caller assigns _entries[key] under lock. @@ -1758,6 +1790,8 @@ class SessionStore: migrated_legacy = bool(recovered) if not isinstance(recovered, dict): return None + if not self._recovered_row_matches_source_scope(recovered, source): + return None if not self._recovered_row_allowed_for_active_profile( requested_session_key=session_key, recovered=recovered, @@ -2153,22 +2187,39 @@ class SessionStore: # 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. + # + # Adoption policy (composed from #20583/#66398 and #68925): + # - The legacy entry's recorded origin names a workspace → migrate + # only when it matches the incoming workspace (precise). + # - Scope-less origin, DM → first workspace claims it once + # (claim-once): a 1:1 DM has a single human peer, so continuity + # across the key-format change outweighs the ambiguity risk. + # - Scope-less origin, channel/group → refuse: channel ids collide + # across workspaces and a shared transcript leaking to a second + # tenant is exactly the bug this fix removes. 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 + legacy_entry = self._entries.get(legacy_key) + if session_key not in self._entries and legacy_entry is not None: + origin_scope = ( + getattr(legacy_entry.origin, "scope_id", None) + if legacy_entry.origin is not None + else None + ) + if origin_scope is not None: + adopt = origin_scope == source.scope_id + else: + adopt = source.chat_type == "dm" + if adopt 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( @@ -2313,6 +2364,15 @@ class SessionStore: recovered = self._query_recoverable_session( session_key=session_key, source=source, now=now, ) + recovered_from_legacy = False + if recovered is None and legacy_session_key is not None: + recovered = self._query_recoverable_session( + session_key=session_key, + lookup_session_key=legacy_session_key, + source=source, + now=now, + ) + recovered_from_legacy = recovered is not None if recovered is not None: with self._lock: published = self._entries.get(session_key) @@ -2321,6 +2381,13 @@ class SessionStore: published = recovered entry = published _needs_save = True + if recovered_from_legacy and published is recovered: + self._record_gateway_session_peer( + recovered.session_id, + session_key, + source, + display_name=recovered.display_name, + ) if entry is None: # Create a candidate outside the lock, then publish only if another diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index 86d4d2eeac7c..2913cc7a1d8a 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -2,11 +2,13 @@ import json import pytest from dataclasses import replace +from datetime import datetime from pathlib import Path from unittest.mock import patch, MagicMock from gateway.config import Platform, HomeChannel, GatewayConfig, PlatformConfig from gateway.platforms.base import MessageEvent from gateway.session import ( + SessionEntry, SessionSource, SessionStore, build_session_context, @@ -1431,6 +1433,268 @@ class TestWhatsAppSessionKeyConsistency: assert key == "agent:main:telegram:dm:99:topic-1" +class TestSlackWorkspaceSessionKeys: + def test_same_thread_and_user_in_distinct_workspaces_get_distinct_keys(self): + # Given + first = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_ALPHA", + ) + second = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_BETA", + ) + + # When + first_key = build_session_key(first) + second_key = build_session_key(second) + + # Then + assert first_key == "agent:main:slack:channel:T_ALPHA:C123:1700000000.000001" + assert second_key == "agent:main:slack:channel:T_BETA:C123:1700000000.000001" + assert first_key != second_key + + def test_thread_per_user_isolation_keeps_user_suffix_after_workspace(self): + # Given + source = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_ALPHA", + ) + + # When + key = build_session_key(source, thread_sessions_per_user=True) + + # Then + assert key == "agent:main:slack:channel:T_ALPHA:C123:1700000000.000001:U123" + + def test_dm_key_is_workspace_scoped_when_workspace_is_present(self): + # Given. NOTE: adapted from #68925's original expectation (unscoped + # DM keys). The salvaged #20583/#66398 design scopes DM keys too: + # Slack D... conversation ids are workspace-local, so two workspaces + # can present the same DM id and must not share a session. Scope-less + # DM sources (single-workspace installs) keep byte-identical keys. + source = SessionSource( + platform=Platform.SLACK, + chat_id="D123", + chat_type="dm", + user_id="U123", + scope_id="T_ALPHA", + ) + + # When + key = build_session_key(source) + + # Then + assert key == "agent:main:slack:dm:T_ALPHA:D123" + unscoped = replace(source, scope_id=None, guild_id=None) + assert build_session_key(unscoped) == "agent:main:slack:dm:D123" + + def test_non_slack_key_ignores_scope(self): + # Given + source = SessionSource( + platform=Platform.DISCORD, + chat_id="C123", + chat_type="channel", + user_id="U123", + scope_id="GUILD_ALPHA", + ) + + # When + key = build_session_key(source) + + # Then + assert key == "agent:main:discord:channel:C123:U123" + + def test_matching_workspace_reuses_and_migrates_legacy_routing_entry( + self, tmp_path, monkeypatch + ): + # Given + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + source = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_ALPHA", + ) + legacy_key = "agent:main:slack:channel:C123:1700000000.000001" + legacy_entry = SessionEntry( + session_key=legacy_key, + session_id="legacy-session", + created_at=datetime.now(), + updated_at=datetime.now(), + origin=source, + platform=Platform.SLACK, + chat_type="channel", + ) + (tmp_path / "sessions.json").write_text( + json.dumps({legacy_key: legacy_entry.to_dict()}), encoding="utf-8" + ) + store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + + # When + reused = store.get_or_create_session(source) + + # Then + scoped_key = "agent:main:slack:channel:T_ALPHA:C123:1700000000.000001" + assert reused.session_id == "legacy-session" + assert reused.session_key == scoped_key + assert scoped_key in store._entries + assert legacy_key not in store._entries + + def test_scope_less_legacy_entry_is_not_adopted_by_a_workspace( + self, tmp_path, monkeypatch + ): + # Given + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + legacy_source = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + ) + incoming = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_BETA", + ) + legacy_key = "agent:main:slack:channel:C123:1700000000.000001" + legacy_entry = SessionEntry( + session_key=legacy_key, + session_id="ambiguous-legacy-session", + created_at=datetime.now(), + updated_at=datetime.now(), + origin=legacy_source, + platform=Platform.SLACK, + chat_type="channel", + ) + (tmp_path / "sessions.json").write_text( + json.dumps({legacy_key: legacy_entry.to_dict()}), encoding="utf-8" + ) + store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + + # When + routed = store.get_or_create_session(incoming) + + # Then + assert routed.session_id != "ambiguous-legacy-session" + assert routed.session_key == "agent:main:slack:channel:T_BETA:C123:1700000000.000001" + assert store._entries[legacy_key].session_id == "ambiguous-legacy-session" + + def test_matching_workspace_recovers_legacy_session_from_db( + self, tmp_path, monkeypatch + ): + # Given + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + source = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + scope_id="T_ALPHA", + ) + legacy_key = "agent:main:slack:channel:C123:1700000000.000001" + original = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + original._db.create_session( + session_id="legacy-db-session", + source="slack", + user_id="U_FIRST_PARTICIPANT", + session_key=legacy_key, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + ) + original._db.record_gateway_session_peer( + "legacy-db-session", + source="slack", + user_id="U_FIRST_PARTICIPANT", + session_key=legacy_key, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + origin_json=json.dumps(source.to_dict()), + ) + original.append_to_transcript( + "legacy-db-session", {"role": "user", "content": "legacy context"} + ) + original._db.close() + restarted = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + + # When + recovered = restarted.get_or_create_session(source) + + # Then + assert recovered.session_id == "legacy-db-session" + assert recovered.session_key == "agent:main:slack:channel:T_ALPHA:C123:1700000000.000001" + assert restarted._db.get_session("legacy-db-session")["session_key"] == recovered.session_key + + def test_scope_less_legacy_db_session_is_not_adopted_by_a_workspace( + self, tmp_path, monkeypatch + ): + # Given + import hermes_state + + monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", tmp_path / "state.db") + legacy_source = SessionSource( + platform=Platform.SLACK, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + user_id="U123", + ) + incoming = replace(legacy_source, scope_id="T_BETA") + legacy_key = "agent:main:slack:channel:C123:1700000000.000001" + original = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + original._db.create_session( + session_id="ambiguous-db-session", + source="slack", + user_id="U123", + session_key=legacy_key, + chat_id="C123", + chat_type="channel", + thread_id="1700000000.000001", + ) + original._record_gateway_session_peer( + "ambiguous-db-session", legacy_key, legacy_source + ) + original.append_to_transcript( + "ambiguous-db-session", {"role": "user", "content": "other workspace"} + ) + original._db.close() + restarted = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + + # When + routed = restarted.get_or_create_session(incoming) + + # Then + assert routed.session_id != "ambiguous-db-session" + assert routed.session_key == "agent:main:slack:channel:T_BETA:C123:1700000000.000001" + + class TestWhatsAppIdentifierPublicHelpers: """Contract tests for the public WhatsApp identifier helpers.