fix(gateway): stamp profile before busy/approval checks in handle_message

gateway/platforms/base.py

@@ -2438,6 +2438,15 @@ def set_status_text(self, chat_id: str, text: Optional[str]) -> None:
     def __init__(self, config: PlatformConfig, platform: Platform):
         self.config = config
         self.platform = platform
+        # Set by GatewayRunner for secondary multiplexed profiles (None for the
+        # default profile). Must be known synchronously at construction time —
+        # NOT via the message-handler wrapper (_make_profile_message_handler),
+        # which only stamps event.source.profile *inside* self._message_handler,
+        # too late for handle_message()'s busy/approval/draining checks below,
+        # which run before self._message_handler is ever called and silently
+        # fall back to the default profile's adapter when source.profile is
+        # still unset. See gateway/platforms/base.py::handle_message.
+        self.profile_name: Optional[str] = None
         self._message_handler: Optional[MessageHandler] = None
         # Optional hook (e.g. Telegram DM topic recovery) that rewrites
         # ``event.source.thread_id`` before session keying. Returns the
@@ -4822,6 +4831,19 @@ async def handle_message(self, event: MessageEvent) -> None:
         if not self._message_handler:
             return
 
+        # Stamp the owning profile before any routing decision below reads
+        # it. _make_profile_message_handler() (gateway/run.py) also stamps
+        # this, but only inside self._message_handler — after the
+        # busy/approval/draining checks and session_key build that follow.
+        # Without this early stamp, those checks resolve source.profile as
+        # unset and _adapter_for_source() silently falls back to the default
+        # profile's adapter, misrouting secondary-profile replies.
+        # getattr(..., None): some tests construct adapters via
+        # object.__new__() and skip __init__, so profile_name may not be set.
+        _profile_name = getattr(self, "profile_name", None)
+        if getattr(event, "source", None) is not None and not event.source.profile:
+            event.source.profile = _profile_name
+
         coerce_plaintext_gateway_command(event)
 
         # Rewrite ``event.source.thread_id`` via the installed recovery hook
@@ -4834,6 +4856,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=_profile_name,
         )
 
         # On-entry self-heal: if the adapter still has an _active_sessions

gateway/run.py

@@ -9614,6 +9614,10 @@ def _configure_profile_adapter(
         platform: Platform,
     ) -> None:
         """Install the profile-scoped handlers shared by startup and reconnect."""
