fix(gateway): preserve shared route transport adapter

gateway/authz_mixin.py

@@ -92,6 +92,9 @@ class GatewayAuthorizationMixin:
         """Resolve the live adapter for an inbound ``SessionSource``."""
         if source is None:
             return None
+        transport_adapter = self._registered_transport_adapter(source)
+        if transport_adapter is not None:
+            return transport_adapter
         # ``getattr`` guards test fixtures that build a bare source via
         # SimpleNamespace and omit ``profile`` (see AGENTS.md pitfall #17).
         return self._authorization_adapter(
@@ -99,6 +102,43 @@ class GatewayAuthorizationMixin:
             getattr(source, "profile", None),
         )
 
+    def _registered_transport_adapter(self, source: SessionSource):
+        """Return the registered adapter that created *source*, if retained.
+
+        ``source.profile`` is the runtime/session namespace. A chat-based
+        profile route can therefore differ from the adapter profile when one
+        shared credential serves several routed runtimes. ``build_source``
+        keeps the receiving adapter as in-process provenance so replies and
+        intake-policy checks stay on that transport without weakening the
+        fail-closed fallback for restored or hand-built sources.
+        """
+        adapter_ref = getattr(source, "_transport_adapter_ref", None)
+        adapter = adapter_ref() if callable(adapter_ref) else None
+        platform = getattr(source, "platform", None)
+        if adapter is None or platform is None:
+            return None
+        if adapter is (getattr(self, "adapters", None) or {}).get(platform):
+            return adapter
+        profile_maps = getattr(self, "_profile_adapters", None) or {}
+        for profile_adapters in profile_maps.values():
+            if adapter is profile_adapters.get(platform):
+                return adapter
+        return None
+
+    def _adapter_profile_for_source(self, source: SessionSource) -> Optional[str]:
+        """Resolve the transport-owning profile for adapter policy lookups."""
+        adapter = self._registered_transport_adapter(source)
+        platform = getattr(source, "platform", None)
+        if adapter is not None:
+            if adapter is (getattr(self, "adapters", None) or {}).get(platform):
+                return None
+            for profile, profile_adapters in (
+                getattr(self, "_profile_adapters", None) or {}
+            ).items():
+                if adapter is profile_adapters.get(platform):
+                    return profile
+        return getattr(source, "profile", None)
+
     def _adapter_authorization_is_upstream(
         self,
         platform: Optional[Platform],
@@ -311,6 +351,8 @@ class GatewayAuthorizationMixin:
         if source.platform in {Platform.HOMEASSISTANT, Platform.WEBHOOK}:
             return True
 
+        adapter_profile = self._adapter_profile_for_source(source)
+
         # Relay (and any adapter whose authorization is enforced by a trusted
         # authenticated upstream): the Team Gateway connector authenticates this
         # gateway's WS with a per-instance secret and resolves owner-only author
@@ -340,7 +382,7 @@ class GatewayAuthorizationMixin:
         # tests) — defensive against accidental fail-open.
         if source.delivered_via_upstream_relay is True or self._adapter_authorization_is_upstream(
             source.platform,
-            profile=source.profile,
+            profile=adapter_profile,
         ):
             return True
 
@@ -538,23 +580,23 @@ class GatewayAuthorizationMixin:
             # fail-open.)
             if self._adapter_enforces_own_access_policy(
                 source.platform,
-                profile=source.profile,
+                profile=adapter_profile,
             ):
                 if source.chat_type in {"group", "forum", "channel"}:
                     effective_policy = self._adapter_group_policy(
                         source.platform,
-                        profile=source.profile,
+                        profile=adapter_profile,
                     )
                     if self._adapter_group_has_sender_allowlist(
                         source.platform,
                         source.chat_id,
-                        profile=source.profile,
+                        profile=adapter_profile,
                     ):
                         return True
                 else:
                     effective_policy = self._adapter_dm_policy(
                         source.platform,
-                        profile=source.profile,
+                        profile=adapter_profile,
                     )
                 if effective_policy == "allowlist":
                     return True

gateway/platforms/base.py

@@ -17,6 +17,7 @@ import subprocess
 import sys
 import time
 import uuid
+import weakref
 from abc import ABC, abstractmethod
 from urllib.parse import urlsplit
 
@@ -5788,7 +5789,7 @@ class BasePlatformAdapter(ABC):
                     self.platform, chat_id, exc_info=True,
                 )
 
