diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 6ad6b01b8a..c3924d774c 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -125,6 +125,44 @@ def _release_singleton_lock(handle) -> None: pass +def _wake_scope_id(adapter: Any, sub: dict) -> Optional[str]: + """Return the tenant scope (Slack workspace) a subscription's wake keys to. + + ``build_session_key()`` includes ``SessionSource.scope_id`` on platforms + where one bot serves several isolated tenants, so a wake source must carry + the same scope as inbound messages from that chat to resolve to the same + session. + + The subscription's persisted ``delivery_metadata`` wins over the adapter's + live chat → scope mapping, because it records the scope the subscription was + created from; the mapping only covers rows that stored no metadata. ``None`` + means the chat has no scope, which is what an unscoped platform's key + contains. + """ + delivery_meta = sub.get("delivery_metadata") + if isinstance(delivery_meta, dict): + for key in ("scope_id", "slack_team_id", "team_id"): + value = delivery_meta.get(key) + if value: + return str(value) + resolver = getattr(adapter, "scope_id_for_chat", None) + if callable(resolver): + try: + resolved = resolver(str(sub.get("chat_id") or "")) + except Exception as exc: + # An adapter-side lookup failure yields no scope, never an error. + logger.debug( + "kanban notifier: scope lookup failed for chat %s: %s", + sub.get("chat_id"), + exc, + exc_info=True, + ) + return None + if resolved: + return str(resolved) + return None + + class GatewayKanbanWatchersMixin: """Kanban watcher / notifier / dispatcher loops for GatewayRunner.""" @@ -791,6 +829,7 @@ class GatewayKanbanWatchersMixin: user_id=sub.get("user_id"), user_id_alt=sub.get("user_id_alt"), profile=sub_profile or None, + scope_id=_wake_scope_id(adapter, sub), ) # deliver_wake preserves the synthetic # MessageEvent/handle_message path for diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index 691e25cbe4..50d215ea65 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -2401,6 +2401,20 @@ class SlackAdapter(BasePlatformAdapter): """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 scope_id_for_chat(self, chat_id: str) -> Optional[str]: + """Return the workspace (team) id that owns ``chat_id``. + + Reads the channel → workspace map maintained by + ``_remember_channel_team``. Returns ``None`` for unknown channels and + for channels claimed by more than one workspace (which that map drops), + so callers get no scope rather than a wrong one. + """ + if not chat_id: + return None + channel_team = getattr(self, "_channel_team", None) or {} + team_id = channel_team.get(str(chat_id)) + return str(team_id) if team_id else None + def _get_client(self, chat_id: str, team_id: Optional[str] = None) -> Any: """Return the workspace-specific WebClient for a channel.""" if team_id and team_id in self._team_clients: diff --git a/tests/gateway/test_kanban_wake_scope.py b/tests/gateway/test_kanban_wake_scope.py new file mode 100644 index 0000000000..0077acd154 --- /dev/null +++ b/tests/gateway/test_kanban_wake_scope.py @@ -0,0 +1,230 @@ +"""Kanban wake events must key to the same session as inbound messages. + +Slack session keys include the workspace id, so the wake source the notifier +rebuilds from a subscription row must carry it too. The contract asserted here: +the key built from the wake source byte-matches the key built from an inbound +source for the same conversation, a scope-less key does not, and platforms +without tenant scoping keep their exact key shape. +""" + +import asyncio +from dataclasses import replace +from unittest.mock import AsyncMock, MagicMock + +from gateway.config import Platform, PlatformConfig +from gateway.kanban_watchers import _wake_scope_id +from gateway.run import GatewayRunner +from gateway.session import build_session_key +from hermes_cli import kanban_db as kb +from plugins.platforms.slack.adapter import SlackAdapter + +TEAM = "T0B8U2M6NRE" +CHANNEL = "C0BCDG3H66P" +THREAD = "1720000000.000100" +USER = "U0BCE4NRVKN" + + +class UnscopedAdapter: + """Push-capable adapter for a platform without tenant scoping.""" + + def __init__(self): + self.sent = [] + self.handled = [] + + async def send(self, chat_id, text, metadata=None): + self.sent.append({"chat_id": chat_id, "text": text, "metadata": metadata or {}}) + + async def handle_message(self, event): + self.handled.append(event) + + +def _slack_adapter(channel_team=None): + """Real SlackAdapter with only its I/O stubbed.""" + adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake-token")) + adapter._app = MagicMock() + adapter._app.client = AsyncMock() + adapter._running = True + adapter.send = AsyncMock() + adapter.handle_message = AsyncMock() + if channel_team: + adapter._channel_team.update(channel_team) + return adapter + + +def _runner(adapter, platform=Platform.SLACK): + runner = GatewayRunner.__new__(GatewayRunner) + runner._running = True + runner.adapters = {platform: adapter} + runner._kanban_sub_fail_counts = {} + # A gateway whose dispatcher owns the singleton lock. + runner._kanban_dispatcher_lock_handle = object() + return runner + + +async def _one_notifier_tick(monkeypatch, runner): + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + if delay == 5: + return None + runner._running = False + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + await runner._kanban_notifier_watcher(interval=1) + + +def _completed_subscription(**sub_kwargs): + conn = kb.connect() + try: + tid = kb.create_task( + conn, + title="wake scope", + assignee="worker", + session_id="origin-session", + ) + kb.add_notify_sub(conn, task_id=tid, **sub_kwargs) + kb.complete_task(conn, tid, summary="done") + return tid + finally: + conn.close() + + +def _wake_source_from(adapter): + assert adapter.handle_message.await_count == 1, ( + f"expected exactly one wake injection, got {adapter.handle_message.await_count}" + ) + return adapter.handle_message.await_args.args[0].source + + +def test_slack_wake_resumes_the_creators_workspace_scoped_session(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope.db")) + kb.init_db() + _completed_subscription( + platform="slack", + chat_id=CHANNEL, + chat_type="group", + thread_id=THREAD, + user_id=USER, + # Slack sources are stamped with slack_team_id when subscribing. + delivery_metadata={"slack_team_id": TEAM, "thread_id": THREAD}, + ) + + adapter = _slack_adapter() + asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter))) + + wake = _wake_source_from(adapter) + assert wake.scope_id == TEAM + + inbound = adapter.build_source( + chat_id=CHANNEL, + chat_type="group", + user_id=USER, + thread_id=THREAD, + scope_id=TEAM, + ) + wake_key = build_session_key(wake) + assert wake_key == build_session_key(inbound) + assert TEAM in wake_key + # A scope-less source keys to a different session for the same chat. + assert build_session_key(replace(wake, scope_id=None, guild_id=None)) != wake_key + + +def test_slack_wake_falls_back_to_the_adapter_channel_workspace_map(tmp_path, monkeypatch): + """Subscriptions that stored no workspace resolve it from the adapter.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-fallback.db")) + kb.init_db() + _completed_subscription( + platform="slack", + chat_id=CHANNEL, + chat_type="group", + thread_id=THREAD, + delivery_metadata={"thread_id": THREAD, "chat_type": "group"}, + ) + + adapter = _slack_adapter(channel_team={CHANNEL: TEAM}) + asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter))) + + wake = _wake_source_from(adapter) + assert wake.scope_id == TEAM + assert TEAM in build_session_key(wake) + + +def test_unknown_channel_keeps_the_previous_unscoped_wake(tmp_path, monkeypatch): + """An unresolvable workspace yields an unscoped key, not a wrong scope.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-unknown.db")) + kb.init_db() + _completed_subscription( + platform="slack", + chat_id=CHANNEL, + chat_type="group", + ) + + adapter = _slack_adapter() + asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter))) + + wake = _wake_source_from(adapter) + assert wake.scope_id is None + assert build_session_key(wake) == f"agent:main:slack:group:{CHANNEL}" + + +def test_unscoped_platform_wake_key_is_byte_identical(tmp_path, monkeypatch): + """Platforms without tenant scoping must keep their exact key shape.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-telegram.db")) + kb.init_db() + _completed_subscription( + platform="telegram", + chat_id="chat-dm", + chat_type="dm", + ) + + adapter = UnscopedAdapter() + asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter, Platform.TELEGRAM))) + + assert len(adapter.handled) == 1 + wake = adapter.handled[0].source + assert wake.scope_id is None + assert build_session_key(wake) == "agent:main:telegram:dm:chat-dm" + + +def test_wake_scope_id_prefers_persisted_metadata_over_the_adapter_map(): + """Persisted metadata wins; the adapter map is only a fallback.""" + adapter = SlackAdapter.__new__(SlackAdapter) + adapter._channel_team = {CHANNEL: "T_STALE"} + + assert _wake_scope_id( + adapter, {"chat_id": CHANNEL, "delivery_metadata": {"slack_team_id": TEAM}} + ) == TEAM + assert _wake_scope_id(adapter, {"chat_id": CHANNEL}) == "T_STALE" + assert _wake_scope_id(adapter, {"chat_id": "C_OTHER"}) is None + + +def test_wake_scope_id_degrades_when_the_adapter_lookup_raises(): + class Exploding: + def scope_id_for_chat(self, chat_id): + raise RuntimeError("adapter state gone") + + assert _wake_scope_id(Exploding(), {"chat_id": CHANNEL}) is None + + +def test_wake_scope_id_is_none_for_adapters_without_the_hook(): + """Adapters that don't resolve scopes leave the wake unscoped.""" + assert _wake_scope_id(UnscopedAdapter(), {"chat_id": CHANNEL}) is None + + +def test_slack_adapter_reports_the_channel_workspace(): + adapter = SlackAdapter.__new__(SlackAdapter) + adapter._channel_team = {CHANNEL: TEAM} + + assert adapter.scope_id_for_chat(CHANNEL) == TEAM + assert adapter.scope_id_for_chat("C_UNKNOWN") is None + assert adapter.scope_id_for_chat("") is None + + +def test_slack_adapter_reports_no_scope_for_ambiguous_channels(): + """A channel claimed by two workspaces resolves to no scope.""" + adapter = _slack_adapter() + adapter._remember_channel_team("D_SHARED", "T_ONE") + adapter._remember_channel_team("D_SHARED", "T_TWO") + + assert adapter.scope_id_for_chat("D_SHARED") is None