fix(gateway): resolve authorization routing in multiplexed busy sessions

Route primary-adapter busy callbacks through the same transport authorization scope as normal multiplexed messages, while preserving the routed profile for session state.

Adds a regression test for a Signal primary transport routed to a secondary profile.

(cherry picked from commit 429171ae0dcb78fdd20500b93c415ee13d00091f)

gateway/run_adapters.py

@@ -1039,7 +1039,7 @@ class GatewayAdapterLifecycleMixin:
         adapter.set_message_handler(message_handler or self._primary_message_handler())
         adapter.set_fatal_error_handler(fatal_error_handler or self._handle_adapter_fatal_error)
         adapter.set_session_store(self.session_store)
-        adapter.set_busy_session_handler(busy_session_handler or self._handle_active_session_busy_message)
+        adapter.set_busy_session_handler(busy_session_handler or self._primary_busy_session_handler())
         _set_reaction = getattr(adapter, "set_reaction_handler", None)
         if callable(_set_reaction):
             _set_reaction(self._handle_reaction_event)
@@ -1342,6 +1342,38 @@ class GatewayAdapterLifecycleMixin:
 
         return _handler
 
+    def _make_default_profile_busy_session_handler(self):
+        """Scope primary busy messages like normal routed messages.
+
+        A primary adapter admits a message under its own allowlist, while a
+        multiplex route can run its session under a secondary profile. Busy
+        callbacks bypass the normal handler, so carry both identities here.
+        """
+        from gateway.run import _async_profile_runtime_scope, get_hermes_home
+        default_home = Path(get_hermes_home())
+
+        async def _handler(event, _session_key):
+            source = event.source
+            source._authorization_profile_home = default_home
+            if (
+                not getattr(source, "profile", None)
+                and getattr(source, "profile_route_rejected", False) is not True
+                and not self._stamp_routed_profile(source)
+            ):
+                source.profile_route_rejected = True
+            if getattr(source, "profile_route_rejected", False) is True:
+                return True
+            profile_home = (
+                self._resolve_profile_home_for_source(source)
+                if getattr(source, "profile", None) else default_home
+            )
+            async with _async_profile_runtime_scope(profile_home):
+                return await self._handle_active_session_busy_message(
+                    event, self._session_key_for_source(source)
+                )
+
+        return _handler
+
     def _stamp_routed_profile(self, source) -> bool:
         """Stamp ``source.profile`` from ``profile_routes``; False when the route is rejected."""
         from gateway.profile_routing import ProfileRouteRejected
@@ -1355,6 +1387,13 @@ class GatewayAdapterLifecycleMixin:
         """Return the correctly scoped handler for a primary adapter."""
         return self._make_default_profile_message_handler() if self._multiplex_on() else self._handle_message
 
+    def _primary_busy_session_handler(self):
+        """Return the correctly scoped busy-session handler for a primary adapter."""
+        return (
+            self._make_default_profile_busy_session_handler()
+            if self._multiplex_on() else self._handle_active_session_busy_message
+        )
+
     def _multiplex_on(self) -> bool:
         return bool(getattr(self.config, "multiplex_profiles", False))
 

gateway/run_busy.py

@@ -665,8 +665,9 @@ class GatewayBusySessionMixin:
         # Same authorization gate as the cold path, else unauthorized users in shared threads
         # inject messages into a session they don't own.
         from gateway.run import _AGENT_PENDING_SENTINEL
-        # See #17775.
-        if not self._is_user_authorized(event.source):
+        # See #17775. A primary transport can route a turn into a secondary
+        # profile, so authorize in the stamped transport scope.
+        if not self._is_user_authorized_for_source(event.source):
             logger.warning(
                 "Dropping message from unauthorized user in active session: "
                 "user=%s (%s), platform=%s, session=%s", event.source.user_id, event.source.user_name,

tests/gateway/test_64674_multiplex_primary_token_scope.py

@@ -267,6 +267,66 @@ class TestPrimaryMessageRuntimeScope:
         with pytest.raises(secret_scope.UnscopedSecretError):
             secret_scope.get_secret("DISCORD_BOT_TOKEN")
 
+    @pytest.mark.asyncio
+    async def test_primary_busy_path_authorizes_in_transport_scope(
+        self, tmp_path, monkeypatch
+    ):
+        """Routed busy messages must read the admitting adapter's allowlist.
+
+        Signal on Rémi's host is a primary/default transport that routes turns
+        into Sharik via gateway.profile_routes.  A follow-up can arrive while
+        the Sharik agent is still busy; that busy callback used to skip the
+        default-profile scoping wrapper and check SIGNAL_ALLOWED_USERS inside
+        Sharik's unrelated secret scope.
+        """
+        from agent import secret_scope
+        from gateway import run as run_mod
+        from gateway.run import GatewayRunner
+        from gateway.session import SessionSource
+
+        home = tmp_path / "home"
+        home.mkdir()
+        (home / ".env").write_text(
+            "SIGNAL_ALLOWED_USERS=+15550001111\n"
+            "SIGNAL_ALLOW_ALL_USERS=false\n",
+            encoding="utf-8",
+        )
+        sharik = tmp_path / "profiles" / "sharik"
+        sharik.mkdir(parents=True)
+        (sharik / ".env").write_text("# no Signal allowlist here\n", encoding="utf-8")
+        (sharik / "config.yaml").write_text("{}\n", encoding="utf-8")
+
+        monkeypatch.setattr(run_mod, "get_hermes_home", lambda: home)
+        secret_scope.set_multiplex_active(True)
+
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = GatewayConfig(multiplex_profiles=True)
+        runner._resolve_profile_home_for_source = lambda _source: sharik  # type: ignore[method-assign]
+        runner._session_key_for_source = lambda source: f"agent:{source.profile}:signal:dm:{source.chat_id}"  # type: ignore[method-assign]
+        runner._is_user_authorized_for_source = GatewayRunner._is_user_authorized_for_source.__get__(runner)  # type: ignore[method-assign]
+
+        async def _busy(event, session_key):
+            assert session_key == "agent:sharik:signal:dm:+15550001111"
+            # This is what _handle_active_session_busy_message now calls.
+            return runner._is_user_authorized_for_source(event.source)
+
+        runner._handle_active_session_busy_message = _busy  # type: ignore[method-assign]
+        handler = runner._primary_busy_session_handler()
+        event = SimpleNamespace(
+            source=SessionSource(
+                platform=Platform.SIGNAL,
+                chat_id="+15550001111",
+                chat_type="dm",
+                user_id="+15550001111",
+                user_name="Remi",
+                profile="sharik",
+            )
+        )
+
+        assert await handler(event, "stale-unrouted-key") is True
+        with pytest.raises(secret_scope.UnscopedSecretError):
+            secret_scope.get_secret("SIGNAL_ALLOWED_USERS")
+
 
 class TestReconnectDropsEmptyToken:
     @pytest.mark.asyncio