fix(gateway): namespace telegram topic mode by profile under multiplex

gateway/platforms/base.py

@@ -93,6 +93,12 @@ def _thread_metadata_for_source(source, reply_to_message_id: str | None = None)
         anchor = reply_to_message_id or getattr(source, "message_id", None)
         if anchor is not None:
             metadata["telegram_reply_to_message_id"] = str(anchor)
+    # Routed Hermes profile for shared state.db namespaces (topic bindings
+    # under multiplex / profile_routes). Outbound prune paths must not
+    # assume the transport adapter's static profile stamp.
+    profile = str(getattr(source, "profile", None) or "").strip()
+    if profile:
+        metadata["hermes_profile"] = profile
     return metadata
 
 

gateway/run.py

@@ -6587,6 +6587,18 @@ def _session_key_for_source(self, source: SessionSource) -> str:
             profile=_profile,
         )
 
+    @staticmethod
+    def _telegram_topic_profile_name(source: SessionSource) -> str:
+        """Profile namespace for Telegram topic-mode rows (issue #76423).
+
+        Prefer the profile already stamped on the routed event
+        (``source.profile``). Do **not** fall back to the process-global
+        active profile here — under multiplex that can mis-attribute
+        topic state across bots sharing one ``state.db``.
+        """
+        name = str(getattr(source, "profile", None) or "").strip()
+        return name if name else "default"
+
     def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool:
         """Return whether Telegram DM topic mode is active for this chat."""
         if source.platform != Platform.TELEGRAM or source.chat_type != "dm":
@@ -6600,6 +6612,7 @@ def _telegram_topic_mode_enabled(self, source: SessionSource) -> bool:
             raw = session_db.is_telegram_topic_mode_enabled(
                 chat_id=str(source.chat_id),
                 user_id=str(source.user_id),
+                profile_name=self._telegram_topic_profile_name(source),
             )
         except Exception:
             logger.debug("Failed to read Telegram topic mode state", exc_info=True)
@@ -6636,6 +6649,17 @@ def _is_telegram_topic_lane(self, source: SessionSource) -> bool:
 
     _TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 30.0
 
+    def _telegram_topic_cooldown_key(self, source: SessionSource) -> Optional[str]:
+        """Cooldown key for topic-mode cooldowns: (profile, chat_id).
+
+        Profiles sharing a Telegram private chat_id under multiplex must not
+        suppress each other's lobby reminders / capability hints (#76423).
+        """
+        chat_id = str(source.chat_id or "")
+        if not chat_id:
+            return None
+        return f"{self._telegram_topic_profile_name(source)}:{chat_id}"
+
     def _should_send_telegram_lobby_reminder(self, source: SessionSource) -> bool:
         """Rate-limit root-DM lobby reminders to one message per cooldown window.
 
@@ -6645,15 +6669,15 @@ def _should_send_telegram_lobby_reminder(self, source: SessionSource) -> bool:
         """
         if not hasattr(self, "_telegram_lobby_reminder_ts"):
             self._telegram_lobby_reminder_ts = {}
-        chat_id = str(source.chat_id or "")
-        if not chat_id:
+        key = self._telegram_topic_cooldown_key(source)
+        if not key:
             return True
         import time as _time
         now = _time.monotonic()
-        last = self._telegram_lobby_reminder_ts.get(chat_id, 0.0)
+        last = self._telegram_lobby_reminder_ts.get(key, 0.0)
         if now - last < self._TELEGRAM_LOBBY_REMINDER_COOLDOWN_S:
             return False
-        self._telegram_lobby_reminder_ts[chat_id] = now
+        self._telegram_lobby_reminder_ts[key] = now
         return True
 
     def _telegram_topic_root_lobby_message(self) -> str:
