fix(gateway): resolve per-profile home channel in multiplex mode

gateway/config.py

@@ -773,8 +773,100 @@ def _is_platform_connected(self, platform: Platform, config: PlatformConfig) ->
 
         return False
 
+    # Env-var names for per-platform home channel lookups. Kept in sync
+    # with the keys written by /sethome (see ``hermes_cli.config.save_env_value``)
+    # and consumed by the adapter's home_target_env_var helper.
+    _HOME_CHANNEL_ENV_KEYS = {
+        Platform.TELEGRAM: "TELEGRAM_HOME_CHANNEL",
+        Platform.DISCORD: "DISCORD_HOME_CHANNEL",
+        Platform.WHATSAPP: "WHATSAPP_HOME_CHANNEL",
+        Platform.WHATSAPP_CLOUD: "WHATSAPP_CLOUD_HOME_CHANNEL",
+        Platform.SLACK: "SLACK_HOME_CHANNEL",
+        Platform.SIGNAL: "SIGNAL_HOME_CHANNEL",
+        Platform.MATTERMOST: "MATTERMOST_HOME_CHANNEL",
+        Platform.MATRIX: "MATRIX_HOME_CHANNEL",
+        Platform.EMAIL: "EMAIL_HOME_CHANNEL",
+        Platform.SMS: "SMS_HOME_CHANNEL",
+        Platform.DINGTALK: "DINGTALK_HOME_CHANNEL",
+        Platform.FEISHU: "FEISHU_HOME_CHANNEL",
+        Platform.WECOM: "WECOM_HOME_CHANNEL",
+        Platform.WEIXIN: "WEIXIN_HOME_CHANNEL",
+        Platform.BLUEBUBBLES: "BLUEBUBBLES_HOME_CHANNEL",
+        Platform.QQBOT: "QQBOT_HOME_CHANNEL",
+        Platform.YUANBAO: "YUANBAO_HOME_CHANNEL",
+    }
+    _HOME_CHANNEL_NAME_ENV_KEYS = {
+        Platform.TELEGRAM: "TELEGRAM_HOME_CHANNEL_NAME",
+        Platform.DISCORD: "DISCORD_HOME_CHANNEL_NAME",
+    }
+    _HOME_CHANNEL_THREAD_ENV_KEYS = {
+        Platform.TELEGRAM: "TELEGRAM_HOME_CHANNEL_THREAD_ID",
+        Platform.DISCORD: "DISCORD_HOME_CHANNEL_THREAD_ID",
+    }
+
     def get_home_channel(self, platform: Platform) -> Optional[HomeChannel]:
-        """Get the home channel for a platform."""
+        """Get the home channel for a platform.
+
+        Resolves live from the active profile's secret scope when one is
+        installed (multiplexed gateway), so each profile can have its own
+        home channel even within a single process. Falls back to the value
+        cached at config-load time when no scope is active (single-profile
+        gateway or pre-startup callers).
+        """
+        # Try a live read first — this is what makes per-profile home
+        # channels work in multiplex mode. ``get_secret`` honors the active
+        # profile's secret scope, falls back to os.environ in single-profile
+        # mode, and fails closed in multiplex without a scope.
+        env_key = self._HOME_CHANNEL_ENV_KEYS.get(platform)
+        if env_key:
+            live_chat_id = None
+            try:
+                from agent.secret_scope import get_secret as _get_secret
+                live_chat_id = _get_secret(env_key) or None
+            except Exception:
+                # secret_scope unavailable (early import, no agent context).
+                # Fall through to the cached value below.
+                pass
+            if not live_chat_id:
+                # Fallback: in single-profile mode, get_secret's os.environ
+                # path should have returned a value. If it didn't, try
+                # os.getenv directly so startup-time callers without a
+                # scope still work. Skip this in multiplex mode to avoid
+                # leaking a cross-profile value.
+                try:
+                    from agent.secret_scope import is_multiplex_active as _is_mp
+                    in_multiplex = _is_mp()
+                except Exception:
+                    in_multiplex = False
+                if not in_multiplex:
+                    import os as _os
+                    live_chat_id = _os.getenv(env_key) or None
+
+            if live_chat_id:
+                name_key = self._HOME_CHANNEL_NAME_ENV_KEYS.get(platform)
+                thread_key = self._HOME_CHANNEL_THREAD_ENV_KEYS.get(platform)
+                name, thread_id = None, None
+                try:
+                    from agent.secret_scope import get_secret as _get_secret
+                    name = _get_secret(name_key) if name_key else None
+                    thread_id = _get_secret(thread_key) if thread_key else None
+                except Exception:
+                    pass
+                if name is None and name_key:
+                    import os as _os
+                    name = _os.getenv(name_key)
+                if thread_id is None and thread_key:
+                    import os as _os
+                    thread_id = _os.getenv(thread_key)
+                return HomeChannel(
+                    platform=platform,
+                    chat_id=str(live_chat_id),
+                    name=name or "Home",
+                    thread_id=str(thread_id) if thread_id else None,
+                )
+
+        # Fall back to the cached value (set at config load or by /sethome's
+        # in-memory sync). Preserves behavior for callers without a scope.
         config = self.platforms.get(platform)
         if config:
             return config.home_channel

