diff --git a/agent/title_generator.py b/agent/title_generator.py
index 3f617093c0..a7f1e158e1 100644
--- a/agent/title_generator.py
+++ b/agent/title_generator.py
@@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
 # so silent-drops (e.g. OpenRouter 402 exhausting the fallback chain)
 # become visible instead of piling up as NULL session titles.
 FailureCallback = Callable[[str, BaseException], None]
+TitleCallback = Callable[[str], None]
 
 _TITLE_PROMPT = (
     "Generate a short, descriptive title (3-7 words) for a conversation that starts with the "
@@ -90,6 +91,7 @@ def auto_title_session(
     assistant_response: str,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
+    title_callback: Optional[TitleCallback] = None,
 ) -> None:
     """Generate and set a session title if one doesn't already exist.
 
@@ -119,6 +121,11 @@ def auto_title_session(
     try:
         session_db.set_session_title(session_id, title)
         logger.debug("Auto-generated session title: %s", title)
+        if title_callback is not None:
+            try:
+                title_callback(title)
+            except Exception:
+                logger.debug("Auto-title callback failed", exc_info=True)
     except Exception as e:
         logger.debug("Failed to set auto-generated title: %s", e)
 
@@ -131,6 +138,7 @@ def maybe_auto_title(
     conversation_history: list,
     failure_callback: Optional[FailureCallback] = None,
     main_runtime: dict = None,
+    title_callback: Optional[TitleCallback] = None,
 ) -> None:
     """Fire-and-forget title generation after the first exchange.
 
@@ -152,7 +160,11 @@ def maybe_auto_title(
     thread = threading.Thread(
         target=auto_title_session,
         args=(session_db, session_id, user_message, assistant_response),
-        kwargs={"failure_callback": failure_callback, "main_runtime": main_runtime},
+        kwargs={
+            "failure_callback": failure_callback,
+            "main_runtime": main_runtime,
+            "title_callback": title_callback,
+        },
         daemon=True,
         name="auto-title",
     )
diff --git a/gateway/assets/telegram-botfather-threads-settings.jpg b/gateway/assets/telegram-botfather-threads-settings.jpg
new file mode 100644
index 0000000000..b1de115acd
Binary files /dev/null and b/gateway/assets/telegram-botfather-threads-settings.jpg differ
diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py
index 167d47237e..ad5ed66920 100644
--- a/gateway/platforms/telegram.py
+++ b/gateway/platforms/telegram.py
@@ -688,6 +688,29 @@ class TelegramAdapter(BasePlatformAdapter):
                 )
             return None
 
+    async def rename_dm_topic(
+        self,
+        chat_id: int,
+        thread_id: int,
+        name: str,
+    ) -> None:
+        """Rename a forum topic in a private (DM) chat."""
+        if not self._bot:
+            return
+        try:
+            chat_id_arg = int(chat_id)
+        except (TypeError, ValueError):
+            chat_id_arg = chat_id
+        await self._bot.edit_forum_topic(
+            chat_id=chat_id_arg,
+            message_thread_id=int(thread_id),
+            name=name,
+        )
+        logger.info(
+            "[%s] Renamed DM topic in chat %s thread_id=%s -> '%s'",
+            self.name, chat_id, thread_id, name,
+        )
+
     def _persist_dm_topic_thread_id(self, chat_id: int, topic_name: str, thread_id: int) -> None:
         """Save a newly created thread_id back into config.yaml so it persists across restarts."""
         try:
diff --git a/gateway/run.py b/gateway/run.py
index 40c4bdb453..6fd19472c2 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -1050,6 +1050,7 @@ class GatewayRunner:
         )
         self.delivery_router = DeliveryRouter(self.config)
         self._running = False
+        self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None
         self._shutdown_event = asyncio.Event()
         self._exit_cleanly = False
         self._exit_with_failure = False
