fix(telegram): polish topic mode — CASCADE, General-topic handling, rename guard, debounce

Five follow-ups to topic mode based on integration audit:

1. ON DELETE CASCADE on telegram_dm_topic_bindings.session_id. Session
   pruning (manual /delete, auto-cleanup, any future prune job) would
   have thrown 'FOREIGN KEY constraint failed' for sessions bound to a
   topic. Migration bumped to v2, rebuilds the bindings table in place
   if FK lacks CASCADE. Idempotent; only runs once per DB.

2. Never auto-rename operator-declared topics. If an operator has
   extra.dm_topics configured AND a user runs /topic, messages in those
   pre-declared topics would previously trigger auto-rename and silently
   mutate operator config. _rename_telegram_topic_for_session_title now
   early-returns when _get_dm_topic_info returns a dict for this
   (chat_id, thread_id). Uses class-based lookup (not hasattr) so
   MagicMock test fixtures don't accidentally trip the guard.

3. General topic handling. Telegram's General (pinned top) topic in a
   forum-enabled private chat may send messages with message_thread_id=1
   or omit thread_id entirely depending on client. Both are now treated
   as the root lobby, not a topic lane. Prevents users from
   accidentally burning a session on the General topic.

4. Debounce the root-lobby reminder. 30-second cooldown per chat so a
   user who forgets topic mode is enabled and types ten messages in the
   root gets one reminder, not ten. Explicit command replies
   (/new-in-lobby, /topic <session-id>) still land every time.

5. Docs: added under-the-hood invariants for the above, plus a
   Downgrade section explaining that rolling back to a pre-/topic
   Hermes build leaves the DB tables orphaned but harmless — DMs just
   revert to native per-thread isolation.

Tests:
- test_operator_declared_topic_is_not_auto_renamed
- test_general_topic_is_treated_as_root_lobby
- test_lobby_reminder_is_debounced_per_chat
- test_binding_survives_session_deletion_via_cascade
- test_migration_rebuilds_v1_binding_table_with_cascade_fk

Validated: 4803/4804 tests pass (tests/gateway/ + tests/test_hermes_state.py).
Sole failure is a pre-existing test_teams::test_send_typing flake
unrelated to this PR.

gateway/run.py

@@ -1475,23 +1475,52 @@ class GatewayRunner:
         # opt into topic mode) means topic mode is off for this chat.
         return raw is True
 
+    # Telegram's General (pinned top) topic in forum-enabled private chats.
+    # Bot API behavior varies: some clients omit message_thread_id for
+    # General, others send "1". Treat both as "root" for lobby/lane purposes.
+    _TELEGRAM_GENERAL_TOPIC_IDS = frozenset({"", "1"})
+
     def _is_telegram_topic_root_lobby(self, source: SessionSource) -> bool:
-        """True for the main Telegram DM when topic mode has made it a lobby."""
-        return (
-            source.platform == Platform.TELEGRAM
-            and source.chat_type == "dm"
-            and not source.thread_id
-            and self._telegram_topic_mode_enabled(source)
-        )
+        """True for the main Telegram DM (or General topic) when topic mode has made it a lobby."""
+        if source.platform != Platform.TELEGRAM or source.chat_type != "dm":
+            return False
+        if not self._telegram_topic_mode_enabled(source):
+            return False
+        tid = str(source.thread_id or "")
+        return tid in self._TELEGRAM_GENERAL_TOPIC_IDS
 
     def _is_telegram_topic_lane(self, source: SessionSource) -> bool:
         """True for a user-created Telegram private-chat topic lane."""
