diff --git a/contributors/emails/lucasxavier926@gmail.com b/contributors/emails/lucasxavier926@gmail.com new file mode 100644 index 000000000000..54ac6128ef4f --- /dev/null +++ b/contributors/emails/lucasxavier926@gmail.com @@ -0,0 +1,2 @@ +69k4xmdfm2-blip +# multiplex adapter key namespace diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index d3efafaeb8b0..070db6e4e2e8 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -3082,6 +3082,14 @@ def __init__(self, config: PlatformConfig, platform: Platform): self._post_delivery_callbacks: Dict[str, Any] = {} self._expected_cancelled_tasks: set[asyncio.Task] = set() self._busy_session_handler: Optional[Callable[[MessageEvent, str], Awaitable[bool]]] = None + # Owning profile for a multiplexed secondary adapter, installed by + # ``GatewayRunner._configure_profile_adapter``. Adapter-level session + # keys must carry the profile namespace, but ``source.profile`` is only + # stamped later by the runner's profile message handler — so at adapter + # ingress every bot in a multiplexed gateway would otherwise derive the + # same ``agent:main:`` key (see ``_session_key_profile``). ``None`` on a + # primary/single-profile adapter, which keeps the legacy namespace. + self._owner_profile: Optional[str] = None # Optional authorization check, registered by GatewayRunner. Used by # adapters that fetch external context (e.g. Slack thread history) to # mark senders not on the allowlist as unverified in LLM context, @@ -3746,6 +3754,57 @@ def set_session_store(self, session_store: Any) -> None: thread replies without explicit mentions). """ self._session_store = session_store + + def set_owner_profile(self, profile_name: Optional[str]) -> None: + """Declare which multiplex profile owns this adapter. + + Installed by ``GatewayRunner._configure_profile_adapter`` for secondary + profiles. Read by :meth:`_session_key_profile` so adapter-level keys + land in this profile's namespace instead of the shared ``agent:main:``. + """ + name = (profile_name or "").strip() or None + self._owner_profile = None if name == "default" else name + + def _session_key_profile(self, source: Optional[Any] = None) -> Optional[str]: + """Resolve the profile namespace for an adapter-derived session key. + + Adapter ingress runs BEFORE the runner stamps ``source.profile`` + (``_make_profile_message_handler``), so the session store's resolver + falls back to the *active* profile and every bot in a multiplexed + gateway derives the same ``agent:main:`` key. Batching dicts, + ``_active_sessions`` and the busy-session guard are keyed on that + string, so two profiles sharing a chat id — which is EVERY Telegram DM, + where ``chat.id`` is the user's own id — collide on one lane. + + Resolution order: + 1. ``source.profile`` when already stamped (relay/connector ingress). + 2. ``self._owner_profile`` — this adapter's own credential owner. + 3. The session store's resolver (active profile / no-multiplex None). + + ``getattr`` throughout: adapters are routinely constructed without + ``BasePlatformAdapter.__init__`` (``object.__new__`` in tests, subclasses + that build their own state), so no attribute here may be assumed to + exist — see the ``object.__new__`` pitfall in AGENTS.md. Every candidate + is also type-checked: a duck-typed/mock session store returns a truthy + non-string from ``_resolve_profile_for_key``, which would otherwise be + interpolated straight into the key as ``agent::``. + """ + for candidate in ( + getattr(source, "profile", None) if source is not None else None, + getattr(self, "_owner_profile", None), + ): + if isinstance(candidate, str) and candidate.strip(): + return candidate + store = getattr(self, "_session_store", None) + resolver = getattr(store, "_resolve_profile_for_key", None) if store else None + if callable(resolver): + try: + resolved = resolver(source) + except Exception: + return None + if isinstance(resolved, str) and resolved.strip(): + return resolved + return None def _history_media_paths_for_session(self, session_key: str) -> Optional[set]: """Return media paths already delivered in prior turns of this session. @@ -6003,12 +6062,11 @@ async def handle_message(self, event: MessageEvent) -> None: if needs_topic_recovery: await asyncio.to_thread(self._apply_topic_recovery, event) - _sk_store = getattr(self, "_session_store", None) session_key = build_session_key( event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=_sk_store._resolve_profile_for_key(event.source) if _sk_store else None, + profile=self._session_key_profile(event.source), ) expected_session_key = str( (event.metadata or {}).get("gateway_session_key") or "" diff --git a/gateway/run.py b/gateway/run.py index 07952e208528..6952198e200e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -15183,6 +15183,15 @@ def _configure_profile_adapter( self._make_profile_fatal_error_handler(profile_name, platform) ) adapter.set_session_store(self.session_store) + # Declare credential ownership BEFORE any inbound event can be handled. + # Adapter-level session keys (text/media batching, _active_sessions, the + # busy guard) are derived at ingress, before _make_profile_message_handler + # stamps source.profile — without this every secondary bot would key into + # the default profile's `agent:main:` lane and share it (see + # BasePlatformAdapter._session_key_profile). + _set_owner = getattr(adapter, "set_owner_profile", None) + if callable(_set_owner): + _set_owner(profile_name) adapter.set_busy_session_handler( self._make_profile_busy_session_handler(profile_name) ) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index ef0cdd14d74d..ad1a6ab1191f 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -8564,7 +8564,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) def _enqueue_text_event(self, event: MessageEvent) -> None: diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py index 5b2e4ec4bd17..d2f3352657cb 100644 --- a/plugins/platforms/feishu/adapter.py +++ b/plugins/platforms/feishu/adapter.py @@ -3428,6 +3428,7 @@ def _media_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), + profile=self._session_key_profile(event.source), ) return f"{session_key}:media:{event.message_type.value}" @@ -3736,7 +3737,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) @staticmethod diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index d6600d611942..0da6f37c962c 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -4276,7 +4276,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: thread_sessions_per_user=self.config.extra.get( "thread_sessions_per_user", False ), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) def _enqueue_text_event(self, event: MessageEvent) -> None: diff --git a/plugins/platforms/raft/adapter.py b/plugins/platforms/raft/adapter.py index 7e0494fa9d9d..7ad661f91060 100644 --- a/plugins/platforms/raft/adapter.py +++ b/plugins/platforms/raft/adapter.py @@ -741,6 +741,7 @@ async def handle_message(self, event: MessageEvent) -> None: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), + profile=self._session_key_profile(event.source), ) if session_key in self._active_sessions: diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index 50d215ea65e3..e017838fce82 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -8363,6 +8363,7 @@ def _build_thread_session_key( source, group_sessions_per_user=gspu, thread_sessions_per_user=tspu, + profile=self._session_key_profile(source), ) except Exception: return None diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 84332a69eefe..b1f9c67db1c9 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -9626,7 +9626,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) def _enqueue_text_event(self, event: MessageEvent) -> None: @@ -9731,6 +9731,7 @@ def _photo_batch_key(self, event: MessageEvent, msg: Message) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), + profile=self._session_key_profile(event.source), ) media_group_id = getattr(msg, "media_group_id", None) if media_group_id: diff --git a/plugins/platforms/wecom/adapter.py b/plugins/platforms/wecom/adapter.py index 4770ec721104..0a5fa168bbd0 100644 --- a/plugins/platforms/wecom/adapter.py +++ b/plugins/platforms/wecom/adapter.py @@ -608,7 +608,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) def _enqueue_text_event(self, event: MessageEvent) -> None: diff --git a/plugins/platforms/whatsapp/adapter.py b/plugins/platforms/whatsapp/adapter.py index 7f16158d4c14..fd2138724442 100644 --- a/plugins/platforms/whatsapp/adapter.py +++ b/plugins/platforms/whatsapp/adapter.py @@ -1399,7 +1399,7 @@ def _text_batch_key(self, event: MessageEvent) -> str: event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=event.source.profile, + profile=self._session_key_profile(event.source), ) def _enqueue_text_event(self, event: MessageEvent) -> None: diff --git a/tests/gateway/test_multiplex_adapter_session_key_namespace.py b/tests/gateway/test_multiplex_adapter_session_key_namespace.py new file mode 100644 index 000000000000..3934da7fc806 --- /dev/null +++ b/tests/gateway/test_multiplex_adapter_session_key_namespace.py @@ -0,0 +1,195 @@ +"""Regression tests: adapter-level session keys must carry the profile (#88391). + +Incident shape (Aug 2026, local install, two Telegram bots on one multiplexed +gateway): a Telegram private chat reports the user's own id as ``chat.id``, so +every bot in the multiplexer sees an identical ``chat_id`` for the same human. + +``BasePlatformAdapter.handle_message`` and each adapter's ``_text_batch_key`` / +``_photo_batch_key`` derive a session key at INGRESS — before +``GatewayRunner._make_profile_message_handler`` stamps ``source.profile``. The +key therefore fell back to the *active* profile's namespace, so both bots +produced ``agent:main:telegram:dm:`` and shared one lane: the text-batching +dict, ``_active_sessions`` and the busy-session guard are all keyed on that +string. Observed in production logs: 60 flushes, zero carrying ``agent:medicina:``. + +Fix under test: ``set_owner_profile`` records credential ownership on the +adapter, and ``_session_key_profile`` resolves the namespace as +``source.profile`` → ``_owner_profile`` → session-store resolver, so a secondary +adapter keys into its own namespace even before the runner stamps the source. +""" + +import pytest + +from gateway.config import Platform +from gateway.platforms.base import BasePlatformAdapter +from gateway.session import SessionSource, build_session_key + + +UID = "8693894969" + + +def _source(profile=None): + """Telegram DM shape: chat_id == user_id, thread_id None — same for every bot.""" + return SessionSource( + platform=Platform.TELEGRAM, + chat_id=UID, + chat_type="dm", + user_id=UID, + user_name="Lucas", + profile=profile, + ) + + +class _Adapter(BasePlatformAdapter): + """Minimal concrete adapter — only the key-derivation seam is under test.""" + + name = "stub" + + def __init__(self): + self._session_store = None + self._owner_profile = None + + # BasePlatformAdapter declares these abstract; none is exercised here. + async def connect(self): ... + async def disconnect(self): ... + async def send(self, *a, **k): ... + async def send_message(self, *a, **k): ... + async def get_chat_info(self, *a, **k): ... + async def start_listening(self): ... + + +class _Store: + """Stand-in for GatewaySessionStore's namespace resolver.""" + + def __init__(self, active="default", multiplex=True): + self._active = active + self._multiplex = multiplex + + def _resolve_profile_for_key(self, source=None): + if not self._multiplex: + return None + if source is not None and getattr(source, "profile", None): + return source.profile + return self._active + + +class TestOwnerProfileKeying: + def test_secondary_adapter_keys_into_own_namespace(self): + """The bug: unstamped source + active profile 'default' collapsed a + secondary bot's key onto agent:main:.""" + a = _Adapter() + a._session_store = _Store(active="default") + a.set_owner_profile("medicina") + key = build_session_key(_source(), profile=a._session_key_profile(_source())) + assert key.startswith("agent:medicina:"), key + + def test_two_bots_same_chat_do_not_collide(self): + """Two adapters, one chat id: the keys must differ or the batching dict, + _active_sessions and the busy guard merge both bots into one lane.""" + default_a, secondary_a = _Adapter(), _Adapter() + default_a._session_store = _Store(active="default") + secondary_a._session_store = _Store(active="default") + secondary_a.set_owner_profile("medicina") + src = _source() + k_default = build_session_key(src, profile=default_a._session_key_profile(src)) + k_secondary = build_session_key(src, profile=secondary_a._session_key_profile(src)) + assert k_default != k_secondary, f"both bots share one lane: {k_default}" + assert k_default.startswith("agent:main:") + assert k_secondary.startswith("agent:medicina:") + + def test_stamped_source_wins_over_owner(self): + """Connector/relay ingress stamps source.profile — it must take priority + so a shared-ingress adapter routes per event, not per credential.""" + a = _Adapter() + a._session_store = _Store(active="default") + a.set_owner_profile("medicina") + assert a._session_key_profile(_source(profile="finances")) == "finances" + + def test_primary_adapter_unchanged(self): + """No owner + active default ⇒ legacy agent:main:, byte-identical.""" + a = _Adapter() + a._session_store = _Store(active="default") + key = build_session_key(_source(), profile=a._session_key_profile(_source())) + assert key == build_session_key(_source()) + assert key.startswith("agent:main:") + + def test_single_profile_gateway_unchanged(self): + """Multiplexing off ⇒ resolver returns None ⇒ legacy namespace.""" + a = _Adapter() + a._session_store = _Store(multiplex=False) + assert a._session_key_profile(_source()) is None + key = build_session_key(_source(), profile=a._session_key_profile(_source())) + assert key == build_session_key(_source()) + + def test_owner_default_is_normalized_to_none(self): + """'default' must collapse to None, not produce 'agent:default:'.""" + a = _Adapter() + a._session_store = None + a.set_owner_profile("default") + assert a._owner_profile is None + assert build_session_key(_source(), profile=a._session_key_profile(_source())) \ + == build_session_key(_source()) + + @pytest.mark.parametrize("blank", [None, "", " "]) + def test_blank_owner_is_none(self, blank): + a = _Adapter() + a._session_store = None + a.set_owner_profile(blank) + assert a._owner_profile is None + + def test_owner_used_when_store_absent(self): + """A secondary adapter must not depend on the store being installed.""" + a = _Adapter() + a._session_store = None + a.set_owner_profile("medicina") + assert a._session_key_profile(_source()) == "medicina" + + def test_adapter_without_base_init_does_not_raise(self): + """Adapters are routinely built via ``object.__new__`` (tests) or by a + subclass that never calls ``BasePlatformAdapter.__init__``, so + ``_owner_profile``/``_session_store`` may be entirely absent. Resolution + must degrade to the legacy namespace instead of AttributeError — this + broke every text-batching suite on the first cut of the fix. + """ + bare = object.__new__(_Adapter) + assert not hasattr(bare, "_owner_profile") + assert bare._session_key_profile(_source()) is None + assert build_session_key(_source(), profile=bare._session_key_profile(_source())) \ + == build_session_key(_source()) + + def test_resolver_exception_falls_back_to_none(self): + class _Boom: + def _resolve_profile_for_key(self, source=None): + raise RuntimeError("store unavailable") + + a = _Adapter() + a._session_store = _Boom() + assert a._session_key_profile(_source()) is None + + def test_non_string_resolver_result_is_rejected(self): + """A duck-typed/mock session store returns a truthy non-string, which + would be interpolated into the key as ``agent::`` and + corrupt every lookup. Real regression: it broke Slack's thread-reply + suite, whose fixture store is a bare MagicMock. + """ + from unittest.mock import MagicMock + + a = _Adapter() + a._session_store = MagicMock() # resolver returns a MagicMock + assert a._session_key_profile(_source()) is None + assert build_session_key(_source(), profile=a._session_key_profile(_source())) \ + == build_session_key(_source()) + + @pytest.mark.parametrize("junk", [123, object(), ["medicina"], b"medicina", " "]) + def test_non_string_or_blank_owner_is_ignored(self, junk): + a = _Adapter() + a._session_store = None + a._owner_profile = junk # bypass the setter's normalisation + assert a._session_key_profile(_source()) is None + + def test_no_source_still_resolves_owner(self): + """Some call sites derive a key without an event (idle/wake paths).""" + a = _Adapter() + a._session_store = _Store(active="default") + a.set_owner_profile("medicina") + assert a._session_key_profile(None) == "medicina" diff --git a/tests/gateway/test_multiplex_busy_input_mode.py b/tests/gateway/test_multiplex_busy_input_mode.py index 3a2f901df759..5e3d8af1bd92 100644 --- a/tests/gateway/test_multiplex_busy_input_mode.py +++ b/tests/gateway/test_multiplex_busy_input_mode.py @@ -249,7 +249,12 @@ async def test_secondary_adapter_busy_guard_stamps_profile_before_resolving_mode "steer", ) event = _event(profile=None) - adapter_session_key = build_session_key(event.source) + # Seed the lane the adapter itself derives. A profile-owned adapter keys its + # own _active_sessions in its own namespace (agent:research:...) — see + # BasePlatformAdapter._session_key_profile. Seeding the unstamped + # agent:main: key here asserted the pre-fix behaviour, where every profile's + # adapter collapsed onto the default lane. + adapter_session_key = build_session_key(event.source, profile="research") adapter._active_sessions[adapter_session_key] = asyncio.Event() routed_source = _event(profile="research").source