diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 7238c2c50952..35fec8e8e203 100644 --- a/gateway/platforms/base.py +++ b/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 diff --git a/gateway/run.py b/gateway/run.py index 0bbb976258cd..f614055653a5 100644 --- a/gateway/run.py +++ b/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) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 09f97598e264..ce806691c484 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/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: diff --git a/tests/gateway/test_multiplex_busy_session_profile_routing.py b/tests/gateway/test_multiplex_busy_session_profile_routing.py new file mode 100644 index 000000000000..820e643fe04b --- /dev/null +++ b/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:`` 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: 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()