-        return (
-            source.platform == Platform.TELEGRAM
-            and source.chat_type == "dm"
-            and bool(source.thread_id)
-            and self._telegram_topic_mode_enabled(source)
-        )
+        if source.platform != Platform.TELEGRAM or source.chat_type != "dm":
+            return False
+        if not self._telegram_topic_mode_enabled(source):
+            return False
+        tid = str(source.thread_id or "")
+        if not tid or tid in self._TELEGRAM_GENERAL_TOPIC_IDS:
+            return False
+        return True
+
+    _TELEGRAM_LOBBY_REMINDER_COOLDOWN_S = 30.0
+
+    def _should_send_telegram_lobby_reminder(self, source: SessionSource) -> bool:
+        """Rate-limit root-DM lobby reminders to one message per cooldown window.
+
+        A user who forgets multi-session mode is enabled and types several
+        prompts in the root DM would otherwise get a reminder for every
+        message. Cap it so the first one lands and the rest stay quiet.
+        """
+        if not hasattr(self, "_telegram_lobby_reminder_ts"):
+            self._telegram_lobby_reminder_ts = {}
+        chat_id = str(source.chat_id or "")
+        if not chat_id:
+            return True
+        import time as _time
+        now = _time.monotonic()
+        last = self._telegram_lobby_reminder_ts.get(chat_id, 0.0)
+        if now - last < self._TELEGRAM_LOBBY_REMINDER_COOLDOWN_S:
+            return False
+        self._telegram_lobby_reminder_ts[chat_id] = now
+        return True
 
     def _telegram_topic_root_lobby_message(self) -> str:
         return (
@@ -5617,7 +5646,11 @@ class GatewayRunner:
         # execution of a dangerous command.
 
         if self._is_telegram_topic_root_lobby(source):
-            return self._telegram_topic_root_lobby_message()
+            # Debounce the lobby reminder so a user who forgets about
+            # topic mode and fires ten prompts doesn't get ten copies.
+            if self._should_send_telegram_lobby_reminder(source):
+                return self._telegram_topic_root_lobby_message()
+            return None
 
         # ── Claim this session before any await ───────────────────────
         # Between here and _run_agent registering the real AIAgent, there
@@ -9705,6 +9738,28 @@ class GatewayRunner:
         """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
+
+        # 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.
+        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):
+                try:
+                    operator_topic = get_info(adapter, str(source.chat_id), str(source.thread_id))
+                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
+
         session_db = getattr(self, "_session_db", None)
         if session_db is not None:
             try:
@@ -9718,7 +9773,6 @@ class GatewayRunner:
                 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)

hermes_state.py

@@ -2155,6 +2155,11 @@ class SessionDB:
         reconciliation. Operators must be able to upgrade Hermes, keep the old
         Telegram bot behavior running, and only mutate topic-mode state when the
         user executes /topic to opt into the feature.
+
+        Schema versions:
+          v1 — initial shape (no ON DELETE CASCADE on session_id FK)
+          v2 — session_id FK gets ON DELETE CASCADE so session pruning
+               automatically clears bindings.
         """
         def _do(conn):
             conn.executescript(
@@ -2177,7 +2182,7 @@ class SessionDB:
                     thread_id TEXT NOT NULL,
                     user_id TEXT NOT NULL,
                     session_key TEXT NOT NULL,
-                    session_id TEXT NOT NULL REFERENCES sessions(id),
+                    session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
                     managed_mode TEXT NOT NULL DEFAULT 'auto',
                     linked_at REAL NOT NULL,
                     updated_at REAL NOT NULL,
@@ -2191,10 +2196,55 @@ class SessionDB:
                 ON telegram_dm_topic_bindings(user_id, chat_id);
                 """
             )
+
+            # v1 → v2: rebuild telegram_dm_topic_bindings if its session_id FK
+            # lacks ON DELETE CASCADE. SQLite can't ALTER a foreign key, so we
+            # rebuild the table. Only runs once per DB (version gate).
+            current = conn.execute(
+                "SELECT value FROM state_meta WHERE key = ?",
+                ("telegram_dm_topic_schema_version",),
+            ).fetchone()
+            current_version = int(current[0]) if current and str(current[0]).isdigit() else 0
+            if current_version < 2:
+                fk_rows = conn.execute(
+                    "PRAGMA foreign_key_list('telegram_dm_topic_bindings')"
+                ).fetchall()
+                needs_rebuild = any(
+                    row[2] == "sessions" and (row[6] or "") != "CASCADE"
+                    for row in fk_rows
+                )
+                if needs_rebuild:
+                    conn.executescript(
+                        """
+                        CREATE TABLE telegram_dm_topic_bindings_new (
+                            chat_id TEXT NOT NULL,
+                            thread_id TEXT NOT NULL,
+                            user_id TEXT NOT NULL,
+                            session_key TEXT NOT NULL,
+                            session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
+                            managed_mode TEXT NOT NULL DEFAULT 'auto',
+                            linked_at REAL NOT NULL,
+                            updated_at REAL NOT NULL,
+                            PRIMARY KEY (chat_id, thread_id)
+                        );
+                        INSERT INTO telegram_dm_topic_bindings_new
+                            SELECT chat_id, thread_id, user_id, session_key,
+                                   session_id, managed_mode, linked_at, updated_at
+                            FROM telegram_dm_topic_bindings;
+                        DROP TABLE telegram_dm_topic_bindings;
+                        ALTER TABLE telegram_dm_topic_bindings_new
+                            RENAME TO telegram_dm_topic_bindings;
+                        CREATE UNIQUE INDEX idx_telegram_dm_topic_bindings_session
+                            ON telegram_dm_topic_bindings(session_id);
+                        CREATE INDEX idx_telegram_dm_topic_bindings_user
+                            ON telegram_dm_topic_bindings(user_id, chat_id);
+                        """
+                    )
+
             conn.execute(
                 "INSERT INTO state_meta (key, value) VALUES (?, ?) "
                 "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
-                ("telegram_dm_topic_schema_version", "1"),
+                ("telegram_dm_topic_schema_version", "2"),
             )
         self._execute_write(_do)
 

tests/gateway/test_telegram_topic_mode.py

@@ -461,7 +461,7 @@ async def test_topic_root_command_explicitly_migrates_and_enables_topic_mode(tmp
 
     assert "Telegram multi-session topics are enabled" in result
     assert "All Messages" in result
-    assert session_db.get_meta("telegram_dm_topic_schema_version") == "1"
+    assert session_db.get_meta("telegram_dm_topic_schema_version") == "2"
     assert session_db.is_telegram_topic_mode_enabled(chat_id="208214988", user_id="208214988")
     assert runner._telegram_topic_mode_enabled(_make_source()) is True
     runner._run_agent.assert_not_called()
@@ -825,3 +825,161 @@ async def test_auto_generated_title_does_not_rename_topic_bound_to_other_session
     )
 
     runner.adapters[Platform.TELEGRAM].rename_dm_topic.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_operator_declared_topic_is_not_auto_renamed(tmp_path):
+    """Topics registered in extra.dm_topics keep their operator-chosen name."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    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
+
+    # Give the adapter a concrete class with _get_dm_topic_info so the
+    # class-based lookup in _rename_telegram_topic_for_session_title
+    # actually finds it (a MagicMock auto-attr would be skipped).
+    class _FakeAdapter:
+        def _get_dm_topic_info(self, chat_id, thread_id):
+            return {"name": "Research", "skill": "arxiv"}
+
+        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",
+        "Auto-generated title",
+    )
+
+    fake.rename_dm_topic.assert_not_called()
+
+
+def test_general_topic_is_treated_as_root_lobby(tmp_path):
+    """Messages in the Telegram General topic (thread_id=1) route to the lobby, not a lane."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    runner = _make_runner(session_db=db)
+
+    general_source = _make_source(thread_id="1")
+    assert runner._is_telegram_topic_root_lobby(general_source) is True
+    assert runner._is_telegram_topic_lane(general_source) is False
+
+    no_thread_source = _make_source(thread_id=None)
+    assert runner._is_telegram_topic_root_lobby(no_thread_source) is True
+    assert runner._is_telegram_topic_lane(no_thread_source) is False
+
+    real_topic = _make_source(thread_id="17585")
+    assert runner._is_telegram_topic_root_lobby(real_topic) is False
+    assert runner._is_telegram_topic_lane(real_topic) is True
+
+
+def test_lobby_reminder_is_debounced_per_chat(tmp_path):
+    """Consecutive root-DM prompts should only surface one lobby reminder per cooldown."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    runner = _make_runner(session_db=db)
+
+    source = _make_source(thread_id=None)
+    assert runner._should_send_telegram_lobby_reminder(source) is True
+    # Next call inside the cooldown window must return False.
+    assert runner._should_send_telegram_lobby_reminder(source) is False
+    assert runner._should_send_telegram_lobby_reminder(source) is False
+
+    # A different chat gets its own window.
+    other = _make_source(thread_id=None)
+    # Swap chat_id so the debounce key is different.
+    from dataclasses import replace
+    other = replace(other, chat_id="999999999")
+    assert runner._should_send_telegram_lobby_reminder(other) is True
+
+
+def test_binding_survives_session_deletion_via_cascade(tmp_path):
+    """Deleting a session with a topic binding must not raise FK errors."""
+    import sqlite3
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    db.create_session(session_id="sess-to-delete", source="telegram", user_id="208214988")
+    db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="17585",
+        user_id="208214988",
+        session_key="agent:main:telegram:dm:208214988:17585",
+        session_id="sess-to-delete",
+    )
+
+    # Before: binding exists.
+    binding = db.get_telegram_topic_binding(chat_id="208214988", thread_id="17585")
+    assert binding is not None
+
+    # Delete the session. Without ON DELETE CASCADE this would raise
+    # sqlite3.IntegrityError: FOREIGN KEY constraint failed.
+    db._conn.execute("DELETE FROM sessions WHERE id = ?", ("sess-to-delete",))
+    db._conn.commit()
+
+    # After: binding row automatically cleared.
+    binding_after = db.get_telegram_topic_binding(chat_id="208214988", thread_id="17585")
+    assert binding_after is None
+
+
+def test_migration_rebuilds_v1_binding_table_with_cascade_fk(tmp_path):
+    """v1 → v2 migration rebuilds the bindings table when FK lacks ON DELETE CASCADE."""
+    import sqlite3
+    db_path = tmp_path / "state.db"
+    db = SessionDB(db_path=db_path)
+
+    # Simulate a v1-shaped DB: migration ran without ON DELETE CASCADE.
+    db.apply_telegram_topic_migration()  # Creates v2 (our new shape)
+    # Drop the v2 bindings table and recreate it in the old v1 shape.
+    with db._lock:
+        db._conn.execute("DROP TABLE telegram_dm_topic_bindings")
+        db._conn.execute(
+            """
+            CREATE TABLE telegram_dm_topic_bindings (
+                chat_id TEXT NOT NULL,
+                thread_id TEXT NOT NULL,
+                user_id TEXT NOT NULL,
+                session_key TEXT NOT NULL,
+                session_id TEXT NOT NULL REFERENCES sessions(id),
+                managed_mode TEXT NOT NULL DEFAULT 'auto',
+                linked_at REAL NOT NULL,
+                updated_at REAL NOT NULL,
+                PRIMARY KEY (chat_id, thread_id)
+            )
+            """
+        )
+        # Also rewind the version marker so migration treats this as v1.
+        db._conn.execute(
+            "UPDATE state_meta SET value = '1' WHERE key = 'telegram_dm_topic_schema_version'"
+        )
+        db._conn.commit()
+
+    # Sanity check: FK has no CASCADE action yet.
+    fk_rows = db._conn.execute(
+        "PRAGMA foreign_key_list('telegram_dm_topic_bindings')"
+    ).fetchall()
+    assert any(row[2] == "sessions" and (row[6] or "") != "CASCADE" for row in fk_rows)
+
+    # Re-run migration — should upgrade to v2 shape.
+    db.apply_telegram_topic_migration()
+
+    fk_rows_after = db._conn.execute(
+        "PRAGMA foreign_key_list('telegram_dm_topic_bindings')"
+    ).fetchall()
+    assert any(row[2] == "sessions" and row[6] == "CASCADE" for row in fk_rows_after)
+
+    version = db._conn.execute(
+        "SELECT value FROM state_meta WHERE key = 'telegram_dm_topic_schema_version'"
+    ).fetchone()
+    assert version is not None and version[0] == "2"