+        # Must be set synchronously here, not just via the message-handler
+        # wrapper below — handle_message()'s busy/approval/draining checks
+        # read self.profile_name before self._message_handler ever runs.
+        adapter.profile_name = profile_name
         adapter.set_message_handler(self._make_profile_message_handler(profile_name))
         adapter.set_fatal_error_handler(
             self._make_profile_fatal_error_handler(profile_name, platform)

plugins/platforms/telegram/adapter.py

@@ -8256,11 +8256,16 @@ def _text_batch_key(self, event: MessageEvent) -> str:
         """
         from gateway.session import build_session_key
         self._apply_topic_recovery(event)
+        # event.source.profile isn't stamped yet at this point in the receive
+        # pipeline (that only happens inside handle_message, called later) —
+        # use self.profile_name, which is known synchronously on this adapter.
+        # getattr(..., None): some tests construct adapters via
+        # object.__new__() and skip __init__, so profile_name may not be set.
         return 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=event.source.profile,
+            profile=getattr(self, "profile_name", None),
         )
 
     def _enqueue_text_event(self, event: MessageEvent) -> None:
@@ -8353,10 +8358,15 @@ async def _flush_text_batch(self, key: str) -> None:
     def _photo_batch_key(self, event: MessageEvent, msg: Message) -> str:
         """Return a batching key for Telegram photos/albums."""
         from gateway.session import build_session_key
+        # event.source.profile isn't stamped yet this early in the receive
+        # pipeline — use self.profile_name instead (see _text_batch_key).
+        # getattr(..., None): some tests construct adapters via
+        # object.__new__() and skip __init__, so profile_name may not be set.
         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=getattr(self, "profile_name", None),
         )
         media_group_id = getattr(msg, "media_group_id", None)
         if media_group_id:

tests/gateway/test_multiplex_busy_session_profile_routing.py

@@ -0,0 +1,143 @@
+"""Regression tests: a secondary profile's busy-session state must be keyed
+under its own ``agent:<profile>`` namespace, not the default profile's
+``agent:main`` bucket.
+
+``BasePlatformAdapter.handle_message`` runs its active-session busy check
+(and builds the session key used everywhere downstream) *before* it ever
+calls ``self._message_handler``. ``_make_profile_message_handler`` (see
+``gateway/run.py``) only stamps ``event.source.profile`` *inside* that
+handler -- too late for the busy check above it. Without an adapter-level
+``profile_name`` known synchronously at construction time, every secondary
+profile's busy/pending/debounce state silently collided under ``agent:main``,
+and any busy-session reply for that profile went out through the *default*
+profile's adapter/bot instead of its own.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import types
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+# Minimal telegram stub so importing gateway.platforms.base does not pull in
+# the real python-telegram-bot dependency (mirrors test_active_session_text_merge.py).
+_tg = sys.modules.get("telegram") or types.ModuleType("telegram")
+_tg.constants = sys.modules.get("telegram.constants") or types.ModuleType("telegram.constants")
+_ct = MagicMock()
+_ct.PRIVATE = "private"
+_ct.GROUP = "group"
+_ct.SUPERGROUP = "supergroup"
+_tg.constants.ChatType = _ct
+sys.modules.setdefault("telegram", _tg)
+sys.modules.setdefault("telegram.constants", _tg.constants)
+sys.modules.setdefault("telegram.ext", types.ModuleType("telegram.ext"))
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+)
+from gateway.session import SessionSource, build_session_key
+
+
+def _make_event(text: str, chat_id: str = "12345") -> MessageEvent:
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id=chat_id,
+        chat_type="dm",
+        user_id="u1",
+        user_name=None,
+        thread_id=None,
+    )
+    return MessageEvent(
+        text=text, message_type=MessageType.TEXT, source=source, message_id=f"msg-{text[:8]}"
+    )
+
+
+class _DummyAdapter(BasePlatformAdapter):  # type: ignore[misc]
+    async def connect(self, *, is_reconnect: bool = False):
+        pass
+
+    async def disconnect(self):
+        pass
+
+    async def get_chat_info(self, chat_id):
+        return None
+
+    async def send(self, *args, **kwargs):
+        return SendResult(success=True, message_id="x")
+
+
+def _make_adapter(profile_name: str | None) -> BasePlatformAdapter:
+    """Build a BasePlatformAdapter without running its heavy __init__."""
+    adapter = object.__new__(_DummyAdapter)
+    adapter.config = PlatformConfig(enabled=True, token="***")
+    adapter.platform = Platform.TELEGRAM
+    adapter.profile_name = profile_name
+    adapter._message_handler = AsyncMock(return_value=None)
+    adapter._busy_session_handler = None
+    adapter._active_sessions = {}
+    adapter._pending_messages = {}
+    adapter._session_tasks = {}
+    adapter._background_tasks = set()
+    adapter._post_delivery_callbacks = {}
+    adapter._expected_cancelled_tasks = set()
+    adapter._fatal_error_code = None
+    adapter._fatal_error_message = None
+    adapter._fatal_error_retryable = True
+    adapter._fatal_error_handler = None
+    adapter._running = True
+    adapter._busy_text_mode = "queue"
+    adapter._busy_text_debounce_seconds = 0.1
+    adapter._busy_text_hard_cap_seconds = 1.0
+    adapter._text_debounce = {}
+    adapter._auto_tts_default = False
+    adapter._auto_tts_enabled_chats = set()
+    adapter._auto_tts_disabled_chats = set()
+    adapter._typing_paused = set()
+    return adapter
+
+
+@pytest.mark.asyncio
+async def test_secondary_profile_source_profile_stamped_before_busy_check():
+    """event.source.profile must be set synchronously -- before the busy
+    check runs -- not deferred to inside self._message_handler."""
+    adapter = _make_adapter("coder")
+    event = _make_event("hello")
+    assert event.source.profile is None  # unstamped, as a real inbound event is
+
+    await adapter.handle_message(event)
+
+    assert event.source.profile == "coder"
+
+
+@pytest.mark.asyncio
+async def test_secondary_profile_busy_session_keys_are_profile_scoped():
+    """A secondary profile's busy-session state must live under its own
+    agent:<profile> namespace, not collide with agent:main."""
+    coder_adapter = _make_adapter("coder")
+    coder_adapter._busy_text_mode = ""  # direct-merge, no debounce (see test_active_session_text_merge.py)
+
+    probe_source = _make_event("first").source
+    default_key = build_session_key(probe_source, profile=None)
+    coder_key = build_session_key(probe_source, profile="coder")
+    assert default_key != coder_key  # sanity: the two namespaces must differ
+
+    # Simulate an in-flight turn on the coder profile only.
+    coder_adapter._active_sessions[coder_key] = asyncio.Event()
+
+    # A follow-up arrives on the coder bot while its session is busy.
+    await coder_adapter.handle_message(_make_event("are you there?"))
+
+    # Must be recognized as busy under the coder-scoped key -- not silently
+    # dispatched straight to self._message_handler under the wrong (or no)
+    # namespace, which is what happened before source.profile was stamped
+    # early enough for this check to see it.
+    assert coder_key in coder_adapter._pending_messages
+    assert default_key not in coder_adapter._pending_messages
+    coder_adapter._message_handler.assert_not_called()