fix(gateway): multiplexed bots stop sharing one session lane (#88404, salvage #88437)

contributors/emails/lucasxavier926@gmail.com

@@ -0,0 +1,2 @@
+69k4xmdfm2-blip
+# multiplex adapter key namespace

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:<MagicMock ...>:``.
+        """
+        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 ""

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

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:

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

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:

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:

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

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:

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:

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:

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:<uid>`` 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:<MagicMock ...>:`` 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"