fix(gateway): honor per-platform session isolation overrides

gateway/config_loader.py

@@ -279,13 +279,24 @@ def bridge_platform_shared_keys(
                 if isinstance(ov_data, dict)
             }
         enabled_was_explicit = cfg_toplevel and "enabled" in platform_cfg
-        if not bridged and not enabled_was_explicit and not has_channel_overrides:
+        nested_extra = platform_cfg.get("extra") if cfg_toplevel else None
+        has_nested_extra = isinstance(nested_extra, dict)
+        if (
+            not bridged
+            and not enabled_was_explicit
+            and not has_channel_overrides
+            and not has_nested_extra
+        ):
             continue
         plat_data = _dict_slot(platforms_data, plat.value)
         extra = _dict_slot(plat_data, "extra")
         if enabled_was_explicit:
             plat_data["enabled"] = platform_cfg["enabled"]
             extra["_enabled_explicit"] = True
+        # Preserve a platform's own nested ``extra`` values. Top-level bridged
+        # keys are applied afterwards and retain their documented precedence.
+        if has_nested_extra:
+            extra.update(nested_extra)
         extra.update(bridged)
 
 

gateway/run.py

@@ -2168,7 +2168,7 @@ def _load_bridge_config(config_path: Path) -> dict:
     ChannelOverride, Platform, GatewayConfig, PlatformConfig, _getenv, load_gateway_config)
 from gateway.session import (
     AsyncSessionStore, SessionStore, SessionSource, SessionContext, build_session_key,
-    profile_from_session_key_namespace)
+    profile_from_session_key_namespace, resolve_runner_session_isolation)
 # Telegram topic routing (#22773, regression fixed #52060): a
 # ``telegram:<positive_chat_id>:<numeric_thread_id>`` cron target is ambiguous — a forum-style topic in a
 # private chat and a genuine Bot API channel Direct-Messages topic share the same shape and need OPPOSITE
@@ -3917,6 +3917,12 @@ def _warn_if_docker_media_delivery_is_risky(self) -> None:
     exit_reason = property(lambda self: self._exit_reason)
     exit_code = property(lambda self: self._exit_code)
 
+    def _resolve_session_isolation_for_source(
+        self, source: SessionSource
+    ) -> tuple[bool, bool]:
+        """Resolve the store's effective scope, with a bare-runner config fallback."""
+        return resolve_runner_session_isolation(self, source)
+
     def _session_key_for_source(self, source: SessionSource) -> str:
         """Resolve the current session key for a source, honoring gateway config when available."""
         if hasattr(self, "session_store") and self.session_store is not None:
@@ -3944,10 +3950,13 @@ def _session_key_for_source(self, source: SessionSource) -> str:
                     _profile = get_active_profile_name() or "default"
                 except Exception:
                     _profile = None
+        group_per_user, thread_per_user = self._resolve_session_isolation_for_source(source)
         return build_session_key(
-            source, group_sessions_per_user=getattr(config, "group_sessions_per_user", True),
-            thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False),
-            profile=_profile)
+            source,
+            group_sessions_per_user=group_per_user,
+            thread_sessions_per_user=thread_per_user,
+            profile=_profile,
+        )
 
     # Telegram General topic in forum-enabled private chats: clients omit message_thread_id or send "1"; both = root.
     _TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"})

gateway/run_inbound.py

@@ -27,7 +27,7 @@
 )
 from gateway.session import (
     SessionSource, build_session_context, is_shared_multi_user_session,
-    neutralize_untrusted_inline_text,
+    neutralize_untrusted_inline_text, resolve_runner_session_isolation,
 )
 from gateway.turn_lease import TurnLeaseTimeoutError
 from typing import Any, Dict, List, Optional, Tuple
@@ -1408,9 +1408,11 @@ def _restore_pending_one_turn_model_override(self, session_key: str, run_generat
 
     def _prefix_inbound_sender_context(self, event: MessageEvent, source: SessionSource, message_text: str) -> str:
         """Attribute the sender in shared multi-user sessions and prepend history-backfill channel context."""
+        group_per_user, thread_per_user = resolve_runner_session_isolation(self, source)
         _is_shared_multi_user = is_shared_multi_user_session(
-            source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
-            thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False),
+            source,
+            group_sessions_per_user=group_per_user,
+            thread_sessions_per_user=thread_per_user,
         )
         if _is_shared_multi_user and source.user_name:
             # Display names are attacker-influenceable: neutralize newlines/control chars or a

gateway/session.py

@@ -621,6 +621,52 @@ def build_channel_continuity_note(entry: "SessionEntry", source: SessionSource)
     )
 
 
