diff --git a/docs/profile-routing.md b/docs/profile-routing.md
new file mode 100644
index 000000000000..52c6934b47ab
--- /dev/null
+++ b/docs/profile-routing.md
@@ -0,0 +1,115 @@
+# Profile-Based Routing for Inbound Messages
+
+> **Audience:** Gateway operators and contributors
+> **Source files:** `gateway/profile_routing.py`, `gateway/run.py` (`_profile_name_for_source`), `gateway/platforms/base.py` (`build_source`), `gateway/config.py`
+> **Related:** [Session Lifecycle](session-lifecycle.md), `docs/design/profile-builder.md`
+
+## Overview
+
+By default a single gateway run uses one profile (memory, persona, tools). **Profile-based
+routing** lets one gateway instance serve **multiple isolated profiles**, selecting which
+profile handles an inbound message based on *where the message came from* — the platform,
+server (`guild_id`), channel (`chat_id`), and/or thread (`thread_id`).
+
+This is the inbound counterpart to multiplexing: instead of running N gateways, run one
+gateway and route per-community / per-channel / per-thread to a dedicated profile. Each
+profile keeps fully isolated state (`MEMORY.md`, `USER.md`, `SOUL.md`, sessions, tools).
+
+Routing is **platform-generic**: it works for Discord, Telegram, Feishu, Slack, and every
+adapter — not just Discord.
+
+## Configuring routes
+
+Routes live under `profile_routes` in `config.yaml`. Both the top-level and the nested
+`gateway.profile_routes` forms are accepted (the nested form is what
+`hermes config set gateway.profile_routes ...` writes).
+
+```yaml
+profile_routes:
+  # Route an entire Discord server (guild) to one profile.
+  - name: server-default
+    platform: discord
+    guild_id: "1234567890"
+    profile: server-profile
+
+  # Override a specific channel within that server with a different profile.
+  - name: support-channel
+    platform: discord
+    guild_id: "1234567890"
+    chat_id: "9876543210"
+    profile: support-profile
+
+  # Pin a Telegram group to a profile (Telegram has no guild_id — chat_id only).
+  - name: tg-group
+    platform: telegram
+    chat_id: "-1001234567890"
+    profile: tg-profile
+
+  # Route a single Discord thread.
+  - name: standup-thread
+    platform: discord
+    guild_id: "1234567890"
+    chat_id: "9876543210"
+    thread_id: "1111111111"
+    profile: standup
+```
+
+### Fields
+
+| Field | Required | Description |
+|---|---|---|
+| `name` | yes | Human-readable route identifier (used in logs). |
+| `platform` | yes | Adapter platform: `discord`, `telegram`, `feishu`, `slack`, … |
+| `profile` | yes | Target profile name (must exist under `~/.hermes/profiles/<name>`). |
+| `guild_id` | no | Server/guild (Discord). |
+| `chat_id` | no | Channel/group/DM id. |
+| `thread_id` | no | Thread id within a channel. |
+| `enabled` | no | Default `true`; set `false` to disable a route without removing it. |
+
+## Matching rules
+
+A route matches an inbound source when **every discriminator the route declares is satisfied**
+(conjunctive / AND). A field the route leaves unset is ignored.
+
+- **`platform`** must equal the source platform exactly.
+- **`thread_id`** (if set) must equal the source thread id.
+- **`chat_id`** (if set) must match the source channel **or** its parent — a thread in a
+  channel matches the channel's route (hierarchical match for Discord forums/threads).
+- **`guild_id`** (if set) must equal the source guild.
+
+> A route declaring **both** `guild_id` and `chat_id` requires both to hold. A channel match
+> alone does not satisfy a guild constraint — this is intentional and tested.
+
+When multiple routes match, the **most specific** one wins. Specificity is additive:
+
+| Discriminator | Weight |
+|---|---|
+| `thread_id` | 8 |
+| `chat_id` | 4 |
+| `guild_id` | 2 |
+| (platform only) | 1 |
+
+So a thread route (8) beats a channel route (4) beats a guild route (2) within the same server.
+If no route matches, the message uses the default/active profile.
+
+## How it works at runtime
+
+1. An inbound message arrives at a platform adapter.
+2. `BasePlatformAdapter.build_source` builds the `SessionSource` for the message. Every
+   adapter carries a back-reference to the running `GatewayRunner`
+   (`gateway_runner`, injected in `gateway/run.py`), so it asks the runner to resolve the
+   target profile via `_profile_name_for_source`.
+3. `_profile_name_for_source` runs the configured routes through `match_profile_route` and
+   stamps `source.profile` with the winning route's profile (or leaves it unset).
+4. Downstream, `_resolve_profile_home_for_source` chooses the profile home directory
+   (`source.profile` → active profile → `default`) and the session is scoped per-profile, so
+   each routed community gets isolated memory and conversation state.
+
+Because `gateway_runner` is injected for **all** adapters (declared on `BasePlatformAdapter`),
+every platform goes through this path — not just Discord.
+
+## Migration / coexistence with multiplexing
+
+`profile_routes` is independent of `gateway.multiplex_profiles`. Multiplexing splits the
+gateway across model credentials; profile routing splits conversation state across profiles.
+They compose: you may multiplex credentials while also routing channels to distinct profiles.
diff --git a/gateway/config.py b/gateway/config.py
index 87f1c7014788..ac7e5828a3ec 100644
--- a/gateway/config.py
+++ b/gateway/config.py
@@ -721,6 +721,11 @@ class GatewayConfig:
     # fresh session exactly as if the reset policy had fired.  0 = disabled.
     session_store_max_age_days: int = 90
 
+    # Profile-based routing: route specific guilds/channels/threads to
+    # different profiles. See gateway/profile_routing.py. Each entry is a
+    # dict with: name, platform, profile, and optional guild_id/chat_id/thread_id.
+    profile_routes: list = field(default_factory=list)
+
     def get_connected_platforms(self) -> List[Platform]:
         """Return list of platforms that are enabled and configured."""
         connected = []
@@ -827,6 +832,7 @@ def to_dict(self) -> Dict[str, Any]:
             "unauthorized_dm_behavior": self.unauthorized_dm_behavior,
             "streaming": self.streaming.to_dict(),
             "session_store_max_age_days": self.session_store_max_age_days,
+            "profile_routes": self.profile_routes,
         }
     
     @classmethod
@@ -919,6 +925,10 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
         except (TypeError, ValueError):
             session_store_max_age_days = 90
 
+        # Parse profile routes (validated by gateway.profile_routing)
+        from gateway.profile_routing import parse_profile_routes
+        profile_routes = parse_profile_routes(data.get("profile_routes") or [])
+
         return cls(
             platforms=platforms,
             default_reset_policy=default_policy,
@@ -941,6 +951,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
             unauthorized_dm_behavior=unauthorized_dm_behavior,
             streaming=StreamingConfig.from_dict(data.get("streaming", {})),
             session_store_max_age_days=session_store_max_age_days,
+            profile_routes=profile_routes,
         )
 
     def get_unauthorized_dm_behavior(self, platform: Optional[Platform] = None) -> str:
@@ -1047,11 +1058,27 @@ def load_gateway_config() -> GatewayConfig:
             if "thread_sessions_per_user" in yaml_cfg:
                 gw_data["thread_sessions_per_user"] = yaml_cfg["thread_sessions_per_user"]
 