tests/test_hermes_state.py

@@ -1565,7 +1565,7 @@ class TestSchemaInit:
         }
         assert "telegram_dm_topic_mode" in tables
         assert "telegram_dm_topic_bindings" in tables
-        assert db.get_meta("telegram_dm_topic_schema_version") == "1"
+        assert db.get_meta("telegram_dm_topic_schema_version") == "2"
         db.close()
 
     def test_telegram_topic_binding_roundtrip_requires_explicit_schema(self, tmp_path):
@@ -1593,7 +1593,7 @@ class TestSchemaInit:
         assert binding["user_id"] == "208214988"
         assert binding["session_key"] == "telegram:dm:208214988:thread:17585"
         assert binding["session_id"] == "topic-session"
-        assert db.get_meta("telegram_dm_topic_schema_version") == "1"
+        assert db.get_meta("telegram_dm_topic_schema_version") == "2"
         db.close()
 
     def test_telegram_topic_binding_refuses_to_relink_session_to_another_topic(self, tmp_path):

website/docs/user-guide/messaging/telegram.md

@@ -480,10 +480,14 @@ Shows the current topic's binding: session title, session ID, and hints for `/ne
 ### Under the hood
 
 - Activation persists to `telegram_dm_topic_mode(chat_id, user_id, enabled, ...)` in `state.db`