+def effective_session_thread_id(source: SessionSource) -> Optional[str]:
+    """Return the thread identity used by session routing."""
+    if source.thread_id:
+        return source.thread_id
+    if source.platform == Platform.DISCORD and source.chat_type != "dm":
+        return source.prospective_thread_id
+    return None
+
+
+def effective_session_chat_type(source: SessionSource) -> str:
+    """Return the chat type represented by the routed session key."""
+    if effective_session_thread_id(source) and not source.thread_id:
+        return "thread"
+    return source.chat_type
+
+
+def resolve_session_isolation(
+    config: Any, source: SessionSource
+) -> tuple[bool, bool]:
+    """Resolve global session-isolation defaults plus platform overrides."""
+    group_per_user = getattr(config, "group_sessions_per_user", True)
+    thread_per_user = getattr(config, "thread_sessions_per_user", False)
+    platform_cfg = getattr(config, "platforms", {}).get(source.platform)
+    extra = getattr(platform_cfg, "extra", None) if platform_cfg else None
+    if isinstance(extra, dict):
+        group_per_user = extra.get("group_sessions_per_user", group_per_user)
+        thread_per_user = extra.get("thread_sessions_per_user", thread_per_user)
+    return group_per_user, thread_per_user
+
+
+def resolve_runner_session_isolation(
+    runner: Any, source: SessionSource
+) -> tuple[bool, bool]:
+    """Resolve through a runner's real store, with a config fallback for bare callers."""
+    store = getattr(runner, "session_store", None)
+    resolver = getattr(store, "_resolve_session_isolation", None)
+    if callable(resolver):
+        try:
+            resolved = resolver(source)
+            if isinstance(resolved, tuple) and len(resolved) == 2:
+                return bool(resolved[0]), bool(resolved[1])
+        except Exception:
+            pass
+    return resolve_session_isolation(getattr(runner, "config", None), source)
+
+
 def is_shared_multi_user_session(
     source: SessionSource, *, group_sessions_per_user: bool = True,
     thread_sessions_per_user: bool = False,
@@ -629,7 +675,11 @@ def is_shared_multi_user_session(
     isolation rules in :func:`build_session_key`)."""
     if source.chat_type == "dm":
         return False
-    return not (thread_sessions_per_user if source.thread_id else group_sessions_per_user)
+    return not (
+        thread_sessions_per_user
+        if effective_session_thread_id(source)
+        else group_sessions_per_user
+    )
 
 
 def _session_key_namespace(profile: Optional[str]) -> str:
@@ -680,8 +730,8 @@ def build_session_key(
     # Discord auto-thread continuity: key a channel-initiating message on the thread it WILL be
     # delivered into (prospective_thread_id), and normalize the chat_type slot to "thread" so
     # in-thread follow-ups byte-match. A real thread_id always wins. DMs use thread_id only.
-    thread_id = source.thread_id or (None if is_dm else source.prospective_thread_id)
-    chat_type_slot = "thread" if thread_id and not source.thread_id else source.chat_type
+    thread_id = effective_session_thread_id(source)
+    chat_type_slot = effective_session_chat_type(source)
     if is_dm:
         # No chat_id: fall back to the sender id before the bare per-platform sink, or every
         # chat_id-less DM shares one agent.
@@ -1268,9 +1318,11 @@ def build_session_context(
 ) -> SessionContext:
     """Build a full session context (for system prompt injection)."""
     connected = config.get_connected_platforms()
+    group_per_user, thread_per_user = resolve_session_isolation(config, source)
     shared = is_shared_multi_user_session(
-        source, group_sessions_per_user=getattr(config, "group_sessions_per_user", True),
-        thread_sessions_per_user=getattr(config, "thread_sessions_per_user", False),
+        source,
+        group_sessions_per_user=group_per_user,
+        thread_sessions_per_user=thread_per_user,
     )
     context = SessionContext(
         source=source, connected_platforms=connected, shared_multi_user_session=shared,

gateway/session_recovery.py

@@ -73,35 +73,49 @@ def _active_profile_name() -> str:
     def _recovered_row_allowed_for_active_profile(
         self, *, requested_session_key: str, recovered: Dict[str, Any]
     ) -> bool:
-        """Prevent a gateway from reviving another profile's row. Single-profile: the row's
-        namespace must match the ACTIVE profile. Multiplexed: it must match the requested key's
-        namespace (the active profile is meaningless there). Keyless rows stay adoptable.
-
-        Multiplexed: several profiles serve traffic at once, so the active profile is meaningless — the
-        requested key carries the profile the turn was routed to, and the recovered row must sit in the same
-        ``agent:<ns>:`` namespace (#74285). Rows with no key namespace stay adoptable in both modes
-        (legacy/keyless data owned by this store).
+        """Prevent a gateway from reviving another profile's row.
+
+        Single-profile recovery follows the durable owner first, then the row's key namespace.
+        Multiplexed recovery follows the requested key's profile namespace because no one process-active
+        profile owns all routed turns. Rows with neither owner nor key remain adoptable legacy data.
         """
+        multiplex_profiles = getattr(self.config, "multiplex_profiles", False)
+        active_profile = self._active_profile_name()
+        requested_profile = self._profile_from_session_key(requested_session_key)
+
+        durable_profile = str(recovered.get("profile_name") or "").strip()
+        if durable_profile:
+            if multiplex_profiles:
+                return requested_profile is None or durable_profile == requested_profile
+            return durable_profile == active_profile
+
         recovered_key = str(recovered.get("session_key") or "")
         if not recovered_key or recovered_key == requested_session_key:
             return True
         recovered_profile = self._profile_from_session_key(recovered_key)
         if recovered_profile is None:
             return True
-        if getattr(self.config, "multiplex_profiles", False):
-            requested_profile = self._profile_from_session_key(requested_session_key)
+        if multiplex_profiles:
             return requested_profile is None or recovered_profile == requested_profile
-        return recovered_profile == self._active_profile_name()
+        return recovered_profile == active_profile
 
     def _generate_session_key(self, source: SessionSource, key_source: Optional[SessionSource] = None) -> str:
         """Session key for *source* (profile from *source*; key from *key_source* if given)."""
         from gateway.session import build_session_key
+        group_per_user, thread_per_user = self._resolve_session_isolation(
+            key_source if key_source is not None else source
+        )
         return build_session_key(
             key_source if key_source is not None else source,
-            group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
-            thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False),
+            group_sessions_per_user=group_per_user,
+            thread_sessions_per_user=thread_per_user,
             profile=self._resolve_profile_for_key(source))
 
+    def _resolve_session_isolation(self, source: SessionSource) -> tuple[bool, bool]:
+        """Resolve group/thread isolation from platform config, then global defaults."""
+        from gateway.session import resolve_session_isolation
+        return resolve_session_isolation(self.config, source)
+
     def _legacy_slack_session_key(self, source: SessionSource) -> Optional[str]:
         """Pre-workspace Slack key for an explicitly scoped source. Deliberately Slack-only: an
         unscoped Slack session may be claimed by only one workspace (old key cannot tell teams)."""
@@ -169,12 +183,14 @@ def _find_gateway_session_row(
         """Query one durable gateway session row. Scoped Slack lookups disable SessionDB's
         platform/chat/user fallback: that tuple has no workspace id and could revive another team's
         session; the caller performs one explicit exact lookup of the old unscoped key instead."""
+        from gateway.session import effective_session_thread_id
+
         return self._peer_row(
             self._db_for_key(session_key), source=source.platform.value, session_key=session_key,
             user_id=source.user_id,
             chat_id=source.chat_id if allow_peer_fallback else None,
             chat_type=source.chat_type if allow_peer_fallback else None,
-            thread_id=source.thread_id, raise_on_lookup_error=raise_on_lookup_error)
+            thread_id=effective_session_thread_id(source), raise_on_lookup_error=raise_on_lookup_error)
 
     @staticmethod
     def _peer_row(db, *, source: str, session_key: str, raise_on_lookup_error: bool = False,
@@ -333,13 +349,16 @@ def _record_gateway_session_peer(
         db = self._db_for_key(session_key)
         if not db or not source:
             return
+        from gateway.session import effective_session_chat_type, effective_session_thread_id
+
         recorder = getattr(db, "record_gateway_session_peer", None)
         if not callable(recorder):
             return
         from gateway.session_identity import transport_profile_of
         peer = dict(
             source=source.platform.value, user_id=source.user_id, session_key=session_key,
-            chat_id=source.chat_id, chat_type=source.chat_type, thread_id=source.thread_id)
+            chat_id=source.chat_id, chat_type=effective_session_chat_type(source),
+            thread_id=effective_session_thread_id(source))
         try:
             recorder(
                 session_id, **peer, display_name=display_name or source.chat_name,
@@ -416,15 +435,17 @@ def _session_create_kwargs(
         """kwargs for ``SessionDB.create_session``. Identity (origin_json) and lineage
         (parent/_reset_from) land atomically in the INSERT so a crash right after cannot strand the
         row unroutable."""
+        from gateway.session import effective_session_chat_type, effective_session_thread_id
         from gateway.session_identity import transport_profile_of
+
         return {
             "session_id": session_id,
             "source": source_value,
             "user_id": origin.user_id if origin else None,
             "session_key": session_key,
             "chat_id": origin.chat_id if origin else None,
-            "chat_type": origin.chat_type if origin else None,
-            "thread_id": origin.thread_id if origin else None,
+            "chat_type": effective_session_chat_type(origin) if origin else None,
+            "thread_id": effective_session_thread_id(origin) if origin else None,
             "profile_name": origin.profile if origin else None,
             "transport_profile": transport_profile_of(origin),
             "origin_json": _origin_json(origin),

gateway/slash_commands_session.py

@@ -19,7 +19,13 @@
 from gateway.config import Platform
 from gateway.platforms.base import EphemeralReply
 from gateway.platforms.event import MessageEvent, MessageType
-from gateway.session import SessionSource, build_session_key, is_shared_multi_user_session
+from gateway.session import (
+    SessionSource,
+    build_session_key,
+    effective_session_thread_id,
+    is_shared_multi_user_session,
+    resolve_runner_session_isolation,
+)
 from gateway.session_transcript import TranscriptReadError
 from gateway.slash_commands_branch_thread import (
     BRANCH_THREAD_PLATFORMS, branch_dest_source, branch_thread_parent, format_thread_ref, parse_branch_args,
@@ -274,7 +280,9 @@ def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSour
         if origin.platform != current.platform or origin.chat_id != current.chat_id:
             return False
         # thread_id is part of every session key: threads of one chat are DIFFERENT sessions.
-        if _sattr(current, "thread_id") != _sattr(origin, "thread_id"):
+        if str(effective_session_thread_id(current) or "") != str(
+            effective_session_thread_id(origin) or ""
+        ):
             return False
         if _sattr(current, "chat_type").lower() in _DM_CHAT_TYPES:
             # An equal non-empty chat_id IS the DM key. build_session_key falls back to the
@@ -296,9 +304,12 @@ def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSour
     def _is_shared_session_source(self, source: SessionSource) -> bool:
         """Whether *source*'s session key is shared by every participant (not per-user); mirrors
         build_session_key's isolation rules so the guards stay in lock-step with the key."""
+        group_per_user, thread_per_user = resolve_runner_session_isolation(self, source)
         return is_shared_multi_user_session(
-            source, group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
-            thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False))
+            source,
+            group_sessions_per_user=group_per_user,
+            thread_sessions_per_user=thread_per_user,
+        )
 
     def _resume_caller_is_admin(self, source: SessionSource) -> bool:
         """Whether *source* is an EXPLICITLY-configured admin (cross-origin /resume, /sessions).
@@ -323,8 +334,9 @@ def _persisted_row_proves_owner(self, source: SessionSource, row: dict) -> bool:
         if not caller_uid:
             return False
         row_thread = str(row.get("thread_id") or "")
+        caller_thread = str(effective_session_thread_id(source) or "")
         if not (row_src and caller_src and str(row_src) == str(caller_src)
-                and row_thread == _sattr(source, "thread_id")):
+                and row_thread == caller_thread):
             return False  # blank/legacy source cannot prove the platform; other thread = other session
         row_uid = str(row.get("user_id") or "")
         row_chat = str(row.get("chat_id") or "")

tests/gateway/test_config.py

@@ -245,6 +245,31 @@ def test_roundtrip_preserves_unauthorized_dm_behavior(self):
         assert restored.unauthorized_dm_behavior == "ignore"
         assert restored.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair"
 
+    @pytest.mark.parametrize("with_bridged_key", [False, True])
+    def test_top_level_platform_nested_extra_survives_loading(
+        self, tmp_path, monkeypatch, with_bridged_key
+    ):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        policy = "  group_policy: open\n" if with_bridged_key else ""
+        (hermes_home / "config.yaml").write_text(
+            "whatsapp:\n"
+            "  extra:\n"
+            "    group_sessions_per_user: false\n"
+            "    bridge_port: 3000\n"
+            "    group_policy: allowlist\n"
+            f"{policy}",
+            encoding="utf-8",
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        extra = load_gateway_config().platforms[Platform.WHATSAPP].extra
+
+        assert extra["group_sessions_per_user"] is False
+        assert extra["bridge_port"] == 3000
+        if with_bridged_key:
+            assert extra["group_policy"] == "open"
+
     def test_email_defaults_to_ignore_for_unauthorized_dm_behavior(self):
         config = GatewayConfig(
             platforms={Platform.EMAIL: PlatformConfig(enabled=True)},