-            # Multiplexing flag: accept both the top-level key and the nested
-            # gateway.multiplex_profiles form (written by
-            # ``hermes config set gateway.multiplex_profiles true``).
-            if "multiplex_profiles" in yaml_cfg:
-                gw_data["multiplex_profiles"] = yaml_cfg["multiplex_profiles"]
+            # Multiplexing flag: accept either top-level ``multiplex_profiles``
+            # or the nested ``gateway.multiplex_profiles`` form (the latter is
+            # what ``hermes config set gateway.multiplex_profiles true`` writes).
+            _mp = yaml_cfg.get("multiplex_profiles")
+            if _mp is None:
+                _gw_section = yaml_cfg.get("gateway")
+                if isinstance(_gw_section, dict):
+                    _mp = _gw_section.get("multiplex_profiles")
+            if _mp is not None:
+                gw_data["multiplex_profiles"] = _mp
+
+            # Profile-based routing rules: accept either top-level
+            # ``profile_routes`` or the nested ``gateway.profile_routes`` form
+            # (matching the multiplex_profiles parity above).
+            _pr = yaml_cfg.get("profile_routes")
+            if _pr is None:
+                _gw_section = yaml_cfg.get("gateway")
+                if isinstance(_gw_section, dict):
+                    _pr = _gw_section.get("profile_routes")
+            if isinstance(_pr, list):
+                gw_data["profile_routes"] = _pr
 
             gateway_section = yaml_cfg.get("gateway")
             if isinstance(gateway_section, dict):
diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 14a7bc336972..e342daf184d7 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -2345,6 +2345,16 @@ class BasePlatformAdapter(ABC):
     # generic seam; Slack is merely the first consumer).
     supports_inchannel_continuable: bool = False
 
+    # Back-reference to the running ``GatewayRunner``, injected by
+    # ``gateway/run.py`` after the adapter is created. Adapters consume it via
+    # ``getattr(self, "gateway_runner", None)`` for cross-platform delivery and
+    # — critically — for inbound profile routing: ``build_source`` resolves the
+    # target profile through ``runner._profile_name_for_source(...)``. Declaring
+    # it on the base (rather than only on adapters that happen to pre-declare
+    # it) means EVERY platform adapter receives the injection, so profile
+    # routing is platform-generic instead of Discord-only.
+    gateway_runner = None  # type: ignore[assignment]  # set by gateway/run.py
+
     def __init__(self, config: PlatformConfig, platform: Platform):
         self.config = config
         self.platform = platform
@@ -5471,10 +5481,47 @@ def build_source(
         auto_thread_created: bool = False,
         auto_thread_initial_name: Optional[str] = None,
     ) -> SessionSource:
-        """Helper to build a SessionSource for this platform."""
+        """Helper to build a SessionSource for this platform.
+
+        When ``gateway.profile_routes`` is configured, the routing engine
+        resolves the matching profile from guild/chat/thread and stamps it on
+        ``source.profile``. Downstream code (``_resolve_profile_home_for_source``
+        in run.py) reads that field to enter ``_profile_runtime_scope`` for
+        per-profile HERMES_HOME isolation.
+        """
         # Normalize empty topic to None
         if chat_topic is not None and not chat_topic.strip():
             chat_topic = None
