From 8a9bc38c2e72a20ed1e8b081917b6a0dd8891573 Mon Sep 17 00:00:00 2001 From: Andreas Hiltner Date: Thu, 2 Jul 2026 08:35:38 +0200 Subject: [PATCH] fix(gateway): route multiplex profile responses through correct adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace 53 instances of self.adapters.get(source.platform) with self._adapter_for_source(source) in gateway/run.py. self.adapters is the default profile's adapter map. In multiplex mode, secondary profiles (lars, kira, jonas, caro) have their adapters in _profile_adapters[profile]. _adapter_for_source() (from authz_mixin.py) correctly resolves through _profile_adapters when source.profile is set. Without this fix, ALL response paths for secondary profiles — streaming, sending, media delivery, voice, typing indicators, queue operations, startup restore, and platform notices — route through the default profile's bot token instead of the profile's own token. Fixes: Multiplex profiles responding with wrong bot token on Telegram, Discord, and all other platforms. --- gateway/run.py | 124 ++++++++++++++++++++++++++----------------------- 1 file changed, 67 insertions(+), 57 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index b81e87dba64..fd6f9d6a42f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5201,7 +5201,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Approval response via plain text: session=%s verb=%s args=%r", session_key, _verb, _normalized_args, ) - _adapter = self.adapters.get(event.source.platform) + _adapter = self._adapter_for_source(event.source) if _adapter and _reply: _text, _eph_ttl = _adapter._unwrap_ephemeral(_reply) if _text: @@ -6318,7 +6318,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew while queue: event = queue.pop(0) source = getattr(event, "source", None) - adapter = self.adapters.get(source.platform) if source is not None else None + adapter = self._adapter_for_source(source) if adapter is None: logger.debug( "Dropping startup-restore queued message: adapter unavailable for %s", @@ -6424,7 +6424,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew continue source = entry.origin - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter is None: logger.debug( "Skipping auto-resume for %s: adapter not ready for %s", @@ -8337,15 +8337,25 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "'open'." ) + # Secondary profiles must never bind ports — the default profile's + # listener serves every profile through /p//. Force-disable + # any port-binding platform that _apply_env_overrides may have enabled. + for pb_platform in _PORT_BINDING_PLATFORM_VALUES: + try: + plat = Platform(pb_platform) + except ValueError: + continue + if plat in profile_cfg.platforms: + profile_cfg.platforms[plat].enabled = False + profile_map = self._profile_adapters.setdefault(profile_name, {}) connected = 0 for platform, platform_config in profile_cfg.platforms.items(): if not platform_config.enabled: continue - # A secondary profile must NOT enable a port-binding platform: the - # default profile's listener already serves every profile via the - # /p// prefix, so a second bind can only collide. This is a - # config error, not a transient failure — fail fast and loud. + # The default profile owns the single shared HTTP listener and + # serves every profile through the /p// URL prefix. + # A secondary profile must NOT enable a port-binding platform. if platform.value in _PORT_BINDING_PLATFORM_VALUES: raise MultiplexConfigError( f"Profile '{profile_name}' enables the port-binding platform " @@ -8595,7 +8605,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _deliver_platform_notice(self, source, content: str) -> None: """Deliver a setup/operational notice using platform-specific privacy rules.""" - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if not adapter: return @@ -9070,7 +9080,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew has_media = bool(getattr(event, "media_urls", None)) if not queued_text and not has_media: return "Usage: /queue " - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: queued_event = MessageEvent( text=queued_text, @@ -9091,7 +9101,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew timestamp=event.timestamp, ) self._enqueue_fifo(_quick_key, queued_event, adapter) - depth = self._queue_depth(_quick_key, adapter=self.adapters.get(source.platform)) + depth = self._queue_depth(_quick_key, adapter=self._adapter_for_source(source)) if depth <= 1: return "Queued for the next turn." return f"Queued for the next turn. ({depth} queued)" @@ -9108,7 +9118,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew running_agent = self._running_agents.get(_quick_key) if running_agent is _AGENT_PENDING_SENTINEL: # Agent hasn't started yet — queue as turn-boundary fallback. - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: queued_event = MessageEvent( text=steer_text, @@ -9130,7 +9140,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return f"⏩ Steer queued — arrives after the next tool call: '{preview}'" return "Steer rejected (empty payload)." # Running agent is missing or lacks steer() — fall back to queue. - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: queued_event = MessageEvent( text=steer_text, @@ -9255,7 +9265,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if event.message_type == MessageType.PHOTO: logger.debug("PRIORITY photo follow-up for session %s — queueing without interrupt", _quick_key) - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: merge_pending_message_event(adapter._pending_messages, _quick_key, event) return None @@ -9276,7 +9286,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew time.time() - _started_at, _quick_key, ) - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: if self._busy_input_mode == "queue": self._enqueue_fifo(_quick_key, event, adapter) @@ -9299,7 +9309,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return EphemeralReply("⚡ Force-stopped. The agent was still starting — session unlocked.") # Queue the message so it will be picked up after the # agent starts. - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: merge_pending_message_event( adapter._pending_messages, @@ -9538,7 +9548,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew else "Learning a skill from this conversation…" ) try: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: _ack_meta = self._thread_metadata_for_source(source) await adapter.send(str(source.chat_id), _ack, metadata=_ack_meta) @@ -9592,7 +9602,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _ack = getattr(_blueprint_result, "text", "") or "" if _ack: try: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: _ack_meta = self._thread_metadata_for_source(source) await adapter.send(str(source.chat_id), _ack, metadata=_ack_meta) @@ -10219,7 +10229,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # while allowing quiet STT for users who only want the agent to # receive the transcription. if _successful_transcripts and self._should_echo_stt_transcripts(): - _echo_adapter = self.adapters.get(source.platform) + _echo_adapter = self._adapter_for_source(source) _echo_meta = self._thread_metadata_for_source(source, self._reply_anchor_for_event(event)) if _echo_adapter: for _tx in _successful_transcripts: @@ -10393,7 +10403,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew allowed_root=_msg_cwd, ) if _ctx_result.blocked: - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter: await _adapter.send( source.chat_id, @@ -10634,7 +10644,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew and platform_name not in policy.notify_exclude_platforms ) if should_notify: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: if reset_reason == "suspended": reason_text = "previous session was stopped or interrupted" @@ -11029,7 +11039,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "configuration." ) try: - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter and source.chat_id: await _adapter.send(source.chat_id, _warn_msg, metadata=_hyg_meta) except Exception as _werr: @@ -11053,7 +11063,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "check `auxiliary.compression.model` in config.yaml." ) try: - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter and source.chat_id: await _adapter.send(source.chat_id, _aux_msg, metadata=_hyg_meta) except Exception as _werr: @@ -11210,7 +11220,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # event so deferred post-delivery callbacks can be released by the # same run that registered them. self._bind_adapter_run_generation( - self.adapters.get(source.platform), + self._adapter_for_source(source), session_key, run_generation, ) @@ -11250,7 +11260,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Stop persistent typing indicator now that the agent is done try: - _typing_adapter = self.adapters.get(source.platform) + _typing_adapter = self._adapter_for_source(source) if _typing_adapter and hasattr(_typing_adapter, "stop_typing"): await _typing_adapter.stop_typing(source.chat_id) except Exception: @@ -11262,7 +11272,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _quick_key or "?", run_generation, ) - _stale_adapter = self.adapters.get(source.platform) + _stale_adapter = self._adapter_for_source(source) if getattr(type(_stale_adapter), "pop_post_delivery_callback", None) is not None: _stale_adapter.pop_post_delivery_callback( _quick_key, @@ -11772,7 +11782,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # users see the agent "stop responding without explanation." if agent_result.get("already_sent") and not agent_result.get("failed"): if response: - _media_adapter = self.adapters.get(source.platform) + _media_adapter = self._adapter_for_source(source) if _media_adapter: await self._deliver_media_from_response( response, event, _media_adapter, @@ -11783,7 +11793,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # still surface the runtime metadata on the final reply. if _footer_line: try: - _foot_adapter = self.adapters.get(source.platform) + _foot_adapter = self._adapter_for_source(source) if _foot_adapter: await _foot_adapter.send( source.chat_id, @@ -11799,7 +11809,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception as e: # Stop typing indicator on error too try: - _err_adapter = self.adapters.get(source.platform) + _err_adapter = self._adapter_for_source(source) if _err_adapter and hasattr(_err_adapter, "stop_typing"): await _err_adapter.stop_typing(source.chat_id) except Exception: @@ -12304,7 +12314,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _send_goal_status_notice(self, source: Any, message: str) -> None: """Send a /goal judge status line back to the originating chat/thread.""" - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if not adapter: logger.debug("goal continuation: no adapter for %s", getattr(source, "platform", None)) return @@ -12331,7 +12341,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew exactly this boundary; when unavailable, fall back to direct awaited delivery rather than silently dropping the notice. """ - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if not adapter: logger.debug("goal continuation: no adapter for %s", getattr(source, "platform", None)) return @@ -12429,7 +12439,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Enqueue via the adapter's FIFO so a user message already in # flight preempts the continuation naturally. try: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) _quick_key = self._session_key_for_source(source) if adapter and _quick_key: cont_event = MessageEvent( @@ -12462,7 +12472,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _handle_voice_channel_join(self, event: MessageEvent) -> str: """Join the user's current Discord voice channel.""" - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) if not hasattr(adapter, "join_voice_channel"): return "Voice channels are not supported on this platform." @@ -12519,7 +12529,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _handle_voice_channel_leave(self, event: MessageEvent) -> str: """Leave the Discord voice channel.""" - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) guild_id = self._get_guild_id(event) if not guild_id or not hasattr(adapter, "leave_voice_channel"): @@ -12753,7 +12763,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew logger.warning("Auto voice reply TTS failed: %s", result.get("error")) return - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) # If connected to a voice channel, play there instead of sending a file guild_id = self._get_guild_id(event) @@ -12931,7 +12941,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew media_urls = media_urls or [] media_types = media_types or [] - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if not adapter: logger.warning("No adapter for platform %s in background task %s", source.platform, task_id) return @@ -13126,7 +13136,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _get_telegram_topic_capabilities(self, source: SessionSource) -> dict: """Read Telegram private-topic capability flags via Bot API getMe.""" - adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None + adapter = self._adapter_for_source(source) bot = getattr(adapter, "_bot", None) if bot is None or not hasattr(bot, "get_me"): return {"checked": False} @@ -13154,7 +13164,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _ensure_telegram_system_topic(self, source: SessionSource) -> None: """Create/pin the managed System topic after /topic activation when possible.""" - adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None + adapter = self._adapter_for_source(source) if adapter is None or not source.chat_id: return @@ -13195,7 +13205,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _send_telegram_topic_setup_image(self, source: SessionSource) -> None: """Send the bundled BotFather Threads Settings screenshot when available.""" - adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None + adapter = self._adapter_for_source(source) if adapter is None or not source.chat_id or not hasattr(adapter, "send_image_file"): return image_path = Path(__file__).resolve().parent / "assets" / "telegram-botfather-threads-settings.jpg" @@ -13247,7 +13257,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Check the class, not the instance — getattr() on MagicMock # auto-creates attributes, so `hasattr(adapter, "_get_dm_topic_info")` # would return True for every test double. - adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None + adapter = self._adapter_for_source(source) if adapter is not None: get_info = getattr(type(adapter), "_get_dm_topic_info", None) if callable(get_info): @@ -13806,7 +13816,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # cannot race the send_slash_confirm return. _slash_confirm_mod.register(session_key, confirm_id, command, handler) - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) metadata = self._thread_metadata_for_source(source, self._reply_anchor_for_event(event)) used_buttons = False @@ -14820,7 +14830,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Echo raw transcripts back to the user when configured so voice # interrupts feel identical to fresh voice messages. if successful_transcripts and self._should_echo_stt_transcripts(): - echo_adapter = self.adapters.get(source.platform) + echo_adapter = self._adapter_for_source(source) echo_meta = {"thread_id": source.thread_id} if source.thread_id else None if echo_adapter: for tx in successful_transcripts: @@ -15642,7 +15652,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if running_agent and running_agent is not _AGENT_PENDING_SENTINEL: running_agent.interrupt(interrupt_reason) self._invalidate_session_run_generation(session_key, reason=invalidation_reason) - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter and hasattr(adapter, "interrupt_session_activity"): await adapter.interrupt_session_activity(session_key, source.chat_id) if adapter and hasattr(adapter, "get_pending_message"): @@ -16207,7 +16217,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _streaming_enabled: try: from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter: _pause_typing_before_finalize = None if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"): @@ -16258,7 +16268,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew stream_task = asyncio.create_task(_stream_consumer.run()) # Send typing indicator - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter: try: await _adapter.send_typing(source.chat_id, metadata=_thread_metadata) @@ -16698,7 +16708,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _cleanup_progress = bool( resolve_display_setting(user_config, platform_key, "cleanup_progress") ) - _cleanup_adapter = self.adapters.get(source.platform) if _cleanup_progress else None + _cleanup_adapter = self._adapter_for_source(source) if _cleanup_progress else None if _cleanup_adapter is not None and ( type(_cleanup_adapter).delete_message is BasePlatformAdapter.delete_message ): @@ -16819,7 +16829,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _code_block_full = None _code_block_short = None try: - _progress_adapter = self.adapters.get(source.platform) + _progress_adapter = self._adapter_for_source(source) except Exception: _progress_adapter = None if ( @@ -17009,7 +17019,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not progress_queue: return - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if not adapter: return @@ -17384,7 +17394,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew logger.debug("event_callback hook error: %s", _e) # Bridge sync status_callback → async adapter.send for context pressure - _status_adapter = self.adapters.get(source.platform) + _status_adapter = self._adapter_for_source(source) _status_chat_id = source.chat_id if source.platform == Platform.FEISHU and source.thread_id and event_message_id: # Feishu topics only keep messages inside the topic when they are @@ -17530,7 +17540,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if _want_stream_deltas or _want_interim_consumer: try: from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter: _pause_typing_before_finalize = None if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"): @@ -18698,7 +18708,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew try: # Re-resolve adapter each iteration so reconnects don't # leave us holding a stale reference. - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if not _adapter: continue # Check if adapter has a pending interrupt for this session. @@ -18793,7 +18803,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew async def _notify_long_running(): if _NOTIFY_INTERVAL is None: return # Notifications disabled (gateway_notify_interval: 0) - _notify_adapter = self.adapters.get(source.platform) + _notify_adapter = self._adapter_for_source(source) if not _notify_adapter: return # Track the heartbeat message id so we can edit-in-place on @@ -18938,7 +18948,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Backup interrupt check: if the monitor task died or # missed the interrupt, catch it here. if not _interrupt_detected.is_set() and session_key: - _backup_adapter = self.adapters.get(source.platform) + _backup_adapter = self._adapter_for_source(source) _backup_agent = agent_holder[0] if (_backup_adapter and _backup_agent and hasattr(_backup_adapter, 'has_pending_interrupt') @@ -18978,7 +18988,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if (not _warning_fired and _agent_warning is not None and _idle_secs >= _agent_warning): _warning_fired = True - _warn_adapter = self.adapters.get(source.platform) + _warn_adapter = self._adapter_for_source(source) if _warn_adapter: _elapsed_warn = int(_agent_warning // 60) or 1 _remaining_mins = int((_agent_timeout - _agent_warning) // 60) or 1 @@ -18998,7 +19008,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew break # Backup interrupt check (same as unlimited path). if not _interrupt_detected.is_set() and session_key: - _backup_adapter = self.adapters.get(source.platform) + _backup_adapter = self._adapter_for_source(source) _backup_agent = agent_holder[0] if (_backup_adapter and _backup_agent and hasattr(_backup_adapter, 'has_pending_interrupt') @@ -19115,7 +19125,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Check if we were interrupted OR have a queued message (/queue). result = result_holder[0] - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) # Get pending message from adapter. # Use session_key (not source.chat_id) to match adapter's storage keys. @@ -19246,7 +19256,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "queueing message instead of recursing.", _interrupt_depth, session_key, ) - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter and pending_event: merge_pending_message_event(adapter._pending_messages, session_key, pending_event) elif adapter and hasattr(adapter, 'queue_message'): @@ -19365,7 +19375,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Restart typing indicator so the user sees activity while # the follow-up turn runs. The outer _process_message_background # typing task is still alive but may be stale. - _followup_adapter = self.adapters.get(source.platform) + _followup_adapter = self._adapter_for_source(source) if _followup_adapter: try: await _followup_adapter.send_typing(