fix(telegram): rename runtime-discovered DM topics

gateway/run.py

@@ -13081,23 +13081,38 @@ async def _rename_telegram_topic_for_session_title(
         # Skip rename when the topic is operator-declared via
         # extra.dm_topics. Those topics have fixed names chosen by the
         # operator (plus optional skill binding); auto-renaming would
-        # silently mutate operator config.
-        #
-        # Check the class, not the instance — getattr() on MagicMock
-        # auto-creates attributes, so `hasattr(adapter, "_get_dm_topic_info")`
-        # would return True for every test double.
+        # silently mutate operator config. Do not treat runtime-discovered
+        # topic-name cache entries as operator-declared though: brand-new DM
+        # topics can arrive as "New Thread" and must still be auto-renamed.
         adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
         if adapter is not None:
-            get_info = getattr(type(adapter), "_get_dm_topic_info", None)
-            if callable(get_info):
+            configured_topics = getattr(adapter, "_dm_topics_config", None)
+            if isinstance(configured_topics, list):
                 try:
-                    operator_topic = get_info(adapter, str(source.chat_id), str(source.thread_id))
+                    for chat_entry in configured_topics:
+                        if not isinstance(chat_entry, dict):
+                            continue
+                        if str(chat_entry.get("chat_id")) != str(source.chat_id):
+                            continue
+                        for topic in chat_entry.get("topics", []) or []:
+                            if (
+                                isinstance(topic, dict)
+                                and str(topic.get("thread_id") or "") == str(source.thread_id)
+                            ):
+                                return
                 except Exception:
-                    operator_topic = None
-                # Only treat dict-shaped returns as operator-declared; a
-                # bare MagicMock or other sentinel shouldn't count.
-                if isinstance(operator_topic, dict):
-                    return
+                    logger.debug("Failed to inspect configured Telegram DM topics", exc_info=True)
+            else:
+                # Compatibility fallback for adapter test doubles or older
+                # adapters that expose only _get_dm_topic_info.
+                get_info = getattr(type(adapter), "_get_dm_topic_info", None)
+                if callable(get_info):
+                    try:
+                        operator_topic = get_info(adapter, str(source.chat_id), str(source.thread_id))
+                    except Exception:
+                        operator_topic = None
+                    if isinstance(operator_topic, dict):
+                        return
 
         session_db = getattr(self, "_session_db", None)
         if session_db is not None:

tests/gateway/test_telegram_topic_mode.py

@@ -970,6 +970,49 @@ async def rename_dm_topic(self, **kwargs):
     fake.rename_dm_topic.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_runtime_discovered_topic_is_auto_renamed(tmp_path):
+    """Runtime topic-name cache entries are not operator-declared dm_topics."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.apply_telegram_topic_migration()
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    db.create_session(session_id="sess-topic", source="telegram", user_id="208214988")
+    db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="17585",
+        user_id="208214988",
+        session_key=build_session_key(_make_source(thread_id="17585")),
+        session_id="sess-topic",
+    )
+    runner = _make_runner(session_db=db)
+    runner._telegram_topic_mode_enabled = lambda source: True
+
+    class _FakeAdapter:
+        _dm_topics_config = []
+
+        def _get_dm_topic_info(self, chat_id, thread_id):
+            return {"name": "New Thread"}
+
+        async def rename_dm_topic(self, **kwargs):
+            return None
+
+    fake = _FakeAdapter()
+    fake.rename_dm_topic = AsyncMock()
+    runner.adapters[Platform.TELEGRAM] = fake
+
+    await runner._rename_telegram_topic_for_session_title(
+        _make_source(thread_id="17585"),
+        "sess-topic",
+        "Voice Transcript Title",
+    )
+
+    fake.rename_dm_topic.assert_awaited_once_with(
+        chat_id="208214988",
+        thread_id="17585",
+        name="Voice Transcript Title",
+    )
+
+
 @pytest.mark.asyncio
 async def test_disable_topic_auto_rename_extra_skips_rename(tmp_path):
     """extra.disable_topic_auto_rename=True must short-circuit auto-rename."""