+
+        # Resolve profile from configured routes (None when no match / no routes)
+        profile = None
+        runner = getattr(self, "gateway_runner", None)
+        if runner is not None:
+            try:
+                profile = runner._profile_name_for_source(
+                    SessionSource(
+                        platform=self.platform,
+                        chat_id=str(chat_id),
+                        chat_name=chat_name,
+                        chat_type=chat_type,
+                        user_id=str(user_id) if user_id else None,
+                        user_name=user_name,
+                        thread_id=str(thread_id) if thread_id else None,
+                        chat_topic=chat_topic.strip() if chat_topic else None,
+                        user_id_alt=user_id_alt,
+                        chat_id_alt=chat_id_alt,
+                        is_bot=is_bot,
+                        guild_id=str(guild_id) if guild_id else None,
+                        parent_chat_id=str(parent_chat_id) if parent_chat_id else None,
+                        message_id=str(message_id) if message_id else None,
+                    )
+                )
+            except Exception:
+                logger.warning(
+                    "Profile resolution failed for %s/%s, defaulting to active profile",
+                    self.platform, chat_id, exc_info=True,
+                )
+
         return SessionSource(
             platform=self.platform,
             chat_id=str(chat_id),
@@ -5490,6 +5537,7 @@ def build_source(
             guild_id=str(guild_id) if guild_id else None,
             parent_chat_id=str(parent_chat_id) if parent_chat_id else None,
             message_id=str(message_id) if message_id else None,
+            profile=profile,
             role_authorized=role_authorized,
             auto_thread_created=auto_thread_created,
             auto_thread_initial_name=auto_thread_initial_name,
diff --git a/gateway/profile_routing.py b/gateway/profile_routing.py
new file mode 100644
index 000000000000..c0ec0acc2bf9
--- /dev/null
+++ b/gateway/profile_routing.py
@@ -0,0 +1,166 @@
+"""Profile-based routing for the gateway with hierarchical matching.
+
+Allows a single Hermes instance to route specific Discord guilds/channels/threads
+to different profiles — each with their own model, tools, memory, and persona.
+
+Matching priority (most specific first):
+  1. platform + chat_id + thread_id (exact thread)  — specificity 14
+  2. platform + chat_id (channel route)             — specificity 6
+  3. platform + guild_id (guild/server route)       — specificity 2
+  4. No match                                       → default profile
+
+Parent-chain matching:
+For Discord threads and forum posts, ``parent_chat_id`` carries the
+direct parent (the channel for a thread, the forum channel for a post).
+Routes keyed on a channel match both direct messages and messages in
+any thread/post whose parent is that channel.
+
+Configuration (config.yaml):
+
+    gateway:
+      profile_routes:
+        - name: server-default
+          platform: discord
+          guild_id: "YOUR_GUILD_ID"
+          profile: server-profile
+
+        - name: special-channel
+          platform: discord
+          guild_id: "YOUR_GUILD_ID"
+          chat_id: "YOUR_CHANNEL_ID"
+          profile: channel-profile
+
+        - name: thread-route
+          platform: discord
+          chat_id: "YOUR_CHANNEL_ID"
+          thread_id: "YOUR_THREAD_ID"
+          profile: thread-profile
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ProfileRoute:
+    """A single routing rule that maps a platform scope to a profile."""
+
+    name: str
+    platform: str
+    profile: str
+    guild_id: Optional[str] = None
+    chat_id: Optional[str] = None
+    thread_id: Optional[str] = None
+    enabled: bool = True
+
+    @property
+    def specificity(self) -> int:
+        """Higher value = more specific match."""
+        s = 0
+        if self.guild_id:
+            s += 2
+        if self.chat_id:
+            s += 4
+        if self.thread_id:
+            s += 8
+        return s
+
+    def matches(
+        self,
+        platform: str,
+        guild_id: Optional[str] = None,
+        chat_id: Optional[str] = None,
+        thread_id: Optional[str] = None,
+        parent_chat_id: Optional[str] = None,
+    ) -> bool:
+        """Return True if this route matches the given source fields.
+
+        All configured discriminators are matched conjunctively (AND): every
+        discriminator that the route declares must hold. ``chat_id`` supports
+        hierarchical matching for Discord forums/threads:
+        - Direct channel match: chat_id == route.chat_id
+        - Thread in channel: parent_chat_id == route.chat_id
+        A route declaring both ``guild_id`` and ``chat_id`` requires both to
+        match (a chat match alone does not satisfy a guild constraint).
+        """
+        if not self.enabled:
+            return False
+        if self.platform != platform:
+            return False
+        if self.thread_id and self.thread_id != thread_id:
+            return False
+        if self.chat_id and self.chat_id != chat_id and self.chat_id != parent_chat_id:
+            return False
+        if self.guild_id and self.guild_id != guild_id:
+            return False
+        return True
+
+
+def parse_profile_routes(raw: Optional[List[Dict[str, Any]]]) -> List[ProfileRoute]:
+    """Parse profile_routes from config.yaml into ProfileRoute objects.
+
+    Returns routes sorted by specificity (most specific first).
+    """
+    if not raw:
+        return []
+    routes: List[ProfileRoute] = []
+    for entry in raw:
+        if not isinstance(entry, dict):
+            continue
+        name = entry.get("name", "")
+        platform = entry.get("platform", "")
+        profile = entry.get("profile", "")
+        if not platform or not profile:
+            logger.warning(
+                "Skipping profile route %s: missing platform or profile",
+                name,
+            )
+            continue
+        # Validate profile name to prevent path traversal. Lazy import avoids a
+        # circular dependency at module load time.
+        try:
+            from hermes_cli.profiles import (
+                normalize_profile_name,
+                validate_profile_name,
+            )
+            profile = normalize_profile_name(profile)
+            validate_profile_name(profile)
+        except (ValueError, ImportError):
+            logger.warning("Skipping profile route %s: invalid profile name %r", name, profile)
+            continue
+        routes.append(
+            ProfileRoute(
+                name=name,
+                platform=platform,
+                profile=profile,
+                guild_id=entry.get("guild_id"),
+                chat_id=entry.get("chat_id"),
+                thread_id=entry.get("thread_id"),
+                enabled=entry.get("enabled", True),
+            )
+        )
+    # Sort: most specific first so the first match wins.
+    routes.sort(key=lambda r: r.specificity, reverse=True)
+    logger.debug("Loaded %d profile routes (most-specific-first)", len(routes))
+    return routes
+
+
+def match_profile_route(
+    routes: List[ProfileRoute],
+    platform: str,
+    guild_id: Optional[str] = None,
+    chat_id: Optional[str] = None,
+    thread_id: Optional[str] = None,
+    parent_chat_id: Optional[str] = None,
+) -> Optional[ProfileRoute]:
+    """Return the best-matching route, or None for no match."""
+    for route in routes:
+        if route.matches(platform, guild_id=guild_id, chat_id=chat_id, thread_id=thread_id, parent_chat_id=parent_chat_id):
+            return route
+    return None
diff --git a/gateway/run.py b/gateway/run.py
index c038f8c14e87..16d335945642 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -8680,6 +8680,20 @@ def _adapter_credential_fingerprint(adapter: Any) -> Optional[str]:
             if isinstance(val, str) and val.strip():
                 token = val.strip()
                 break
+        # Many adapters (e.g. Discord) store the token on their `config`
+        # sub-object rather than directly on the adapter. Without this lookup
+        # those adapters all return None here, the same-token conflict check
+        # is silently skipped, and every profile's adapter for that platform
+        # starts polling the same bot token — producing a per-message race
+        # for which adapter answers. See test_reads_config_token.
+        if not token:
+            cfg = getattr(adapter, "config", None)
+            if cfg is not None:
+                for attr in ("token", "bot_token"):
+                    val = getattr(cfg, attr, None)
+                    if isinstance(val, str) and val.strip():
+                        token = val.strip()
+                        break
         if not token:
             config = getattr(adapter, "config", None)
             val = getattr(config, "token", None)
@@ -8716,12 +8730,14 @@ def _create_adapter(
             if platform_registry.is_registered(platform.value):
                 adapter = platform_registry.create_adapter(platform.value, config)
                 if adapter is not None:
-                    # Adapters that need a back-reference to the gateway runner
-                    # (e.g. for cross-platform admin alerts) declare a
-                    # ``gateway_runner`` attribute. Inject it after creation so
-                    # plugin adapters don't need a custom factory signature.
-                    if hasattr(adapter, "gateway_runner"):
-                        adapter.gateway_runner = self
+                    # Inject a back-reference to the gateway runner so every
+                    # adapter can (a) deliver cross-platform admin alerts and
+                    # (b) resolve inbound profile routing through
+                    # ``runner._profile_name_for_source``. Unconditional:
+                    # ``BasePlatformAdapter`` declares ``gateway_runner``, so
+                    # this reaches ALL platforms (not just the ones that
+                    # pre-declared it), making profile routing platform-generic.
+                    adapter.gateway_runner = self
                     return adapter
                 # Registered but failed to instantiate — don't silently fall
                 # through to built-ins (there are none for plugin platforms).
@@ -17002,19 +17018,99 @@ async def _run_agent(
                 persist_user_timestamp=persist_user_timestamp,
             )
 
+    def _profile_name_for_source(self, source: SessionSource) -> Optional[str]:
+        """Resolve the profile name for an inbound source via configured routes.
+
+        Returns ``None`` when no routes are configured or no route matches.
+        Callers (``build_source``, ``_resolve_profile_home_for_source``) treat
+        ``None`` as "use the default/active profile". When
+        ``gateway.profile_routes`` is configured, the most specific matching
+        route wins (guild < channel < thread). See :mod:`gateway.profile_routing`
+        for matching rules.
+        """
+        config = getattr(self, "config", None)
+        routes = getattr(config, "profile_routes", None)
+        if not routes:
+            return None
+        from gateway.profile_routing import match_profile_route
+        try:
+            matched = match_profile_route(
+                routes,
+                platform=source.platform.value,
+                guild_id=getattr(source, "guild_id", None),
+                chat_id=source.chat_id,
+                thread_id=getattr(source, "thread_id", None),
+                parent_chat_id=getattr(source, "parent_chat_id", None),
+            )
+        except Exception:
+            logger.warning(
+                "Profile route matching failed for %s/%s, falling back to default",
+                source.platform, source.chat_id, exc_info=True,
+            )
+            return None
+        if matched:
+            return matched.profile
+        logger.info(
+            "No profile route matched: platform=%s chat_id=%s thread_id=%s parent_chat_id=%s",
+            source.platform.value, source.chat_id,
+            getattr(source, "thread_id", None), getattr(source, "parent_chat_id", None),
+        )
+        return None
+
     def _resolve_profile_home_for_source(self, source: SessionSource) -> "Path":
         """Resolve which profile's HERMES_HOME should serve this inbound source.
 
-        Prefers the profile the source was routed to (``source.profile`` — set
-        by the /p/<profile>/ URL prefix or a per-credential adapter), falling
-        back to the active profile (the multiplexer's own home).
+        Resolution order:
+          1. ``source.profile`` — set by /p/<profile>/ URL prefix, per-credential
+             adapter ownership, OR profile_routes matching at ``build_source`` time.
+          2. ``_profile_name_for_source`` — re-run routing here as a defensive
+             fallback for sources that bypass ``build_source``.
+          3. The active profile (the multiplexer's own home).
         """
-        from hermes_cli.profiles import get_active_profile_name, get_profile_dir
+        from hermes_cli.profiles import (
+            get_active_profile_name,
+            get_profile_dir,
+            profile_exists,
+        )
+        from hermes_constants import get_hermes_home
+        
+        # Track whether a profile was explicitly requested (vs. falling back to default)
+        explicit_profile = None
         try:
-            name = (source.profile or "").strip() or get_active_profile_name() or "default"
-            return get_profile_dir(name)
+            name = (source.profile or "").strip()
+            if name:
+                explicit_profile = name  # User explicitly set this profile
+            if not name:
+                name = self._profile_name_for_source(source)
+                if name:
+                    explicit_profile = name  # Routing explicitly set this profile
+            if not name:
+                name = get_active_profile_name() or "default"
+            
+            profile_dir = get_profile_dir(name)
+            # Warn if an explicit profile doesn't exist on disk
+            if explicit_profile and not profile_exists(name):
+                logger.warning(
+                    "Profile %r does not exist for source %s/%s (guild_id=%s), "
+                    "falling back to global HERMES_HOME",
+                    explicit_profile,
+                    source.platform.value,
+                    source.chat_id,
+                    getattr(source, "guild_id", None),
+                )
+                return get_hermes_home()
+            return profile_dir
         except Exception:
-            from hermes_constants import get_hermes_home
+            # Catch normalization errors, path errors, etc.
+            logger.warning(
+                "Failed to resolve profile directory for source %s/%s (guild_id=%s), "
+                "falling back to global HERMES_HOME: %s",
+                source.platform.value,
+                source.chat_id,
+                getattr(source, "guild_id", None),
+                explicit_profile or "(no profile)",
+                exc_info=True,
+            )
             return get_hermes_home()
 
     async def _run_agent_inner(
diff --git a/gateway/session.py b/gateway/session.py
index fea3ba4a3c0d..d260ba0ce69d 100644
--- a/gateway/session.py
+++ b/gateway/session.py
@@ -1992,6 +1992,7 @@ def _get_or_create_session_impl(
                     "chat_id": source.chat_id,
                     "chat_type": source.chat_type,
                     "thread_id": source.thread_id,
+                    "profile_name": source.profile,
                 }
 
         if _needs_save:
@@ -2268,6 +2269,7 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) ->
                 "chat_id": old_entry.origin.chat_id if old_entry.origin else None,
                 "chat_type": old_entry.origin.chat_type if old_entry.origin else None,
                 "thread_id": old_entry.origin.thread_id if old_entry.origin else None,
+                "profile_name": old_entry.origin.profile if old_entry.origin else None,
             }
 
         if self._db and db_end_session_id:
diff --git a/hermes_state.py b/hermes_state.py
index da2e484f7cf1..529f26cc0e24 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -756,6 +756,7 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A
     handoff_state TEXT,
     handoff_platform TEXT,
     handoff_error TEXT,
+    profile_name TEXT,
     compression_failure_cooldown_until REAL,
     compression_failure_error TEXT,
     rewind_count INTEGER NOT NULL DEFAULT 0,
@@ -1704,6 +1705,7 @@ def _insert_session_row(
         thread_id: str = None,
         parent_session_id: str = None,
         cwd: str = None,
+        profile_name: str = None,
     ) -> None:
         """Insert a session row, enriching NULL metadata on conflict.
 
@@ -1727,9 +1729,9 @@ def _do(conn):
             conn.execute(
                 """INSERT INTO sessions (
                    id, source, user_id, session_key, chat_id, chat_type, thread_id,
-                   model, model_config, system_prompt, parent_session_id, cwd, started_at
+                   model, model_config, system_prompt, parent_session_id, cwd, profile_name, started_at
                 )
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET
                        model = COALESCE(sessions.model, excluded.model),
                        model_config = COALESCE(sessions.model_config, excluded.model_config),
@@ -1739,7 +1741,8 @@ def _do(conn):
                        chat_type = COALESCE(sessions.chat_type, excluded.chat_type),
                        thread_id = COALESCE(sessions.thread_id, excluded.thread_id),
                        parent_session_id = COALESCE(sessions.parent_session_id, excluded.parent_session_id),
-                       cwd = COALESCE(sessions.cwd, excluded.cwd)""",
+                       cwd = COALESCE(sessions.cwd, excluded.cwd),
+                       profile_name = COALESCE(sessions.profile_name, excluded.profile_name)""",
                 (
                     session_id,
                     source,
@@ -1753,6 +1756,7 @@ def _do(conn):
                     system_prompt,
                     parent_session_id,
                     cwd,
+                    profile_name,
                     time.time(),
                 ),
             )
diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index b216ef29d8ea..a97aef067710 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -6612,12 +6612,20 @@ async def _handle_message(self, message: DiscordMessage, role_authorized: bool =
     # ------------------------------------------------------------------
 
     def _text_batch_key(self, event: MessageEvent) -> str:
-        """Session-scoped key for text message batching."""
+        """Session-scoped key for text message batching.
+
+        Passes ``event.source.profile`` through so routed messages batch
+        under the same namespace the agent run will use (e.g.
+        ``agent:crypto-trader`` instead of ``agent:main``). Without this,
+        the batch key would always land in ``agent:main`` even when the
+        routed profile differs.
+        """
         from gateway.session import build_session_key
         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,
         )
 
     def _enqueue_text_event(self, event: MessageEvent) -> None:
diff --git a/run_agent.py b/run_agent.py
index fe378f396ae9..d2a7919eb46b 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -597,6 +597,13 @@ def _ensure_db_session(self) -> None:
             return
         source = _session_source_for_agent(self.platform)
         try:
+            try:
+                from hermes_cli.profiles import get_active_profile_name
+                _profile_for_session = get_active_profile_name()
+                if _profile_for_session == "default":
+                    _profile_for_session = None
+            except Exception:
+                _profile_for_session = None
             self._session_db.create_session(
                 session_id=self.session_id,
                 source=source,
@@ -606,6 +613,7 @@ def _ensure_db_session(self) -> None:
                 user_id=None,
                 parent_session_id=self._parent_session_id,
                 cwd=_launch_cwd_for_session(source),
+                profile_name=_profile_for_session,
             )
             self._session_db_created = True
         except Exception as e:
diff --git a/tests/gateway/test_config.py b/tests/gateway/test_config.py
index 46ceeb205c03..d099be2497f3 100644
--- a/tests/gateway/test_config.py
+++ b/tests/gateway/test_config.py
@@ -1376,6 +1376,96 @@ def test_existing_platform_configs_accept_home_channel_env_overrides(self):
             assert (home.chat_id, home.name) == expected, platform.value
 
 
+class TestMultiplexProfilesConfig:
+    """Tests for parsing multiplex_profiles (top-level and nested forms)."""
+
+    def test_multiplex_profiles_top_level(self, tmp_path, monkeypatch):
+        """Top-level multiplex_profiles is honored."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "multiplex_profiles: true\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.multiplex_profiles is True
+
+    def test_multiplex_profiles_nested_under_gateway(self, tmp_path, monkeypatch):
+        """gateway.multiplex_profiles (the form written by `hermes config set
+        gateway.multiplex_profiles true`) must be honored. Regression test for
+        the silent-fallback bug where the loader only forwarded the top-level
+        key, so users who wrote it under gateway: got multiplex_profiles=False
+        with no warning."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "gateway:\n  multiplex_profiles: true\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.multiplex_profiles is True, (
+            "gateway.multiplex_profiles: true was silently ignored — "
+            "loader only forwarded the top-level form"
+        )
+
+    def test_multiplex_profiles_default_false(self, tmp_path, monkeypatch):
+        """Default is False when neither form is present."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text("", encoding="utf-8")
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.multiplex_profiles is False
+
+    def test_multiplex_profiles_top_level_overrides_nested(self, tmp_path, monkeypatch):
+        """When both forms are present, top-level wins (matches profile_routes
+        and other parity bridges in load_gateway_config)."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "multiplex_profiles: true\n"
+            "gateway:\n  multiplex_profiles: false\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.multiplex_profiles is True
+
+    def test_multiplex_profiles_explicit_top_level_false_not_consulting_nested(
+        self, tmp_path, monkeypatch
+    ):
+        """Lock in the `is None` vs `is False` distinction: when top-level is
+        explicitly false, the loader must forward False WITHOUT consulting the
+        nested form (so a stale `gateway.multiplex_profiles: true` cannot
+        silently re-enable multiplexing). Guards against a future regression
+        that flips the check to `not _mp`."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "config.yaml").write_text(
+            "multiplex_profiles: false\n"
+            "gateway:\n  multiplex_profiles: true\n",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        config = load_gateway_config()
+
+        assert config.multiplex_profiles is False, (
+            "Explicit top-level false was overridden by nested true — "
+            "loader must respect top-level precedence when key is present"
+        )
+
+
 class TestMultiplexProfilesEnvOverride:
     """GATEWAY_MULTIPLEX_PROFILES env override — the 3-tier precedence chain.
 
diff --git a/tests/gateway/test_multiplex_adapter_registry.py b/tests/gateway/test_multiplex_adapter_registry.py
index 43b89824cb5c..43f26b0ff3e3 100644
--- a/tests/gateway/test_multiplex_adapter_registry.py
+++ b/tests/gateway/test_multiplex_adapter_registry.py
@@ -45,6 +45,58 @@ class _Config:
         assert "config-token" not in fp
 
 
+    def test_reads_config_token(self):
+        """Adapters like Discord store token on `config`, not on self.
+
+        Without the config-token fallback, every Discord adapter in a
+        multiplexed gateway returns None here and the same-token conflict
+        check is silently skipped — N adapters start polling the same bot
+        token and race on every inbound message.
+        """
+        class _Config:
+            token = "discord-bot-token"
+        class _ConfigBackedAdapter:
+            config = _Config()
+        fp = GatewayRunner._adapter_credential_fingerprint(_ConfigBackedAdapter())
+        assert fp is not None
+        assert "discord-bot-token" not in fp
+        assert len(fp) == 16
+
+    def test_distinct_config_tokens_distinct_fp(self):
+        class _CfgA:
+            token = "tok-A"
+        class _CfgB:
+            token = "tok-B"
+        class _A:
+            config = _CfgA()
+        class _B:
+            config = _CfgB()
+        a = GatewayRunner._adapter_credential_fingerprint(_A())
+        b = GatewayRunner._adapter_credential_fingerprint(_B())
+        assert a is not None and b is not None
+        assert a != b
+
+    def test_direct_token_takes_precedence_over_config(self):
+        """If both `adapter.token` and `adapter.config.token` exist, direct wins."""
+        class _Cfg:
+            token = "from-config"
+        class _Both:
+            token = "from-direct"
+            config = _Cfg()
+        fp = GatewayRunner._adapter_credential_fingerprint(_Both())
+        import hashlib
+        expected = hashlib.sha256(b"hermes-mux:from-direct").hexdigest()[:16]
+        assert fp == expected
+
+    def test_config_without_token_returns_none(self):
+        """config present but no token attribute → None (no false positive)."""
+        class _Cfg:
+            pass
+        class _Adapter:
+            config = _Cfg()
+        assert GatewayRunner._adapter_credential_fingerprint(_Adapter()) is None
+
+
 class TestProfileMessageHandler:
     @pytest.mark.asyncio
     async def test_stamps_profile_on_unstamped_source(self):
diff --git a/tests/gateway/test_profile_resolution.py b/tests/gateway/test_profile_resolution.py
new file mode 100644
index 000000000000..0678dd5c8e8d
--- /dev/null
+++ b/tests/gateway/test_profile_resolution.py
@@ -0,0 +1,421 @@
+"""Tests for GatewayRunner._resolve_profile_home_for_source — profile resolution logic."""
+
+import logging
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+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.platforms.base import BasePlatformAdapter
+
+
+@pytest.fixture
+def mock_runner():
+    """Create a minimal mock GatewayRunner with the methods we need."""
+    runner = MagicMock(spec=GatewayRunner)
+    runner.config = MagicMock(profile_routes=[])
+    # Bind the actual methods to the mock
+    runner._profile_name_for_source = GatewayRunner._profile_name_for_source.__get__(runner)
+    runner._resolve_profile_home_for_source = GatewayRunner._resolve_profile_home_for_source.__get__(runner)
+    return runner
+
+
+@pytest.fixture
+def discord_source():
+    """Create a basic Discord SessionSource for testing."""
+    return SessionSource(
+        platform=MagicMock(value="discord"),
+        chat_id="123456",
+        guild_id="789",
+        thread_id=None,
+        parent_chat_id=None,
+    )
+
+
+@pytest.fixture
+def telegram_source():
+    """Create a basic Telegram SessionSource for testing.
+
+    Telegram (like Slack/Feishu/etc.) has no ``guild_id`` — only ``chat_id``.
+    Used to prove profile routing is platform-generic, not Discord-only.
+    """
+    return SessionSource(
+        platform=MagicMock(value="telegram"),
+        chat_id="-1001234567890",
+        guild_id=None,
+        thread_id=None,
+        parent_chat_id=None,
+    )
+
+
+class TestResolutionOrder:
+    """Tests that profile resolution follows the correct priority order."""
+    
+    def test_source_profile_wins_over_routing(self, mock_runner, discord_source):
+        """source.profile should be used even if routing would match."""
+        discord_source.profile = "from-source"
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                with patch("hermes_cli.profiles.profile_exists", return_value=True):
+                    mock_get_dir.return_value = Path("/hermes/profiles/from-source")
+                    result = mock_runner._resolve_profile_home_for_source(discord_source)
+                    
+                    assert result == Path("/hermes/profiles/from-source")
+                    mock_get_dir.assert_called_once_with("from-source")
+    
+    def test_routing_wins_over_active_profile(self, mock_runner, discord_source):
+        """When source.profile is empty, routing should win over active profile."""
+        discord_source.profile = None
+        
+        # Mock routing to return a profile
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                with patch("hermes_cli.profiles.profile_exists", return_value=True):
+                    mock_get_dir.return_value = Path("/hermes/profiles/routed")
+                    
+                    # Manually set routing to return a profile
+                    mock_runner._profile_name_for_source = MagicMock(return_value="routed")
+                    
+                    result = mock_runner._resolve_profile_home_for_source(discord_source)
+                    
+                    assert result == Path("/hermes/profiles/routed")
+                    mock_get_dir.assert_called_once_with("routed")
+    
+    def test_active_profile_fallback(self, mock_runner, discord_source):
+        """When source.profile and routing both return None, active profile is used."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/active")
+                
+                # No routing match
+                mock_runner._profile_name_for_source = MagicMock(return_value=None)
+                
+                result = mock_runner._resolve_profile_home_for_source(discord_source)
+                
+                assert result == Path("/hermes/profiles/active")
+                mock_get_dir.assert_called_once_with("active")
+    
+    def test_default_fallback_when_no_active(self, mock_runner, discord_source):
+        """When even active profile is None, 'default' is used."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value=None):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes")
+                
+                mock_runner._profile_name_for_source = MagicMock(return_value=None)
+                
+                result = mock_runner._resolve_profile_home_for_source(discord_source)
+                
+                assert result == Path("/hermes")
+                mock_get_dir.assert_called_once_with("default")
+
+
+class TestMissingProfileWarning:
+    """Tests for warning when a profile doesn't exist on disk."""
+    
+    def test_nonexistent_profile_warning(self, mock_runner, discord_source, caplog):
+        """When source.profile points to a nonexistent profile, log a WARNING."""
+        discord_source.profile = "nonexistent"
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/nonexistent")
+                with patch("hermes_cli.profiles.profile_exists", return_value=False):
+                    with patch("hermes_constants.get_hermes_home", return_value=Path("/hermes")):
+                        with caplog.at_level(logging.WARNING):
+                            result = mock_runner._resolve_profile_home_for_source(discord_source)
+                            
+                            # Should fall back to global HERMES_HOME
+                            assert result == Path("/hermes")
+                            
+                            # Should have logged a warning
+                            assert len(caplog.records) == 1
+                            assert caplog.records[0].levelname == "WARNING"
+                            assert "nonexistent" in caplog.records[0].message
+                            assert "does not exist" in caplog.records[0].message
+                            assert "discord" in caplog.records[0].message
+                            assert "123456" in caplog.records[0].message
+    
+    def test_nonexistent_routing_profile_warning(self, mock_runner, discord_source, caplog):
+        """When routing returns a nonexistent profile, log a WARNING."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/routed")
+                with patch("hermes_cli.profiles.profile_exists", return_value=False):
+                    with patch("hermes_constants.get_hermes_home", return_value=Path("/hermes")):
+                        # Routing returns a profile that doesn't exist
+                        mock_runner._profile_name_for_source = MagicMock(return_value="routed")
+                        
+                        with caplog.at_level(logging.WARNING):
+                            result = mock_runner._resolve_profile_home_for_source(discord_source)
+                            
+                            # Should fall back to global HERMES_HOME
+                            assert result == Path("/hermes")
+                            
+                            # Should have logged a warning
+                            assert len(caplog.records) == 1
+                            assert "routed" in caplog.records[0].message
+    
+    def test_empty_source_profile_no_warning(self, mock_runner, discord_source, caplog):
+        """When source.profile is empty, silent fallback to active profile (no warning)."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/active")
+                with patch("hermes_cli.profiles.profile_exists", return_value=True):
+                    with caplog.at_level(logging.WARNING):
+                        mock_runner._profile_name_for_source = MagicMock(return_value=None)
+                        
+                        result = mock_runner._resolve_profile_home_for_source(discord_source)
+                        
+                        # Should use active profile
+                        assert result == Path("/hermes/profiles/active")
+                        
+                        # No warnings (active profile exists)
+                        assert not any(r.levelname == "WARNING" for r in caplog.records)
+    
+    def test_existing_profile_no_warning(self, mock_runner, discord_source, caplog):
+        """When the profile exists, no warning should be logged."""
+        discord_source.profile = "existing"
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/existing")
+                with patch("hermes_cli.profiles.profile_exists", return_value=True):
+                    with caplog.at_level(logging.WARNING):
+                        result = mock_runner._resolve_profile_home_for_source(discord_source)
+                        
+                        assert result == Path("/hermes/profiles/existing")
+                        
+                        # No warnings
+                        assert not any(r.levelname == "WARNING" for r in caplog.records)
+
+
+class TestExceptionHandling:
+    """Tests for exception handling in profile resolution."""
+    
+    def test_get_profile_dir_exception_logs_warning(self, mock_runner, discord_source, caplog):
+        """When get_profile_dir raises an exception, log a WARNING with context."""
+        discord_source.profile = "bad-profile"
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir", side_effect=ValueError("Invalid profile name")):
+                with patch("hermes_constants.get_hermes_home", return_value=Path("/hermes")):
+                    with caplog.at_level(logging.WARNING):
+                        result = mock_runner._resolve_profile_home_for_source(discord_source)
+                        
+                        # Should fall back to global HERMES_HOME
+                        assert result == Path("/hermes")
+                        
+                        # Should have logged a warning with exception info
+                        assert len(caplog.records) == 1
+                        assert caplog.records[0].levelname == "WARNING"
+                        assert "bad-profile" in caplog.records[0].message
+                        assert "Failed to resolve profile directory" in caplog.records[0].message
+    
+    def test_exception_with_no_profile_name(self, mock_runner, discord_source, caplog):
+        """Exception when no profile was set should still log a warning."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value=None):
+            with patch("hermes_cli.profiles.get_profile_dir", side_effect=RuntimeError("Filesystem error")):
+                with patch("hermes_constants.get_hermes_home", return_value=Path("/hermes")):
+                    mock_runner._profile_name_for_source = MagicMock(return_value=None)
+                    
+                    with caplog.at_level(logging.WARNING):
+                        result = mock_runner._resolve_profile_home_for_source(discord_source)
+                        
+                        assert result == Path("/hermes")
+                        
+                        # Warning should mention "(no profile)"
+                        assert "(no profile)" in caplog.records[0].message
+
+
+class TestRoutingConsultation:
+    """Tests that _profile_name_for_source is consulted when source.profile is empty."""
+    
+    def test_routing_consulted_when_source_profile_empty(self, mock_runner, discord_source):
+        """_profile_name_for_source should be called when source.profile is empty."""
+        discord_source.profile = None
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/routed")
+                
+                mock_runner._profile_name_for_source = MagicMock(return_value="routed")
+                
+                mock_runner._resolve_profile_home_for_source(discord_source)
+                
+                # Should have called routing
+                mock_runner._profile_name_for_source.assert_called_once_with(discord_source)
+    
+    def test_routing_not_consulted_when_source_profile_set(self, mock_runner, discord_source):
+        """_profile_name_for_source should NOT be called when source.profile is set."""
+        discord_source.profile = "from-source"
+        
+        with patch("hermes_cli.profiles.get_active_profile_name", return_value="active"):
+            with patch("hermes_cli.profiles.get_profile_dir") as mock_get_dir:
+                mock_get_dir.return_value = Path("/hermes/profiles/from-source")
+                
+                mock_runner._profile_name_for_source = MagicMock(return_value="routed")
+                
+                mock_runner._resolve_profile_home_for_source(discord_source)
+                
+                # Should NOT have called routing
+                mock_runner._profile_name_for_source.assert_not_called()
+
+
+class TestNonDiscordProfileRouting:
+    """Profile routing must be platform-generic, not Discord-only.
+
+    Regression coverage for the ``gateway_runner`` injection gap: previously
+    only Discord's adapter pre-declared ``gateway_runner``, so only Discord
+    ever had ``build_source`` call ``_profile_name_for_source``. Telegram /
+    Feishu / Slack / etc. silently fell through to the default profile. These
+    tests pin the resolution half for a non-Discord platform (Telegram).
+    """
+
+    def test_telegram_route_resolves(self, mock_runner, telegram_source):
+        """A configured Telegram route resolves to its profile via the real
+        ``_profile_name_for_source`` (bound onto the mock runner)."""
+        mock_runner.config.profile_routes = [
+            ProfileRoute(name="tg", platform="telegram", profile="tg-profile",
+                         chat_id="-1001234567890"),
+        ]
+        telegram_source.profile = None
+
+        assert mock_runner._profile_name_for_source(telegram_source) == "tg-profile"
+
+    def test_telegram_no_route_returns_none(self, mock_runner, telegram_source):
+        """With no matching Telegram route, resolution returns None (caller
+        falls back to the default/active profile)."""
+        mock_runner.config.profile_routes = [
+            ProfileRoute(name="dc", platform="discord", profile="dc-profile",
+                         chat_id="123456"),
+        ]
+        telegram_source.profile = None
+
+        assert mock_runner._profile_name_for_source(telegram_source) is None
+
+
+class TestGatewayRunnerInjection:
+    """``BasePlatformAdapter`` declares ``gateway_runner`` so the gateway's
+    unconditional injection reaches every platform adapter — the foundation
+    that makes the routing in TestNonDiscordProfileRouting reachable at runtime.
+    """
+
+    def test_base_adapter_declares_gateway_runner(self):
+        from gateway.platforms.base import BasePlatformAdapter
+
+        # Class-level attribute exists and defaults to None.
+        assert hasattr(BasePlatformAdapter, "gateway_runner")
+        assert BasePlatformAdapter.gateway_runner is None
+
+    def test_subclass_inherits_gateway_runner(self):
+        from gateway.platforms.base import BasePlatformAdapter
+
+        class _ToyAdapter(BasePlatformAdapter):
+            pass
+
+        # No manual declaration — yet the attribute is inherited from the base,
+        # so the gateway's ``adapter.gateway_runner = self`` injection reaches
+        # every adapter, not just the ones that pre-declared it (Discord).
+        assert hasattr(_ToyAdapter, "gateway_runner")
+        assert _ToyAdapter.gateway_runner is None
+
+
+# A concrete adapter we can instantiate without the full platform stack.
+# ``build_source`` only reads ``self.platform`` and ``self.gateway_runner``, so a
+# bare instance with those two attrs exercises the real BasePlatformAdapter
+# method end-to-end. Clearing ``__abstractmethods__`` lets ``__new__`` bypass
+# the ABC instantiation guard without stubbing connect/send/get_chat_info/…
+class _StubAdapter(BasePlatformAdapter):
+    pass
+
+
+_StubAdapter.__abstractmethods__ = frozenset()  # type: ignore[attr-defined]
+
+
+def _stub_adapter(platform: Platform, runner) -> "_StubAdapter":
+    a = _StubAdapter.__new__(_StubAdapter)
+    a.platform = platform
+    a.gateway_runner = runner
+    return a
+
+
+class TestAdapterToSessionKeyIntegration:
+    """Adapter -> ``source.profile`` -> session-key integration coverage.
+
+    The review asked for integration coverage for Discord AND a non-Discord
+    platform. These drive a concrete adapter's real ``build_source``
+    (BasePlatformAdapter) with an injected ``gateway_runner``, assert the
+    matched route's profile is stamped on the source, and that the resulting
+    session key is profile-scoped (``agent:<profile>:...`` rather than the
+    shared ``agent:main:...``). The Telegram case is the bug-#2 regression:
+    pre-fix it never received ``gateway_runner`` and fell through to default.
+    """
+
+    @staticmethod
+    def _routes():
+        return [
+            ProfileRoute(name="dc", platform="discord", profile="coder",
+                         guild_id="111", chat_id="222"),
+            ProfileRoute(name="tg", platform="telegram", profile="ops",
+                         chat_id="-1001234567890"),
+        ]
+
+    def test_discord_adapter_stamps_profile_and_scopes_key(self, mock_runner):
+        mock_runner.config.profile_routes = self._routes()
+        adapter = _stub_adapter(Platform.DISCORD, mock_runner)
+
+        source = adapter.build_source(
+            chat_id="222", chat_type="group", guild_id="111", user_id="u1",
+        )
+        assert source.profile == "coder"
+
+        key = build_session_key(source, profile=source.profile)
+        assert key.startswith("agent:coder:"), key
+        # A default-profile key would land in agent:main — must differ.
+        assert key != build_session_key(source, profile=None)
+
+    def test_telegram_adapter_stamps_profile_and_scopes_key(self, mock_runner):
+        """Non-Discord platform (bug #2). The adapter now receives
+        ``gateway_runner``, so ``build_source`` stamps the profile and the
+        session key is isolated under ``agent:ops:`` instead of ``agent:main:``."""
+        mock_runner.config.profile_routes = self._routes()
+        adapter = _stub_adapter(Platform.TELEGRAM, mock_runner)
+
+        source = adapter.build_source(
+            chat_id="-1001234567890", chat_type="group", user_id="u1",
+        )
+        assert source.profile == "ops"
+
+        key = build_session_key(source, profile=source.profile)
+        assert key.startswith("agent:ops:"), key
+        assert key != build_session_key(source, profile=None)
+
+    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``
+        and the session key is the shared ``agent:main:`` namespace — no
+        per-profile isolation. This is the silent fallback the fix removes for
+        non-Discord platforms."""
+        adapter = _stub_adapter(Platform.TELEGRAM, runner=None)
+
+        source = adapter.build_source(
+            chat_id="-1001234567890", chat_type="group", user_id="u1",
+        )
+        assert source.profile is None
+        key = build_session_key(source, profile=source.profile)
+        assert key.startswith("agent:main:"), key
diff --git a/tests/gateway/test_profile_routing.py b/tests/gateway/test_profile_routing.py
new file mode 100644
index 000000000000..abb0c7bcc8ad
--- /dev/null
+++ b/tests/gateway/test_profile_routing.py
@@ -0,0 +1,261 @@
+"""Tests for gateway/profile_routing.py — profile-based routing."""
+
+import pytest
+from gateway.profile_routing import (
+    ProfileRoute,
+    parse_profile_routes,
+    match_profile_route,
+)
+
+
+class TestProfileRoute:
+    def test_specificity_thread(self):
+        r = ProfileRoute(name="t", platform="discord", profile="p",
+                         guild_id="g", chat_id="c", thread_id="t")
+        assert r.specificity == 14  # 2 + 4 + 8
+
+    def test_specificity_channel(self):
+        r = ProfileRoute(name="c", platform="discord", profile="p",
+                         guild_id="g", chat_id="c")
+        assert r.specificity == 6  # 2 + 4
+
+    def test_specificity_guild(self):
+        r = ProfileRoute(name="g", platform="discord", profile="p",
+                         guild_id="g")
+        assert r.specificity == 2
+
+    def test_specificity_minimal(self):
+        r = ProfileRoute(name="m", platform="telegram", profile="p")
+        assert r.specificity == 0
+
+    def test_frozen(self):
+        r = ProfileRoute(name="x", platform="discord", profile="p")
+        with pytest.raises(AttributeError):
+            r.name = "y"
+
+
+class TestProfileRouteMatching:
+    def test_exact_thread_match(self):
+        r = ProfileRoute(name="t", platform="discord", profile="trader",
+                         guild_id="111", chat_id="222", thread_id="333")
+        assert r.matches("discord", guild_id="111", chat_id="222", thread_id="333")
+        assert not r.matches("discord", guild_id="111", chat_id="222", thread_id="444")
+
+    def test_channel_match(self):
+        r = ProfileRoute(name="c", platform="discord", profile="helper",
+                         chat_id="222")
+        assert r.matches("discord", chat_id="222")
+        assert not r.matches("discord", chat_id="333")
+        assert not r.matches("telegram", chat_id="222")
+
+    def test_guild_match(self):
+        r = ProfileRoute(name="g", platform="discord", profile="server",
+                         guild_id="111")
+        assert r.matches("discord", guild_id="111")
+        assert not r.matches("discord", guild_id="222")
+
+    def test_disabled_route_no_match(self):
+        r = ProfileRoute(name="d", platform="discord", profile="off",
+                         guild_id="111", enabled=False)
+        assert not r.matches("discord", guild_id="111")
+
+    def test_guild_route_matches_any_channel_in_guild(self):
+        r = ProfileRoute(name="g", platform="discord", profile="server",
+                         guild_id="111")
+        assert r.matches("discord", guild_id="111", chat_id="222")
+        assert r.matches("discord", guild_id="111", chat_id="222", thread_id="333")
+
+    def test_extra_fields_ignored(self):
+        r = ProfileRoute(name="g", platform="discord", profile="server",
+                         guild_id="111")
+        assert r.matches("discord", guild_id="111", chat_id="any")
+
+    def test_guild_and_chat_are_conjunctive(self):
+        # A route declaring BOTH guild_id and chat_id requires both to match.
+        # Regression guard: previously chat_id was checked first and returned
+        # True before guild_id was ever consulted.
+        r = ProfileRoute(name="gc", platform="discord", profile="scoped",
+                         guild_id="111", chat_id="222")
+        # Both match (direct channel) -> match
+        assert r.matches("discord", guild_id="111", chat_id="222")
+        # Both match via parent (thread inside the channel) -> match
+        assert r.matches("discord", guild_id="111", chat_id="333", parent_chat_id="222")
+        # chat matches but guild differs -> NO match (the bug this guards)
+        assert not r.matches("discord", guild_id="999", chat_id="222")
+        # guild matches but chat differs -> NO match
+        assert not r.matches("discord", guild_id="111", chat_id="333")
+
+
+class TestParseProfileRoutes:
+    def test_empty(self):
+        assert parse_profile_routes(None) == []
+        assert parse_profile_routes([]) == []
+
+    def test_valid_routes_sorted_by_specificity(self):
+        raw = [
+            {"name": "guild", "platform": "discord", "profile": "p", "guild_id": "1"},
+            {"name": "thread", "platform": "discord", "profile": "p",
+             "guild_id": "1", "chat_id": "2", "thread_id": "3"},
+            {"name": "channel", "platform": "discord", "profile": "p", "chat_id": "2"},
+        ]
+        routes = parse_profile_routes(raw)
+        names = [r.name for r in routes]
+        assert names == ["thread", "channel", "guild"]
+
+    def test_skips_invalid(self):
+        raw = [
+            {"platform": "discord"},
+            {"profile": "p"},
+            "not a dict",
+            {"name": "ok", "platform": "telegram", "profile": "p"},
+        ]
+        routes = parse_profile_routes(raw)
+        assert len(routes) == 1
+        assert routes[0].name == "ok"
+
+    def test_enabled_flag(self):
+        raw = [
+            {"name": "off", "platform": "discord", "profile": "p",
+             "guild_id": "1", "enabled": False},
+            {"name": "on", "platform": "discord", "profile": "p", "guild_id": "1"},
+        ]
+        routes = parse_profile_routes(raw)
+        assert not routes[0].enabled
+        assert routes[1].enabled
+
+
+class TestMatchProfileRoute:
+    def test_no_routes(self):
+        assert match_profile_route([], "discord") is None
+
+    def test_returns_first_match(self):
+        routes = [
+            ProfileRoute(name="thread", platform="discord", profile="trader",
+                         guild_id="1", chat_id="2", thread_id="3"),
+            ProfileRoute(name="channel", platform="discord", profile="helper",
+                         chat_id="2"),
+        ]
+        m = match_profile_route(routes, "discord", guild_id="1", chat_id="2", thread_id="3")
+        assert m is not None
+        assert m.profile == "trader"
+
+    def test_falls_through_to_channel(self):
+        routes = [
+            ProfileRoute(name="thread", platform="discord", profile="trader",
+                         guild_id="1", chat_id="2", thread_id="3"),
+            ProfileRoute(name="channel", platform="discord", profile="helper",
+                         chat_id="2"),
+        ]
+        m = match_profile_route(routes, "discord", guild_id="1", chat_id="2")
+        assert m is not None
+        assert m.profile == "helper"
+
+    def test_no_match_returns_none(self):
+        routes = [
+            ProfileRoute(name="r", platform="telegram", profile="p"),
+        ]
+        assert match_profile_route(routes, "discord") is None
+
+
+class TestSessionKeyIntegration:
+    def test_default_profile_key(self):
+        from gateway.session import build_session_key, SessionSource, Platform
+        src = SessionSource(platform=Platform.DISCORD, chat_id="123",
+                            chat_type="channel", user_id="456")
+        key = build_session_key(src)
+        assert key.startswith("agent:main:")
+
+    def test_custom_profile_key(self):
+        from gateway.session import build_session_key, SessionSource, Platform
+        src = SessionSource(platform=Platform.DISCORD, chat_id="123",
+                            chat_type="channel", user_id="456")
+        key = build_session_key(src, profile="trader")
+        assert key.startswith("agent:trader:")
+        assert key == "agent:trader:discord:channel:123:456"
+
+    def test_isolated_sessions(self):
+        from gateway.session import build_session_key, SessionSource, Platform
+        src = SessionSource(platform=Platform.DISCORD, chat_id="123",
+                            chat_type="channel", user_id="456")
+        key_default = build_session_key(src)
+        key_trader = build_session_key(src, profile="trader")
+        assert key_default != key_trader
+
+    def test_dm_profile_scoped(self):
+        from gateway.session import build_session_key, SessionSource, Platform
+        src = SessionSource(platform=Platform.DISCORD, chat_id="999",
+                            chat_type="dm", user_id="111")
+        key = build_session_key(src, profile="bot2")
+        assert key == "agent:bot2:discord:dm:999"
+
+
+
+class TestParentChatIdMatching:
+    """Thread messages carry thread_id as chat_id; parent_chat_id is the channel."""
+
+    def test_channel_route_matches_via_parent_chat_id(self):
+        r = ProfileRoute(name="ch", platform="discord", profile="trader",
+                         chat_id="222")
+        assert r.matches("discord", chat_id="333", parent_chat_id="222")
+
+    def test_channel_route_no_match_wrong_parent(self):
+        r = ProfileRoute(name="ch", platform="discord", profile="trader",
+                         chat_id="222")
+        assert not r.matches("discord", chat_id="333", parent_chat_id="444")
+
+    def test_match_profile_route_with_parent_chat_id(self):
+        routes = [
+            ProfileRoute(name="ch", platform="discord", profile="trader",
+                         chat_id="222"),
+        ]
+        m = match_profile_route(routes, "discord", chat_id="333", parent_chat_id="222")
+        assert m is not None
+        assert m.profile == "trader"
+
+    def test_thread_id_does_not_match_parent_chat_id(self):
+        """thread_id only matches the actual thread_id, never parent_chat_id.
+        Discord snowflakes are globally unique, so thread_id != channel_id."""
+        r = ProfileRoute(name="th", platform="discord", profile="helper",
+                         thread_id="555")
+        assert r.matches("discord", thread_id="555")
+        assert not r.matches("discord", parent_chat_id="555")
+
+    def test_no_parent_chat_id_still_works(self):
+        r = ProfileRoute(name="ch", platform="discord", profile="trader",
+                         chat_id="222")
+        assert r.matches("discord", chat_id="222")
+
+    def test_guild_route_matches_with_parent_chat_id(self):
+        """Guild routes should match regardless of chat_id or parent_chat_id."""
+        r = ProfileRoute(name="g", platform="discord", profile="server",
+                         guild_id="111")
+        assert r.matches("discord", guild_id="111", chat_id="333", parent_chat_id="444")
+
+
+class TestForumPostMatching:
+    """Test that forum posts match via parent_chat_id (direct parent)."""
+
+    def test_forum_channel_route_matches_forum_post(self):
+        """A route on a forum channel should match comments on posts in that forum.
+        
+        In Discord, forum posts (threads) have parent_chat_id = forum channel ID.
+        No cache is needed — the parent relationship is direct.
+        """
+        r = ProfileRoute(name="forum", platform="discord", profile="forum_profile",
+                         chat_id="forum_channel_123")
+        # A comment on a forum post: chat_id=post_thread_id, parent_chat_id=forum_channel_id
+        assert r.matches("discord", chat_id="post_thread_456", parent_chat_id="forum_channel_123")
+
+    def test_forum_post_comment_matches_channel_not_thread_id(self):
+        """Verify that thread_id matching is distinct from parent_chat_id matching."""
+        routes = [
+            ProfileRoute(name="forum", platform="discord", profile="forum_profile",
+                         chat_id="forum_channel_123"),
+            ProfileRoute(name="post", platform="discord", profile="post_profile",
+                         thread_id="post_thread_456"),
+        ]
+        # A comment on the forum post should match the forum channel route, not the thread route
+        m = match_profile_route(routes, "discord", chat_id="post_thread_456", 
+                                 parent_chat_id="forum_channel_123")
+        assert m is not None
+        assert m.profile == "forum_profile"