gateway/run.py

@@ -2853,6 +2853,7 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             has_active_processes_fn=lambda key: process_registry.has_active_for_session(
                 key, max_active_age=_bg_max_age_seconds,
             ),
+            get_profile_db=getattr(self, "_session_db_for", None),
         )
         # One enforced loop-side boundary for the synchronous SessionStore.
         # Sync helpers keep using ``session_store`` directly; async gateway
@@ -3038,9 +3039,14 @@ def __init__(self, config: Optional[GatewayConfig] = None):
 
         # Initialize session database for session_search tool support
         self._session_db = None
+        self._profile_session_dbs: dict = {}
         try:
             from hermes_state import AsyncSessionDB, SessionDB
             self._session_db = AsyncSessionDB(SessionDB())
+            # In multiplex mode, build per-profile DBs so the agent can
+            # write to the correct profile's state.db instead of the global one.
+            if getattr(self.config, "multiplex_profiles", False):
+                self._build_profile_session_dbs()
         except Exception as e:
             # WARNING (not DEBUG) so the failure appears in errors.log — matches
             # cli.py's handling of the same init path.  Users hitting NFS-mounted
@@ -3123,6 +3129,65 @@ def __init__(self, config: Optional[GatewayConfig] = None):
         # dormant before the drained backlog has a chance to update the clock.
         self._scale_to_zero_cooldown_until: float = 0.0
 
+    def _build_profile_session_dbs(self) -> None:
+        """Build per-profile AsyncSessionDBs for multiplex mode.
+
+        Populates ``self._profile_session_dbs`` with one DB per served
+        profile. The active profile's DB is also stored as
+        ``self._session_db`` so legacy code that reads it still works.
+        Idempotent — safe to call once at startup.
+        """
+        try:
+            from hermes_state import AsyncSessionDB, SessionDB
+            from hermes_cli.profiles import (
+                get_active_profile_name,
+                profiles_to_serve,
+                get_profile_dir,
+            )
+            active = get_active_profile_name() or "default"
+            for profile_name, _ in profiles_to_serve(multiplex=True):
+                if profile_name in self._profile_session_dbs:
+                    continue
+                profile_dir = get_profile_dir(profile_name)
+                if profile_dir is None:
+                    continue
+                db_path = profile_dir / "state.db"
+                try:
+                    self._profile_session_dbs[profile_name] = AsyncSessionDB(
+                        SessionDB(db_path=db_path)
+                    )
+                    logger.info(
+                        "Per-profile session DB ready for '%s' at %s",
+                        profile_name, db_path,
+                    )
+                except Exception as _e:
+                    logger.warning(
+                        "Failed to open session DB for profile '%s' at %s: %s",
+                        profile_name, db_path, _e,
+                    )
+            # Replace self._session_db with the active profile's DB so the
+            # default code path writes to the right place.
+            active_db = self._profile_session_dbs.get(active)
+            if active_db is not None:
+                self._session_db = active_db
+        except Exception as e:
+            logger.warning("Failed to build per-profile session DBs: %s", e)
+
+    def _session_db_for(self, source: Any = None) -> Optional[Any]:
+        """Return the per-profile AsyncSessionDB for the given source.
+
+        Used by SessionStore, slash commands, and the agent creation path
+        to route session writes to the correct profile's state.db. Falls
+        back to ``self._session_db`` (the active profile's DB) when
+        multiplex is off, no profile is set on the source, or the
+        profile has no dedicated DB.
+        """
+        if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
+            return self._session_db
+        if source is None or getattr(source, "profile", None) is None:
+            return self._session_db
+        return self._profile_session_dbs.get(source.profile) or self._session_db
+
 
     def _wire_teams_pipeline_runtime(self) -> None:
         """Bind the Teams meeting pipeline runtime to Graph webhook ingress.
@@ -11448,7 +11513,16 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
         if not history and source.platform and source.platform != Platform.LOCAL and source.platform != Platform.WEBHOOK:
             platform_name = source.platform.value
             env_key = _home_target_env_var(platform_name)
-            if not os.getenv(env_key):
+            # In multiplex mode the per-profile home channel lives in the
+            # active profile's secret scope, not os.environ — read through
+            # get_secret so the notice honors the same per-profile resolution
+            # path as the rest of the gateway.
+            try:
+                from agent.secret_scope import get_secret as _get_secret
+                home_channel_set = bool(_get_secret(env_key))
+            except Exception:
+                home_channel_set = bool(os.getenv(env_key))
+            if not home_channel_set:
                 # Slack dispatches all Hermes commands through a single
                 # parent slash command `/hermes`; bare `/sethome` is not
                 # registered and would fail with "app did not respond".
@@ -13360,7 +13434,7 @@ def run_sync():
                     chat_name=source.chat_name,
                     chat_type=source.chat_type,
                     thread_id=source.thread_id,
-                    session_db=getattr(self._session_db, "_db", self._session_db),
+                    session_db=getattr(self._session_db_for(source) or self._session_db, "_db", self._session_db_for(source) or self._session_db),
                     # Reload from disk — do not reuse the startup snapshot (#60955).
                     fallback_model=self._refresh_fallback_model(),
                 )
@@ -18218,6 +18292,12 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
 
             if agent is None:
                 # Config changed or first message — create fresh agent
+                # Use per-profile session DB in multiplex mode so the agent's
+                # create_session() writes to the correct profile's state.db
+                # instead of leaking to the global default state.db.
+                _agent_session_db = self._session_db_for(source)
+                if _agent_session_db is None:
+                    _agent_session_db = self._session_db
                 agent = AIAgent(
                     model=turn_route["model"],
                     **turn_route["runtime"],
@@ -18247,7 +18327,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                     chat_type=source.chat_type,
                     thread_id=source.thread_id,
                     gateway_session_key=session_key,
-                    session_db=getattr(self._session_db, "_db", self._session_db),
+                    session_db=getattr(_agent_session_db, "_db", _agent_session_db),
                     # Reload from disk — do not reuse the startup snapshot (#60955).
                     fallback_model=self._refresh_fallback_model(),
                 )

gateway/session.py

@@ -992,7 +992,7 @@ class SessionStore:
     """
 
     def __init__(self, sessions_dir: Path, config: GatewayConfig,
-                 has_active_processes_fn=None):
+                 has_active_processes_fn=None, get_profile_db=None):
         self.sessions_dir = sessions_dir
         self.config = config
         self._entries: Dict[str, SessionEntry] = {}
@@ -1015,12 +1015,17 @@ def __init__(self, sessions_dir: Path, config: GatewayConfig,
         )
 
         # Initialize SQLite session database
+        # In multiplex mode, we don't use the global state.db — sessions
+        # go to per-profile state.db files via the callback. So we set
+        # self._db = None to avoid writing to the global DB.
         self._db = None
-        try:
-            from hermes_state import SessionDB
-            self._db = SessionDB()
-        except Exception as e:
-            print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}")
+        self._get_profile_db = get_profile_db
+        if not getattr(config, "multiplex_profiles", False):
+            try:
+                from hermes_state import SessionDB
+                self._db = SessionDB()
+            except Exception as e:
+                print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}")
 
     def _ensure_loaded(self) -> None:
         """Load sessions index from disk if not already loaded."""
@@ -1234,7 +1239,16 @@ def _persist_routing_data(self, data: Dict[str, Any], generation: int) -> None:
             if generation <= getattr(self, "_persisted_routing_generation", 0):
                 return
             db_saved = False
-            _db = getattr(self, "_db", None)
+            _db = None
+            if self._get_profile_db is not None:
+                try:
+                    _profile_db = self._get_profile_db(None)
+                    if _profile_db is not None:
+                        _db = getattr(_profile_db, "_db", _profile_db)
+                except Exception:
+                    pass
+            if _db is None:
+                _db = getattr(self, "_db", None)
             if _db:
                 replacer = getattr(_db, "replace_gateway_routing_entries", None)
                 if callable(replacer):
@@ -1491,9 +1505,16 @@ def _record_gateway_session_peer(
         display_name: Optional[str] = None,
     ) -> None:
         """Persist the routing peer for an existing gateway session row."""
-        if not self._db or not source:
+        if not self._get_profile_db or not source:
+            return
+        try:
+            _profile_db = self._get_profile_db(source)
+            if _profile_db is None:
+                return
+            _db = getattr(_profile_db, "_db", _profile_db)
+        except Exception:
             return
-        recorder = getattr(self._db, "record_gateway_session_peer", None)
+        recorder = getattr(_db, "record_gateway_session_peer", None)
         if not callable(recorder):
             return
         try:
@@ -1997,16 +2018,26 @@ def _get_or_create_session_impl(
         if _needs_save:
             self._save_entries()
 
-        # SQLite operations outside the lock (unchanged).
-        if self._db and db_end_session_id:
+        # SQLite operations outside the lock.
+        # Resolve the per-profile DB if a source is available.
+        _db = self._db
+        if self._get_profile_db is not None:
             try:
-                self._db.end_session(db_end_session_id, "session_reset")
+                _profile_db = self._get_profile_db(source)
+                if _profile_db is not None:
+                    _db = getattr(_profile_db, "_db", _profile_db)
+            except Exception:
+                pass  # fall back to self._db
+
+        if _db and db_end_session_id:
+            try:
+                _db.end_session(db_end_session_id, "session_reset")
             except Exception as e:
                 logger.debug("Session DB operation failed: %s", e)
 
-        if self._db and db_create_kwargs:
+        if _db and db_create_kwargs:
             try:
-                self._db.create_session(**db_create_kwargs)
+                _db.create_session(**db_create_kwargs)
                 self._record_gateway_session_peer(
                     session_id,
                     session_key,