fix(kanban): key terminal-event wakes to the creator's workspace scope

Slack session keys include the workspace id since #70190, but the kanban
notifier rebuilds the wake source from a subscription row that has no scope
column, so every terminal-event wake keyed without the workspace.

The legacy-key adoption shipped in the same change (`_legacy_slack_session_key`,
`_recovered_row_matches_source_scope`) resolves that unscoped key onto the same
session_id, so the wake passes the busy guards that are keyed by routing key
(`_active_sessions`, `_running_agents`) and only collides afterwards, on session
id, under the per-session turn lease (#64934) — which serializes it behind the
live turn's flush. On a live Slack gateway that shows up as a duplicate run on
one task plus 400+s of waiting before the woken turn starts.

Same failure mode as #56580 / #72191 (chat_type), one field over, and it needs
no schema change: `_thread_metadata_for_source()` already stamps
`slack_team_id`, the notify subscription persists that dict as
`delivery_metadata`, and the notifier already unpacks it. Rows written by
`kanban_tools._maybe_auto_subscribe` carry no workspace, so fall back to the
adapter's channel → workspace map via `scope_id_for_chat()`, read with getattr
so adapters opt in and unscoped platforms' keys stay byte-identical. Slack
answers it from `_remember_channel_team`, which drops channels claimed by two
workspaces, so an unknown or ambiguous channel degrades to today's behavior
instead of guessing wrong.

Also adds the contributor email mapping the attribution check requires.

Co-authored-by: Junie <junie@jetbrains.com>
This commit is contained in:
Nikita Barkov
2026-08-04 12:37:19 +02:00
committed by Teknium
parent 6a198f8a12
commit 5d3f75110a
3 changed files with 283 additions and 0 deletions

View File

@@ -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

View File

@@ -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:

View File

@@ -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