@@ -6701,6 +6725,7 @@ def _record_telegram_topic_binding(
             user_id=str(source.user_id or ""),
             session_key=session_entry.session_key,
             session_id=session_entry.session_id,
+            profile_name=self._telegram_topic_profile_name(source),
         )
 
     def _sync_telegram_topic_binding(
@@ -6768,6 +6793,7 @@ def _recover_telegram_topic_thread_id(
         try:
             bindings = session_db.list_telegram_topic_bindings_for_chat(
                 chat_id=str(source.chat_id),
+                profile_name=self._telegram_topic_profile_name(source),
             )
         except Exception:
             logger.debug("topic-recover: read failed", exc_info=True)
@@ -10981,6 +11007,9 @@ async def start(self) -> bool:
             adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
             adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform))
             adapter._busy_text_mode = self._busy_text_mode
+            # Topic-mode rows are namespaced by profile in shared state.db
+            # (issue #76423). Primary adapters own the process-active profile.
+            adapter._hermes_profile_name = self._active_profile_name()
 
             # Try to connect
             logger.info("Connecting to %s...", platform.value)
@@ -12353,6 +12382,7 @@ async def _platform_reconnect_watcher(self) -> None:
                     adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
                     adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform))
                     adapter._busy_text_mode = self._busy_text_mode
+                    adapter._hermes_profile_name = self._active_profile_name()
 
                     # Reconnect after an outage: preserve the platform's
                     # server-side update queue so messages sent while the bot
@@ -13297,6 +13327,9 @@ def _configure_profile_adapter(
             self._make_adapter_auth_check(platform, profile_name=profile_name)
         )
         adapter._busy_text_mode = self._busy_text_mode
