mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-24 16:54:43 +00:00
fix(gateway): namespace Slack sessions by workspace
This commit is contained in:
Bob
Teknium
parent
ed67f9aacc
commit
06be0e69b6
2 changed files with 343 additions and 12 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue