diff --git a/gateway/config.py b/gateway/config.py index 87f1c7014788..70882b40c596 100644 --- a/gateway/config.py +++ b/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 diff --git a/gateway/run.py b/gateway/run.py index ccfa8e92c143..928e08c9429a 100644 --- a/gateway/run.py +++ b/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(), ) diff --git a/gateway/session.py b/gateway/session.py index fea3ba4a3c0d..aedb51befbfc 100644 --- a/gateway/session.py +++ b/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, diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 8d4eee356f95..cddb282ede0b 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -272,7 +272,7 @@ async def _handle_reset_command(self, event: MessageEvent) -> Union[str, Ephemer # Set session title if provided with /new _title_arg = event.get_command_args().strip() _title_note = "" - if _title_arg and self._session_db and new_entry: + if _title_arg and new_entry: from hermes_state import SessionDB try: sanitized = SessionDB.sanitize_title(_title_arg) @@ -281,7 +281,9 @@ async def _handle_reset_command(self, event: MessageEvent) -> Union[str, Ephemer _title_note = t("gateway.reset.title_rejected", error=str(e)) if sanitized: try: - await self._session_db.set_session_title(new_entry.session_id, sanitized) + _sess_db = self._session_db_for(source) + if _sess_db is not None: + await _sess_db.set_session_title(new_entry.session_id, sanitized) header = t("gateway.reset.header_titled", title=sanitized) except ValueError as e: _title_note = t("gateway.reset.title_error_untitled", error=str(e)) @@ -531,13 +533,14 @@ def _int_value(value: Any) -> int: # single source of truth; reading it here keeps /status accurate # without duplicating token writes into two stores. db_total_tokens = 0 - if self._session_db: + _sess_db = self._session_db_for(source) + if _sess_db: try: - title = await self._session_db.get_session_title(session_entry.session_id) + title = await _sess_db.get_session_title(session_entry.session_id) except Exception: title = None try: - row = await self._session_db.get_session(session_entry.session_id) + row = await _sess_db.get_session(session_entry.session_id) if isinstance(row, dict): session_row = row db_total_tokens = ( @@ -808,6 +811,8 @@ async def _resume_target_allowed( """ if allow_override and self._resume_caller_is_admin(source): return True + # Use the per-profile session DB + _sess_db = self._session_db_for(source) # Use the live origin only when it resolves to a real SessionSource; a # store that can't resolve it (or an unexpected lookup error) must not # silently allow/deny — fall through to the deterministic DB scoping. @@ -819,7 +824,7 @@ async def _resume_target_allowed( return self._same_origin_chat(source, origin) # Inactive/persisted-only: best-effort scope by DB row source + user. try: - row = await self._session_db.get_session(target_id) or {} + row = await _sess_db.get_session(target_id) or {} except Exception: return False caller_src = source.platform.value if source.platform else None @@ -2431,7 +2436,12 @@ async def _handle_set_home_command(self, event: MessageEvent) -> str: thread_env_key = _home_thread_env_var(platform_name) thread_id = source.thread_id - # Save to .env so it persists across restarts + # Save to .env so it persists across restarts. ``get_home_channel`` + # now reads live from the active profile's secret scope, so the + # subsequent in-process resolution (cron deliveries, restart + # notifications) sees the new value without needing an in-memory + # cache update. This is critical in multiplex mode where multiple + # profiles share one config instance. try: from hermes_cli.config import save_env_value save_env_value(env_key, str(chat_id)) @@ -2441,20 +2451,6 @@ async def _handle_set_home_command(self, event: MessageEvent) -> str: except Exception as e: return t("gateway.set_home.save_failed", error=e) - # Keep the running gateway config in sync too. The pre-restart - # notification path reads self.config before the process reloads env. - if source.platform: - platform_config = self.config.platforms.setdefault( - source.platform, - PlatformConfig(enabled=True), - ) - platform_config.home_channel = HomeChannel( - platform=source.platform, - chat_id=str(chat_id), - name=chat_name, - thread_id=str(thread_id) if thread_id else None, - ) - return t("gateway.set_home.success", name=chat_name, chat_id=chat_id) async def _handle_voice_command(self, event: MessageEvent) -> str: @@ -3211,7 +3207,7 @@ async def _handle_compress_command(self, event: MessageEvent) -> str: skip_memory=True, enabled_toolsets=["memory"], session_id=session_entry.session_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), ) try: tmp_agent._print_fn = lambda *a, **kw: None @@ -3370,7 +3366,8 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st source = event.source if source.platform != Platform.TELEGRAM or source.chat_type != "dm": return t("gateway.topic.not_telegram_dm") - if not self._session_db: + _sess_db = self._session_db_for(source) + if not _sess_db: from hermes_state import format_session_db_unavailable return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) @@ -3414,8 +3411,9 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st await self._send_telegram_topic_setup_image(source) return t("gateway.topic.topics_user_disallowed") + _sess_db = self._session_db_for(source) try: - await self._session_db.enable_telegram_topic_mode( + await _sess_db.enable_telegram_topic_mode( chat_id=str(source.chat_id), user_id=str(source.user_id), has_topics_enabled=capabilities.get("has_topics_enabled"), @@ -3430,7 +3428,7 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st if source.thread_id: try: - binding = await self._session_db.get_telegram_topic_binding( + binding = await _sess_db.get_telegram_topic_binding( chat_id=str(source.chat_id), thread_id=str(source.thread_id), ) @@ -3441,7 +3439,7 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st session_id = str(binding.get("session_id") or "") title = None try: - title = await self._session_db.get_session_title(session_id) + title = await _sess_db.get_session_title(session_id) except Exception: title = None session_label = title or t("gateway.topic.untitled_session") @@ -3460,29 +3458,30 @@ async def _handle_title_command(self, event: MessageEvent) -> str: session_entry = await self.async_session_store.get_or_create_session(source) session_id = session_entry.session_id - if not self._session_db: - from hermes_state import format_session_db_unavailable - return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) - # Ensure session exists in SQLite DB (it may only exist in session_store # if this is the first command in a new session) - existing_title = await self._session_db.get_session_title(session_id) + _sess_db = self._session_db_for(source) + if _sess_db is None: + from hermes_state import format_session_db_unavailable + return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) + existing_title = await _sess_db.get_session_title(session_id) if existing_title is None: # Session doesn't exist in DB yet — create it - try: - await self._session_db.create_session( - session_id=session_id, - source=source.platform.value if source.platform else "unknown", - user_id=source.user_id, - # Persist the messaging origin so a later /resume of this - # titled-but-now-inactive session can prove it belongs to the - # caller's chat/thread (IDOR scoping). - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, - ) - except Exception: - pass # Session might already exist, ignore errors + if _sess_db is not None: + try: + await _sess_db.create_session( + session_id=session_id, + source=source.platform.value if source.platform else "unknown", + user_id=source.user_id, + # Persist the messaging origin so a later /resume of this + # titled-but-now-inactive session can prove it belongs to the + # caller's chat/thread (IDOR scoping). + chat_id=source.chat_id, + chat_type=source.chat_type, + thread_id=source.thread_id, + ) + except Exception: + pass # Session might already exist, ignore errors title_arg = event.get_command_args().strip() if title_arg: @@ -3496,7 +3495,7 @@ async def _handle_title_command(self, event: MessageEvent) -> str: return t("gateway.title.empty_after_clean") # Set the title try: - if await self._session_db.set_session_title(session_id, sanitized): + if await _sess_db.set_session_title(session_id, sanitized): # Propagate the user-chosen title to the visible Telegram # forum topic name too. Auto-generated titles already rename # the topic; without this, /title only updated the DB title @@ -3520,7 +3519,7 @@ async def _handle_title_command(self, event: MessageEvent) -> str: return t("gateway.shared.warn_passthrough", error=e) else: # Show the current title and session ID - title = await self._session_db.get_session_title(session_id) + title = await _sess_db.get_session_title(session_id) if title: return t("gateway.title.current_with_title", session_id=session_id, title=title) else: @@ -3528,11 +3527,12 @@ async def _handle_title_command(self, event: MessageEvent) -> str: async def _handle_resume_command(self, event: MessageEvent) -> str: """Handle /resume command — list or switch to a previous session.""" - if not self._session_db: + source = event.source + _sess_db = self._session_db_for(source) + if not _sess_db: from hermes_state import format_session_db_unavailable return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) - source = event.source session_key = self._session_key_for_source(source) raw_args = event.get_command_args().strip() try: @@ -3555,7 +3555,7 @@ async def _handle_resume_command(self, event: MessageEvent) -> str: async def _list_titled_sessions() -> list[dict]: user_source = source.platform.value if source.platform else None - sessions = await self._session_db.list_sessions_rich(source=user_source, limit=10) + sessions = await _sess_db.list_sessions_rich(source=user_source, limit=10) return [s for s in sessions if s.get("title")][:10] if not name: @@ -3606,17 +3606,17 @@ async def _list_titled_sessions() -> list[dict]: else: # Try direct session ID lookup first (so `/resume <session_id>` # works in the gateway, not just `/resume <title>`). - session = await self._session_db.get_session(name) + session = await _sess_db.get_session(name) if session: target_id = session["id"] else: - target_id = await self._session_db.resolve_session_by_title(name) + target_id = await _sess_db.resolve_session_by_title(name) if not target_id: return t("gateway.resume.not_found", name=name) # Compression creates child continuations that hold the live transcript. # Follow that chain so gateway /resume matches CLI behavior (#15000). try: - target_id = await self._session_db.resolve_resume_session_id(target_id) + target_id = await _sess_db.resolve_resume_session_id(target_id) except Exception as e: logger.debug("Failed to resolve resume continuation for %s: %s", target_id, e) @@ -3682,7 +3682,7 @@ async def _list_titled_sessions() -> list[dict]: self._evict_cached_agent(session_key) # Get the title for confirmation - title = await self._session_db.get_session_title(target_id) or name + title = await _sess_db.get_session_title(target_id) or name # Count messages for context history = await self.async_session_store.load_transcript(target_id) @@ -3704,7 +3704,9 @@ async def _list_titled_sessions() -> list[dict]: async def _handle_sessions_command(self, event: MessageEvent) -> str: """Handle /sessions — list previous sessions for gateway chats.""" - if not self._session_db: + source = event.source + _sess_db = self._session_db_for(source) + if not _sess_db: from hermes_state import format_session_db_unavailable return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) @@ -3714,7 +3716,6 @@ async def _handle_sessions_command(self, event: MessageEvent) -> str: query_session_listing, ) - source = event.source raw_args = event.get_command_args().strip() try: include_all, include_unnamed, target, search_query = ( @@ -3739,7 +3740,7 @@ async def _handle_sessions_command(self, event: MessageEvent) -> str: current_entry = await self.async_session_store.get_or_create_session(source) rows = await asyncio.to_thread( query_session_listing, - getattr(self._session_db, "_db", self._session_db), + getattr(_sess_db, "_db", _sess_db), source=source.platform.value if source.platform else None, current_session_id=current_entry.session_id, include_all_sources=cross_origin, @@ -3777,11 +3778,12 @@ async def _handle_branch_command(self, event: MessageEvent) -> str: """ import uuid as _uuid - if not self._session_db: + source = event.source + _sess_db = self._session_db_for(source) + if not _sess_db: from hermes_state import format_session_db_unavailable return format_session_db_unavailable(prefix=t("gateway.shared.session_db_unavailable_prefix")) - source = event.source session_key = self._session_key_for_source(source) # Load the current session and its transcript @@ -3803,9 +3805,9 @@ async def _handle_branch_command(self, event: MessageEvent) -> str: if branch_name: branch_title = branch_name else: - current_title = await self._session_db.get_session_title(current_entry.session_id) + current_title = await _sess_db.get_session_title(current_entry.session_id) base = current_title or "branch" - branch_title = await self._session_db.get_next_title_in_lineage(base) + branch_title = await _sess_db.get_next_title_in_lineage(base) parent_session_id = current_entry.session_id @@ -3815,7 +3817,7 @@ async def _handle_branch_command(self, event: MessageEvent) -> str: # /sessions even after the parent is reopened and re-ended with a # different end_reason (e.g. tui_shutdown overwriting 'branched'). try: - await self._session_db.create_session( + await _sess_db.create_session( session_id=new_session_id, source=source.platform.value if source.platform else "gateway", model=(self.config.get("model", {}) or {}).get("default") if isinstance(self.config, dict) else None, @@ -3829,7 +3831,7 @@ async def _handle_branch_command(self, event: MessageEvent) -> str: # Copy conversation history to the new session for msg in history: try: - await self._session_db.append_message( + await _sess_db.append_message( session_id=new_session_id, role=msg.get("role", "user"), content=msg.get("content"), @@ -3848,7 +3850,7 @@ async def _handle_branch_command(self, event: MessageEvent) -> str: # Set title try: - await self._session_db.set_session_title(new_session_id, branch_title) + await _sess_db.set_session_title(new_session_id, branch_title) except Exception: pass @@ -3972,7 +3974,8 @@ async def _handle_usage_command(self, event: MessageEvent) -> str: if not provider and getattr(self, "_session_db", None) is not None: try: _entry_for_billing = await self.async_session_store.get_or_create_session(source) - persisted = await self._session_db.get_session(_entry_for_billing.session_id) or {} + _billing_db = self._session_db_for(source) or self._session_db + persisted = await _billing_db.get_session(_entry_for_billing.session_id) or {} except Exception: persisted = {} provider = provider or persisted.get("billing_provider") diff --git a/hermes_state.py b/hermes_state.py index a0b6dfd09ccd..6bf375079e5d 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1636,6 +1636,8 @@ def _do(conn): ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET + user_id = COALESCE(sessions.user_id, excluded.user_id), + source = COALESCE(sessions.source, excluded.source), model = COALESCE(sessions.model, excluded.model), model_config = COALESCE(sessions.model_config, excluded.model_config), system_prompt = COALESCE(sessions.system_prompt, excluded.system_prompt), diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index d66a1d87b76e..64135adf0b01 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -46,12 +46,22 @@ when requester metadata is available (default: true) MATRIX_APPROVAL_TIMEOUT_SECONDS Reaction approval/model-picker timeout (default: 300) + MATRIX_SENDER_PROFILE_MAP JSON object mapping sender MXIDs to Hermes profile names + (e.g. '{"@alice:server":"lexi","@bob:server":"lana"}'). + When a sender matches, their messages are routed to the + specified profile. Can also be set via config.extra + ``sender_profile_map`` (merged with env var). + config.extra.room_profile_map + JSON object mapping room IDs to Hermes profile names. + Room-level mapping takes priority over sender-level. + Example: {"!abc:server":"lexi","!def:server":"family"} """ from __future__ import annotations import asyncio import inspect +import json import logging import mimetypes import os @@ -971,6 +981,40 @@ def __init__(self, config: PlatformConfig): exc, ) + # Per-sender / per-room profile routing + self._sender_profile_map: Dict[str, str] = {} + self._room_profile_map: Dict[str, str] = {} + profile_map_json = os.getenv("MATRIX_SENDER_PROFILE_MAP", "") + if profile_map_json: + try: + self._sender_profile_map = json.loads(profile_map_json) + except json.JSONDecodeError as exc: + logger.warning("Matrix: invalid MATRIX_SENDER_PROFILE_MAP JSON: %s", exc) + # Also check config.extra + sender_map_extra = config.extra.get("sender_profile_map") + if isinstance(sender_map_extra, dict): + self._sender_profile_map.update(sender_map_extra) + room_map_extra = config.extra.get("room_profile_map") + if isinstance(room_map_extra, dict): + self._room_profile_map = room_map_extra + if self._sender_profile_map or self._room_profile_map: + logger.info( + "Matrix: profile routing enabled — senders=%d rooms=%d", + len(self._sender_profile_map), + len(self._room_profile_map), + ) + + def _resolve_profile_for_sender(self, room_id: str, sender: str) -> Optional[str]: + """Resolve Hermes profile name for a given sender MXID and room. + + Priority: room_profile_map > sender_profile_map > None (use default). + """ + if room_id in self._room_profile_map: + return self._room_profile_map[room_id] + if sender in self._sender_profile_map: + return self._sender_profile_map[sender] + return None + def _is_duplicate_event(self, event_id) -> bool: """Return True if this event was already processed. Tracks the ID otherwise.""" if not event_id: @@ -2699,6 +2743,15 @@ async def _resolve_message_context( message_id=event_id, ) + # Per-sender / per-room profile routing + resolved_profile = self._resolve_profile_for_sender(room_id, sender) + if resolved_profile: + source.profile = resolved_profile + logger.debug( + "Matrix: routing %s in %s → profile '%s'", + sender, room_id, resolved_profile, + ) + if thread_id: self._threads.mark(thread_id)