-        return SessionSource(
+        source = SessionSource(
             platform=self.platform,
             chat_id=str(chat_id),
             chat_name=chat_name,
@@ -5809,6 +5810,11 @@ class BasePlatformAdapter(ABC):
             auto_thread_created=auto_thread_created,
             auto_thread_initial_name=auto_thread_initial_name,
         )
+        # In-process transport provenance is deliberately not serialized by
+        # SessionSource.to_dict(). The live receiving adapter is authoritative
+        # for this turn even when profile_routes selects a different runtime.
+        source._transport_adapter_ref = weakref.ref(self)
+        return source
     
     @abstractmethod
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:

tests/gateway/test_multiplex_profile_authz.py

@@ -127,6 +127,32 @@ def test_adapter_for_source_resolves_secondary_profile_adapter(monkeypatch):
     ) is default_adapter
 
 
+def test_chat_routed_source_keeps_receiving_shared_adapter(monkeypatch):
+    """A runtime-only profile route must not discard the shared transport.
+
+    ``source.profile`` selects the routed runtime/session namespace, but the
+    adapter that built the source still owns outbound delivery and intake
+    policy when that profile has no credential of its own.
+    """
+    runner, default_adapter, _secondary_adapter = _make_multiplex_runner(
+        monkeypatch
+    )
+    runner._profile_adapters["routed"] = {}
+
+    source = SessionSource(
+        platform=Platform.WECOM,
+        user_id="allowed-user",
+        chat_id="dm-chat",
+        user_name="allowed-user",
+        chat_type="dm",
+        profile="routed",
+    )
+    assert runner._adapter_for_source(source) is None
+    source._transport_adapter_ref = lambda: default_adapter
+    assert runner._adapter_for_source(source) is default_adapter
+    assert runner._is_user_authorized(source) is True
+
+
 def test_secondary_allowlist_dm_behavior_ignores_unauthorized(monkeypatch):
     """Unauthorized-DM behavior must read the secondary adapter's dm_policy."""
     runner, _default_adapter, secondary_adapter = _make_multiplex_runner(monkeypatch)
@@ -206,4 +232,4 @@ def test_secondary_open_policy_fails_startup_guard(monkeypatch):
     violation = _own_policy_open_startup_violation(secondary_cfg)
     assert violation is not None
     assert "wecom" in violation
-    assert "open policy" in violation
+    assert "open policy" in violation

tests/gateway/test_profile_resolution.py

@@ -9,7 +9,7 @@ import pytest
 from gateway.session import SessionSource, build_session_key
 from gateway.run import GatewayRunner
 from gateway.profile_routing import ProfileRoute
-from gateway.config import Platform
+from gateway.config import GatewayConfig, Platform
 from gateway.platforms.base import BasePlatformAdapter
 
 
@@ -400,11 +400,29 @@ class TestAdapterToSessionKeyIntegration:
             chat_id="-1001234567890", chat_type="group", user_id="u1",
         )
         assert source.profile == "ops"
+        assert source._transport_adapter_ref() is adapter
 
         key = build_session_key(source, profile=source.profile)
         assert key.startswith("agent:ops:"), key
         assert key != build_session_key(source, profile=None)
 
+    def test_chat_route_keeps_shared_adapter_for_delivery(self):
+        runner = object.__new__(GatewayRunner)
+        runner.config = GatewayConfig(
+            multiplex_profiles=True,
+            profile_routes=self._routes(),
+        )
+        runner._profile_adapters = {"ops": {}}
+        adapter = _stub_adapter(Platform.TELEGRAM, runner)
+        runner.adapters = {Platform.TELEGRAM: adapter}
+
+        source = adapter.build_source(
+            chat_id="-1001234567890", chat_type="group", user_id="u1",
+        )
+
+        assert source.profile == "ops"
+        assert runner._adapter_for_source(source) is adapter
+
     def test_adapter_without_runner_falls_back_to_default_namespace(self, mock_runner):
         """Regression anchor: with no ``gateway_runner`` injected (the pre-fix
         state for non-Discord adapters), ``build_source`` leaves ``profile=None``