@@ -1493,17 +1494,19 @@ class GatewayRunner:
     def _telegram_topic_root_lobby_message(self) -> str:
         return (
             "This main chat is reserved for system commands.\n\n"
-            "To chat with Hermes, create a new topic using the + button in "
-            "this bot interface. Each topic works as an independent Hermes "
-            "session."
+            "To start a new Hermes chat, open the All Messages topic at the top "
+            "of this bot interface and send any message there. Telegram will "
+            "create a new topic for that message; each topic works as an "
+            "independent Hermes session."
         )
 
     def _telegram_topic_root_new_message(self) -> str:
         return (
-            "To start a new parallel Hermes chat, create a new topic with the "
-            "+ button in this bot interface.\n\n"
-            "Each topic is an independent Hermes session. Use /new inside a "
-            "topic only if you want to replace that topic's current session."
+            "To start a new parallel Hermes chat, open the All Messages topic "
+            "at the top of this bot interface and send any message there. "
+            "Telegram will create a new topic for it.\n\n"
+            "Each topic is an independent Hermes session. Use /new inside an "
+            "existing topic only if you want to replace that topic's current session."
         )
 
     def _telegram_topic_new_header(self, source: SessionSource) -> Optional[str]:
@@ -1511,9 +1514,9 @@ class GatewayRunner:
             return None
         return (
             "Started a new Hermes session in this topic.\n\n"
-            "Tip: for parallel work, create a new topic with the + button "
-            "instead of using /new here. /new replaces the session attached "
-            "to the current topic."
+            "Tip: for parallel work, open All Messages and send a message there "
+            "to create a separate topic instead of using /new here. /new replaces "
+            "the session attached to the current topic."
         )
 
     def _record_telegram_topic_binding(
@@ -2767,6 +2770,10 @@ class GatewayRunner:
         Returns True if at least one adapter connected successfully.
         """
         logger.info("Starting Hermes Gateway...")
+        try:
+            self._gateway_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._gateway_loop = None
         logger.info("Session storage: %s", self.config.sessions_dir)
         # Log the resolved max_iterations budget so operators can verify the
         # config.yaml → env bridge did the right thing at a glance (instead
@@ -9569,7 +9576,193 @@ class GatewayRunner:
             logger.warning("Manual compress failed: %s", e)
             return f"Compression failed: {e}"
 
-    async def _handle_topic_command(self, event: MessageEvent) -> str:
+    async def _get_telegram_topic_capabilities(self, source: SessionSource) -> dict:
+        """Read Telegram private-topic capability flags via Bot API getMe."""
+        adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
+        bot = getattr(adapter, "_bot", None)
+        if bot is None or not hasattr(bot, "get_me"):
+            return {"checked": False}
+        try:
+            me = await bot.get_me()
+        except Exception:
+            logger.debug("Failed to fetch Telegram getMe topic capabilities", exc_info=True)
+            return {"checked": False}
+
+        def _field(name: str):
+            if hasattr(me, name):
+                return getattr(me, name)
+            api_kwargs = getattr(me, "api_kwargs", None)
+            if isinstance(api_kwargs, dict) and name in api_kwargs:
+                return api_kwargs.get(name)
+            if isinstance(me, dict):
+                return me.get(name)
+            return None
+
+        return {
+            "checked": True,
+            "has_topics_enabled": _field("has_topics_enabled"),
+            "allows_users_to_create_topics": _field("allows_users_to_create_topics"),
+        }
+
+    async def _ensure_telegram_system_topic(self, source: SessionSource) -> None:
+        """Create/pin the managed System topic after /topic activation when possible."""
+        adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
+        if adapter is None or not source.chat_id:
+            return
+
+        thread_id = None
+        create_topic = getattr(adapter, "_create_dm_topic", None)
+        if callable(create_topic):
+            try:
+                thread_id = await create_topic(int(source.chat_id), "System")
+            except Exception:
+                logger.debug("Failed to create Telegram System topic", exc_info=True)
+        if not thread_id:
+            return
+
+        message_id = None
+        try:
+            send_result = await adapter.send(
+                source.chat_id,
+                "System topic for Hermes commands and status.",
+                metadata={"thread_id": str(thread_id)},
+            )
+            message_id = getattr(send_result, "message_id", None)
+        except Exception:
+            logger.debug("Failed to send Telegram System topic intro", exc_info=True)
+        if not message_id:
+            return
+
+        bot = getattr(adapter, "_bot", None)
+        if bot is None or not hasattr(bot, "pin_chat_message"):
+            return
+        try:
+            await bot.pin_chat_message(
+                chat_id=int(source.chat_id),
+                message_id=int(message_id),
+                disable_notification=True,
+            )
+        except Exception:
+            logger.debug("Failed to pin Telegram System topic intro", exc_info=True)
+
+    async def _send_telegram_topic_setup_image(self, source: SessionSource) -> None:
+        """Send the bundled BotFather Threads Settings screenshot when available."""
+        adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
+        if adapter is None or not source.chat_id or not hasattr(adapter, "send_image_file"):
+            return
+        image_path = Path(__file__).resolve().parent / "assets" / "telegram-botfather-threads-settings.jpg"
+        if not image_path.exists():
+            return
+        try:
+            await adapter.send_image_file(
+                chat_id=source.chat_id,
+                image_path=str(image_path),
+                caption="BotFather → Bot Settings → Threads Settings",
+                metadata={"thread_id": str(source.thread_id)} if source.thread_id else None,
+            )
+        except Exception:
+            logger.debug("Failed to send Telegram topic setup image", exc_info=True)
+
+    def _sanitize_telegram_topic_title(self, title: str) -> str:
+        """Return a Bot API-safe forum topic name from a generated session title."""
+        cleaned = re.sub(r"\s+", " ", str(title or "")).strip()
+        if not cleaned:
+            return "Hermes Chat"
+        # Telegram forum topic names are short (currently 1-128 chars). Keep
+        # extra room for multi-byte titles and avoid trailing ellipsis churn.
+        if len(cleaned) > 120:
+            cleaned = cleaned[:117].rstrip() + "..."
+        return cleaned
+
+    async def _rename_telegram_topic_for_session_title(
+        self,
+        source: SessionSource,
+        session_id: str,
+        title: str,
+    ) -> None:
+        """Best-effort rename of a Telegram DM topic when Hermes auto-titles a session."""
+        if not self._is_telegram_topic_lane(source) or not source.chat_id or not source.thread_id:
+            return
+        session_db = getattr(self, "_session_db", None)
+        if session_db is not None:
+            try:
+                binding = session_db.get_telegram_topic_binding(
+                    chat_id=str(source.chat_id),
+                    thread_id=str(source.thread_id),
+                )
+                if binding and str(binding.get("session_id") or "") != str(session_id):
+                    return
+            except Exception:
+                logger.debug("Failed to verify Telegram topic binding before rename", exc_info=True)
+                return
+
+        adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
+        if adapter is None:
+            return
+        topic_name = self._sanitize_telegram_topic_title(title)
+        try:
+            rename_topic = getattr(adapter, "rename_dm_topic", None)
+            if rename_topic is not None:
+                await rename_topic(
+                    chat_id=str(source.chat_id),
+                    thread_id=str(source.thread_id),
+                    name=topic_name,
+                )
+                return
+
+            bot = getattr(adapter, "_bot", None)
+            edit_forum_topic = getattr(bot, "edit_forum_topic", None) if bot is not None else None
+            if edit_forum_topic is None:
+                edit_forum_topic = getattr(bot, "editForumTopic", None) if bot is not None else None
+            if edit_forum_topic is None:
+                return
+            try:
+                await edit_forum_topic(
+                    chat_id=int(source.chat_id),
+                    message_thread_id=int(source.thread_id),
+                    name=topic_name,
+                )
+            except (TypeError, ValueError):
+                await edit_forum_topic(
+                    chat_id=source.chat_id,
+                    message_thread_id=source.thread_id,
+                    name=topic_name,
+                )
+        except Exception:
+            logger.debug("Failed to rename Telegram topic for auto-generated title", exc_info=True)
+
+    def _schedule_telegram_topic_title_rename(
+        self,
+        source: SessionSource,
+        session_id: str,
+        title: str,
+    ) -> None:
+        """Schedule a topic rename from the auto-title background thread."""
+        if not title or not self._is_telegram_topic_lane(source):
+            return
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = getattr(self, "_gateway_loop", None)
+        if loop is None or loop.is_closed():
+            return
+        try:
+            copied_source = dataclasses.replace(source)
+        except Exception:
+            copied_source = source
+        future = asyncio.run_coroutine_threadsafe(
+            self._rename_telegram_topic_for_session_title(copied_source, session_id, title),
+            loop,
+        )
+        def _log_rename_failure(fut) -> None:
+            try:
+                fut.result()
+            except Exception:
+                logger.debug("Telegram topic title rename failed", exc_info=True)
+
+        future.add_done_callback(_log_rename_failure)
+
+    async def _handle_topic_command(self, event: MessageEvent, args: str = "") -> str:
         """Handle /topic for Telegram DM user-managed topic sessions."""
         source = event.source
         if source.platform != Platform.TELEGRAM or source.chat_type != "dm":
@@ -9581,20 +9774,48 @@ class GatewayRunner:
         if args:
             if not source.thread_id:
                 return (
-                    "To restore a session, first create or open a Telegram topic "
-                    "with the + button, then send /topic <session-id> inside that topic."
+                    "To restore a session, first create or open a Telegram topic, "
+                    "then send /topic <session-id> inside that topic. To create a "
+                    "new topic, open All Messages and send any message there."
                 )
             return await self._restore_telegram_topic_session(event, args)
 
+        capabilities = await self._get_telegram_topic_capabilities(source)
+        if capabilities.get("checked"):
+            if capabilities.get("has_topics_enabled") is False:
+                await self._send_telegram_topic_setup_image(source)
+                return (
+                    "Telegram topics are not enabled for this bot yet.\n\n"
+                    "How to enable them:\n"
+                    "1. Open @BotFather.\n"
+                    "2. Choose your bot.\n"
+                    "3. Open Bot Settings → Threads Settings.\n"
+                    "4. Turn on Threaded Mode and make sure users are allowed to create new threads.\n\n"
+                    "Then send /topic again."
+                )
+            if capabilities.get("allows_users_to_create_topics") is False:
+                await self._send_telegram_topic_setup_image(source)
+                return (
+                    "Telegram topics are enabled, but users are not allowed to create topics.\n\n"
+                    "Open @BotFather → choose your bot → Bot Settings → Threads Settings, "
+                    "then turn off 'Disallow users to create new threads'.\n\n"
+                    "Then send /topic again."
+                )
+
         try:
             self._session_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"),
+                allows_users_to_create_topics=capabilities.get("allows_users_to_create_topics"),
             )
         except Exception as exc:
             logger.exception("Failed to enable Telegram topic mode")
             return f"Failed to enable Telegram topic mode: {exc}"
 
+        if not source.thread_id:
+            await self._ensure_telegram_system_topic(source)
+
         if source.thread_id:
             try:
                 binding = self._session_db.get_telegram_topic_binding(
@@ -9617,13 +9838,14 @@ class GatewayRunner:
                     f"Session: {session_label}\n"
                     f"ID: {session_id}\n\n"
                     "Use /new to replace this topic with a fresh session.\n"
-                    "For parallel work, create another topic with the + button."
+                    "For parallel work, open All Messages and send a message there "
+                    "to create another topic."
                 )
             return (
                 "Telegram multi-session topics are enabled.\n\n"
                 "This topic will be used as an independent Hermes session. "
                 "Use /new to replace this topic's current session. For parallel "
-                "work, create another topic with the + button."
+                "work, open All Messages and send a message there to create another topic."
             )
 
         return self._telegram_topic_root_status_message(source)
@@ -9632,7 +9854,9 @@ class GatewayRunner:
         lines = [
             "Telegram multi-session topics are enabled.",
             "",
-            "Create new Hermes chats with the + button in this bot interface.",
+            "To create a new Hermes chat, open All Messages at the top of this "
+            "bot interface and send any message there. Telegram will create a "
+            "new topic for it.",
             "",
         ]
         try:
@@ -9658,7 +9882,7 @@ class GatewayRunner:
             lines.extend([
                 "",
                 "To restore one:",
-                "1. Create or open a topic with the + button.",
+                "1. Create or open a topic. To create a new one, open All Messages and send any message there.",
                 "2. Send /topic <session-id> inside that topic.",
                 f"Example: Send /topic {sessions[0].get('id')} inside a topic.",
             ])
@@ -9667,9 +9891,8 @@ class GatewayRunner:
                 "No previous unlinked Telegram sessions found.",
                 "",
                 "To restore a previous session later:",
-                "1. Create a new topic with the + button.",
-                "2. Open that topic.",
-                "3. Send /topic <session-id>.",
+                "1. Create or open a topic. To create a new one, open All Messages and send any message there.",
+                "2. Send /topic <session-id> inside that topic.",
             ])
         return "\n".join(lines)
 
@@ -13549,20 +13772,29 @@ class GatewayRunner:
                     _title_failure_cb = getattr(
                         agent, "_emit_auxiliary_failure", None
                     )
-                    maybe_auto_title(
-                        self._session_db,
-                        effective_session_id,
-                        message,
-                        final_response,
-                        all_msgs,
-                        failure_callback=_title_failure_cb,
-                        main_runtime={
+                    maybe_auto_title_kwargs = {
+                        "failure_callback": _title_failure_cb,
+                        "main_runtime": {
                             "model": getattr(agent, "model", None),
                             "provider": getattr(agent, "provider", None),
                             "base_url": getattr(agent, "base_url", None),
                             "api_key": getattr(agent, "api_key", None),
                             "api_mode": getattr(agent, "api_mode", None),
                         } if agent else None,
+                    }
+                    if self._is_telegram_topic_lane(source):
+                        maybe_auto_title_kwargs["title_callback"] = lambda title: self._schedule_telegram_topic_title_rename(
+                            source,
+                            effective_session_id,
+                            title,
+                        )
+                    maybe_auto_title(
+                        self._session_db,
+                        effective_session_id,
+                        message,
+                        final_response,
+                        all_msgs,
+                        **maybe_auto_title_kwargs,
                     )
                 except Exception:
                     pass
diff --git a/pyproject.toml b/pyproject.toml
index a58e172795..b5de3d69f6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -139,6 +139,7 @@ py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajector
 
 [tool.setuptools.package-data]
 hermes_cli = ["web_dist/**/*"]
+gateway = ["assets/**/*"]
 
 [tool.setuptools.packages.find]
 include = ["agent", "agent.*", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "tui_gateway", "tui_gateway.*", "cron", "acp_adapter", "plugins", "plugins.*"]
diff --git a/tests/agent/test_title_generator.py b/tests/agent/test_title_generator.py
index e10cba76a8..c498a71ab5 100644
--- a/tests/agent/test_title_generator.py
+++ b/tests/agent/test_title_generator.py
@@ -136,6 +136,21 @@ class TestAutoTitleSession:
             auto_title_session(db, "sess-1", "hi", "hello")
             db.set_session_title.assert_called_once_with("sess-1", "New Title")
 
+    def test_invokes_title_callback_after_setting_title(self):
+        db = MagicMock()
+        db.get_session_title.return_value = None
+        seen = []
+        with patch("agent.title_generator.generate_title", return_value="Readable Session"):
+            auto_title_session(
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                title_callback=seen.append,
+            )
+        db.set_session_title.assert_called_once_with("sess-1", "Readable Session")
+        assert seen == ["Readable Session"]
+
     def test_skips_if_generation_fails(self):
         db = MagicMock()
         db.get_session_title.return_value = None
@@ -182,7 +197,13 @@ class TestMaybeAutoTitle:
             import time
             time.sleep(0.3)
             mock_auto.assert_called_once_with(
-                db, "sess-1", "hello", "hi there", failure_callback=None, main_runtime=None
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                failure_callback=None,
+                main_runtime=None,
+                title_callback=None,
             )
 
     def test_forwards_failure_callback_to_worker(self):
@@ -202,7 +223,13 @@ class TestMaybeAutoTitle:
             import time
             time.sleep(0.3)
             mock_auto.assert_called_once_with(
-                db, "sess-1", "hello", "hi there", failure_callback=_cb, main_runtime=None
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                failure_callback=_cb,
+                main_runtime=None,
+                title_callback=None,
             )
 
     def test_skips_if_no_response(self):
diff --git a/tests/gateway/test_telegram_topic_mode.py b/tests/gateway/test_telegram_topic_mode.py
index ad72514ed5..a797b52352 100644
--- a/tests/gateway/test_telegram_topic_mode.py
+++ b/tests/gateway/test_telegram_topic_mode.py
@@ -63,6 +63,10 @@ def _make_runner(session_db=None):
     )
     adapter = MagicMock()
     adapter.send = AsyncMock()
+    adapter.send_image_file = AsyncMock()
+    adapter._bot = None
+    adapter._create_dm_topic = AsyncMock(return_value=None)
+    adapter.rename_dm_topic = AsyncMock()
     runner.adapters = {Platform.TELEGRAM: adapter}
     runner._voice_mode = {}
     runner.hooks = SimpleNamespace(
@@ -150,7 +154,7 @@ async def test_root_telegram_dm_prompt_is_system_lobby_when_topic_mode_enabled(m
     result = await runner._handle_message(_make_event("hello from root"))
 
     assert "main chat is reserved for system commands" in result
-    assert "+ button" in result
+    assert "All Messages" in result
     runner._run_agent.assert_not_called()
     runner.session_store.get_or_create_session.assert_not_called()
 
@@ -172,8 +176,8 @@ async def test_root_telegram_dm_new_shows_create_topic_instruction(monkeypatch):
     result = await runner._handle_message(_make_event("/new"))
 
     assert "create a new topic" in result
-    assert "+ button" in result
-    assert "Use /new inside a topic" in result
+    assert "All Messages" in result
+    assert "Use /new inside" in result
     runner._run_agent.assert_not_called()
     runner.session_store.reset_session.assert_not_called()
     runner.session_store.get_or_create_session.assert_not_called()
@@ -357,8 +361,8 @@ async def test_new_inside_telegram_topic_resets_current_topic_with_parallel_tip(
     result = await runner._handle_message(_make_event("/new", thread_id="17585"))
 
     assert "Started a new Hermes session in this topic" in result
-    assert "for parallel work" in result
-    assert "+ button" in result
+    assert "parallel work" in result
+    assert "All Messages" in result
     runner.session_store.reset_session.assert_called_once_with(topic_key)
 
 
@@ -379,7 +383,7 @@ async def test_topic_root_command_explicitly_migrates_and_enables_topic_mode(tmp
     result = await runner._handle_message(_make_event("/topic"))
 
     assert "Telegram multi-session topics are enabled" in result
-    assert "+ button" in result
+    assert "All Messages" in result
     assert session_db.get_meta("telegram_dm_topic_schema_version") == "1"
     assert session_db.is_telegram_topic_mode_enabled(chat_id="208214988", user_id="208214988")
     assert runner._telegram_topic_mode_enabled(_make_source()) is True
@@ -462,7 +466,7 @@ async def test_topic_root_command_handles_no_unlinked_sessions(tmp_path, monkeyp
 
     assert "Telegram multi-session topics are enabled" in result
     assert "No previous unlinked Telegram sessions found" in result
-    assert "+ button" in result
+    assert "All Messages" in result
     runner._run_agent.assert_not_called()
 
 
@@ -623,3 +627,124 @@ async def test_first_message_inside_topic_records_topic_binding(tmp_path, monkey
     assert binding["user_id"] == "208214988"
     assert binding["session_id"] == "sess-topic"
     assert binding["session_key"] == build_session_key(_make_source(thread_id="17585"))
+
+
+@pytest.mark.asyncio
+async def test_topic_root_command_checks_getme_capabilities_before_enabling(tmp_path, monkeypatch):
+    import gateway.run as gateway_run
+
+    session_db = SessionDB(db_path=tmp_path / "state.db")
+    runner = _make_runner(session_db=session_db)
+    bot = AsyncMock()
+    bot.get_me.return_value = SimpleNamespace(
+        has_topics_enabled=False,
+        allows_users_to_create_topics=True,
+    )
+    runner.adapters[Platform.TELEGRAM]._bot = bot
+    runner._run_agent = AsyncMock(
+        side_effect=AssertionError("/topic capability failure must not enter the agent loop")
+    )
+
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    result = await runner._handle_message(_make_event("/topic"))
+
+    assert "topics are not enabled" in result
+    assert "Open @BotFather" in result
+    assert session_db.is_telegram_topic_mode_enabled(chat_id="208214988", user_id="208214988") is False
+    bot.get_me.assert_awaited_once()
+    runner.adapters[Platform.TELEGRAM].send_image_file.assert_awaited_once()
+    image_kwargs = runner.adapters[Platform.TELEGRAM].send_image_file.await_args.kwargs
+    assert image_kwargs["chat_id"] == "208214988"
+    assert image_kwargs["image_path"].endswith("telegram-botfather-threads-settings.jpg")
+    runner._run_agent.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_topic_root_command_creates_and_pins_system_topic(tmp_path, monkeypatch):
+    import gateway.run as gateway_run
+
+    session_db = SessionDB(db_path=tmp_path / "state.db")
+    runner = _make_runner(session_db=session_db)
+    adapter = runner.adapters[Platform.TELEGRAM]
+    adapter._create_dm_topic.return_value = 4242
+    adapter.send.return_value = SimpleNamespace(success=True, message_id="777")
+    bot = AsyncMock()
+    bot.get_me.return_value = {
+        "has_topics_enabled": True,
+        "allows_users_to_create_topics": True,
+    }
+    adapter._bot = bot
+
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    result = await runner._handle_message(_make_event("/topic"))
+
+    assert "Telegram multi-session topics are enabled" in result
+    adapter._create_dm_topic.assert_awaited_once_with(208214988, "System")
+    adapter.send.assert_awaited_once_with(
+        "208214988",
+        "System topic for Hermes commands and status.",
+        metadata={"thread_id": "4242"},
+    )
+    bot.pin_chat_message.assert_awaited_once_with(
+        chat_id=208214988,
+        message_id=777,
+        disable_notification=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_auto_generated_title_renames_bound_telegram_topic(tmp_path):
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.apply_telegram_topic_migration()
+    db.create_session("sess-topic", source="telegram", user_id="208214988")
+    db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="42",
+        user_id="208214988",
+        session_key="agent:main:telegram:dm:208214988:42",
+        session_id="sess-topic",
+    )
+    runner = _make_runner(session_db=db)
+    runner._telegram_topic_mode_enabled = lambda source: True
+
+    await runner._rename_telegram_topic_for_session_title(
+        _make_source(thread_id="42"),
+        "sess-topic",
+        "  Build   Telegram Topic UX  ",
+    )
+
+    runner.adapters[Platform.TELEGRAM].rename_dm_topic.assert_awaited_once_with(
+        chat_id="208214988",
+        thread_id="42",
+        name="Build Telegram Topic UX",
+    )
+
+
+@pytest.mark.asyncio
+async def test_auto_generated_title_does_not_rename_topic_bound_to_other_session(tmp_path):
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.apply_telegram_topic_migration()
+    db.create_session("sess-other", source="telegram", user_id="208214988")
+    db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="42",
+        user_id="208214988",
+        session_key="agent:main:telegram:dm:208214988:42",
+        session_id="sess-other",
+    )
+    runner = _make_runner(session_db=db)
+    runner._telegram_topic_mode_enabled = lambda source: True
+
+    await runner._rename_telegram_topic_for_session_title(
+        _make_source(thread_id="42"),
+        "sess-topic",
+        "Wrong Session Title",
+    )
+
+    runner.adapters[Platform.TELEGRAM].rename_dm_topic.assert_not_called()