-- Each topic binding persists to `telegram_dm_topic_bindings(chat_id, thread_id, session_id, ...)`
+- Each topic binding persists to `telegram_dm_topic_bindings(chat_id, thread_id, session_id, ...)` with `ON DELETE CASCADE` on `session_id` — pruning a session automatically clears its topic binding
 - The topic-mode SQLite migration is **opt-in**: it runs on the first `/topic` call, never on gateway startup. Until a user runs `/topic` in this profile, `state.db` is unchanged
 - Each inbound DM message looks up its `(chat_id, thread_id)` binding. If present, the lookup routes the message to the bound session via `SessionStore.switch_session()` so the session-key-to-session-id mapping stays consistent on disk
 - `/new` inside a topic rewrites the binding row to point at the new session ID, so the next message stays on the fresh session
+- Topics declared in `extra.dm_topics` are **never auto-renamed** — the operator-chosen name is preserved even when multi-session mode is enabled
+- The General (pinned top) topic in a forum-enabled DM is treated as the root lobby, regardless of whether Telegram delivers its messages with `message_thread_id=1` or with no thread_id
+- Root-lobby reminders are rate-limited to one message per 30 seconds per chat — a user who forgets topic mode is on and types ten prompts in the root won't get ten replies
+- `/background <prompt>` started inside a topic delivers its result back to the same topic; background sessions don't trigger auto-rename of the owning topic
 
 ### Disabling multi-session mode
 
@@ -496,6 +500,10 @@ sqlite3 ~/.hermes/state.db \
 
 Existing topics in Telegram won't disappear — they'll just stop being gated as independent sessions on the Hermes side. The binding rows can also be cleared with `DELETE FROM telegram_dm_topic_bindings WHERE chat_id = '<your_chat_id>'`.
 
+### Downgrading Hermes
+
+If you downgrade to a Hermes version that predates `/topic`, the feature simply stops working — the `telegram_dm_topic_mode` and `telegram_dm_topic_bindings` tables remain in `state.db` but are ignored by older code. DMs revert to the native per-thread isolation (each `message_thread_id` still gets its own session via `build_session_key`), so your existing Telegram topics keep working as parallel sessions. The root DM is no longer a lobby — messages there go into the agent like they used to. Re-upgrading reactivates multi-session mode exactly where it was.
+
 ## Group Forum Topic Skill Binding
 
 Supergroups with **Topics mode** enabled (also called "forum topics") already get session isolation per topic — each `thread_id` maps to its own conversation. But you may want to **auto-load a skill** when messages arrive in a specific group topic, just like DM topic skill binding works.