+        # Secondary adapters always carry the profile they serve so prune
+        # paths namespace topic bindings correctly under multiplex (#76423).
+        adapter._hermes_profile_name = profile_name
 
     async def _run_secondary_profile_reconnect(
         self, profile_name: str, platform: Platform
@@ -16194,6 +16227,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                 binding = (await self._session_db.get_telegram_topic_binding(
                     chat_id=str(source.chat_id),
                     thread_id=str(source.thread_id),
+                    profile_name=self._telegram_topic_profile_name(source),
                 )) if self._session_db else None
             except Exception:
                 logger.debug("Failed to read Telegram topic binding", exc_info=True)
@@ -19812,6 +19846,7 @@ async def _rename_telegram_topic_for_session_title(
                 binding = await session_db.get_telegram_topic_binding(
                     chat_id=str(source.chat_id),
                     thread_id=str(source.thread_id),
+                    profile_name=self._telegram_topic_profile_name(source),
                 )
                 if binding and str(binding.get("session_id") or "") != str(session_id):
                     return
@@ -19923,15 +19958,15 @@ def _should_send_telegram_capability_hint(self, source: SessionSource) -> bool:
         """
         if not hasattr(self, "_telegram_capability_hint_ts"):
             self._telegram_capability_hint_ts = {}
-        chat_id = str(source.chat_id or "")
-        if not chat_id:
+        key = self._telegram_topic_cooldown_key(source)
+        if not key:
             return True
         import time as _time
         now = _time.monotonic()
-        last = self._telegram_capability_hint_ts.get(chat_id, 0.0)
+        last = self._telegram_capability_hint_ts.get(key, 0.0)
         if now - last < self._TELEGRAM_CAPABILITY_HINT_COOLDOWN_S:
             return False
-        self._telegram_capability_hint_ts[chat_id] = now
+        self._telegram_capability_hint_ts[key] = now
         return True
 
     def _telegram_topic_help_text(self) -> str:
@@ -19969,22 +20004,28 @@ async def _disable_telegram_topic_mode_for_chat(self, source: SessionSource) ->
             currently_enabled = await self._session_db.is_telegram_topic_mode_enabled(
                 chat_id=chat_id,
                 user_id=str(source.user_id or ""),
+                profile_name=self._telegram_topic_profile_name(source),
             )
         except Exception:
             currently_enabled = False
         if not currently_enabled:
             return "Multi-session topic mode is not currently enabled for this chat."
         try:
-            await self._session_db.disable_telegram_topic_mode(chat_id=chat_id)
+            await self._session_db.disable_telegram_topic_mode(
+                chat_id=chat_id,
+                profile_name=self._telegram_topic_profile_name(source),
+            )
         except Exception as exc:
             logger.exception("Failed to disable Telegram topic mode")
             return f"Failed to disable topic mode: {exc}"
-        # Reset per-chat debounce state so the user doesn't see a stale
-        # cooldown on the next activation.
-        for attr in ("_telegram_lobby_reminder_ts", "_telegram_capability_hint_ts"):
-            store = getattr(self, attr, None)
-            if isinstance(store, dict):
-                store.pop(chat_id, None)
+        # Reset per-profile+chat debounce state so the user doesn't see a
+        # stale cooldown on the next activation (issue #76423).
+        cooldown_key = self._telegram_topic_cooldown_key(source)
+        if cooldown_key:
+            for attr in ("_telegram_lobby_reminder_ts", "_telegram_capability_hint_ts"):
+                store = getattr(self, attr, None)
+                if isinstance(store, dict):
+                    store.pop(cooldown_key, None)
         return (
             "Multi-session topic mode is now OFF for this chat.\n\n"
             "Existing topics in Telegram aren't removed — they'll just stop "
@@ -20006,6 +20047,7 @@ async def _telegram_topic_root_status_message(self, source: SessionSource) -> st
             sessions = await self._session_db.list_unlinked_telegram_sessions_for_user(
                 chat_id=str(source.chat_id),
                 user_id=str(source.user_id),
+                profile_name=self._telegram_topic_profile_name(source),
                 limit=10,
             )
         except Exception:
@@ -20055,9 +20097,11 @@ async def _restore_telegram_topic_session(self, event: MessageEvent, raw_session
             return "That session does not belong to this Telegram user."
 
         linked = await self._session_db.is_telegram_session_linked_to_topic(session_id=session_id)
+        topic_profile = self._telegram_topic_profile_name(source)
         current_binding = await self._session_db.get_telegram_topic_binding(
             chat_id=str(source.chat_id),
             thread_id=str(source.thread_id),
+            profile_name=topic_profile,
         )
         if linked:
             if not current_binding or current_binding.get("session_id") != session_id:
@@ -20072,6 +20116,7 @@ async def _restore_telegram_topic_session(self, event: MessageEvent, raw_session
                 session_key=session_key,
                 session_id=session_id,
                 managed_mode="restored",
+                profile_name=topic_profile,
             )
         except ValueError as exc:
             if "already linked" in str(exc):
@@ -20435,6 +20480,13 @@ def _thread_metadata_for_source(
             if team_id:
                 metadata = dict(metadata or {})
                 metadata["slack_team_id"] = str(team_id)
+        # Carry routed profile so Telegram prune under profile_routes
+        # namespaces shared state.db rows correctly (issue #76423). Only
+        # stamp when we already have send metadata — do not invent a
+        # metadata dict solely for profile on unthreaded sends.
+        if metadata is not None:
+            metadata = dict(metadata)
+            metadata["hermes_profile"] = self._telegram_topic_profile_name(source)
         return metadata
 
     def _thread_metadata_for_target(

gateway/slash_commands.py

@@ -4234,6 +4234,7 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st
             await self._session_db.enable_telegram_topic_mode(
                 chat_id=str(source.chat_id),
                 user_id=str(source.user_id),
+                profile_name=self._telegram_topic_profile_name(source),
                 has_topics_enabled=capabilities.get("has_topics_enabled"),
                 allows_users_to_create_topics=capabilities.get("allows_users_to_create_topics"),
             )
@@ -4249,6 +4250,7 @@ async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> st
                 binding = await self._session_db.get_telegram_topic_binding(
                     chat_id=str(source.chat_id),
                     thread_id=str(source.thread_id),
+                    profile_name=self._telegram_topic_profile_name(source),
                 )
             except Exception:
                 logger.debug("Failed to read Telegram topic binding", exc_info=True)