fix(gateway): keep Telegram topic fork on branch

gateway/slash_commands.py

@@ -3898,6 +3898,19 @@ async def _list_titled_sessions() -> list[dict]:
         if not new_entry:
             return t("gateway.resume.switch_failed")
 
+        sync_topic_binding = getattr(self, "_sync_telegram_topic_binding", None)
+        if callable(sync_topic_binding):
+            try:
+                await asyncio.to_thread(
+                    sync_topic_binding,
+                    source,
+                    new_entry,
+                    reason="resume-command",
+                )
+            except Exception:
+                logger.debug("Failed to sync Telegram topic binding after resume", exc_info=True)
+
+
         # Conversation boundary: clear ALL conversation-scoped per-session
         # state (model/reasoning overrides #10702, one-turn restores, model
         # notes, last-resolved cache #58403, /queue overflow) + security
@@ -4091,6 +4104,26 @@ async def _handle_branch_command(self, event: MessageEvent) -> str:
         new_entry = await self.async_session_store.switch_session(session_key, new_session_id)
         if not new_entry:
             return t("gateway.branch.switch_failed")
+
+        sync_topic_binding = getattr(self, "_sync_telegram_topic_binding", None)
+        if callable(sync_topic_binding):
+            try:
+                await asyncio.to_thread(
+                    sync_topic_binding,
+                    source,
+                    new_entry,
+                    reason="branch-command",
+                )
+            except Exception:
+                logger.debug("Failed to sync Telegram topic binding after branch", exc_info=True)
+
+        rename_topic = getattr(self, "_rename_telegram_topic_for_session_title", None)
+        if callable(rename_topic):
+            try:
+                await rename_topic(source, new_session_id, branch_title)
+            except Exception:
+                logger.debug("Failed to rename Telegram topic after branch", exc_info=True)
+
         self._clear_session_boundary_security_state(session_key)
 
         # Evict any cached agent for this session

tests/gateway/test_telegram_topic_mode.py

@@ -458,6 +458,156 @@ async def test_new_inside_telegram_topic_rewrites_binding_to_new_session(tmp_pat
     assert binding["session_id"] == "new-topic-session"
 
 
+@pytest.mark.asyncio
+async def test_branch_inside_telegram_topic_rewrites_binding_and_renames_topic(tmp_path):
+    """Regression: /branch inside a Telegram DM topic must keep the topic on the branch.
+
+    Without rewriting telegram_dm_topic_bindings, the next inbound message in
+    the same topic resolves the stale parent binding and immediately switches
+    back to the original session, making /fork look like a no-op.
+    """
+    session_db = SessionDB(db_path=tmp_path / "state.db")
+    session_db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    session_db.create_session(
+        session_id="parent-topic-session",
+        source="telegram",
+        user_id="208214988",
+    )
+    topic_source = _make_source(thread_id="17585")
+    topic_key = build_session_key(topic_source)
+    session_db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="17585",
+        user_id="208214988",
+        session_key=topic_key,
+        session_id="parent-topic-session",
+    )
+
+    parent_entry = SessionEntry(
+        session_key=topic_key,
+        session_id="parent-topic-session",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        origin=topic_source,
+    )
+
+    def _switch_session(session_key, target_session_id):
+        return SessionEntry(
+            session_key=session_key,
+            session_id=target_session_id,
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            platform=Platform.TELEGRAM,
+            chat_type="dm",
+            origin=topic_source,
+        )
+
+    runner = _make_runner(session_db=session_db)
+    runner._async_session_store = SimpleNamespace(
+        _store=runner.session_store,
+        get_or_create_session=AsyncMock(return_value=parent_entry),
+        load_transcript=AsyncMock(return_value=[
+            {"role": "user", "content": "research ozon"},
+            {"role": "assistant", "content": "working"},
+        ]),
+        switch_session=AsyncMock(side_effect=_switch_session),
+    )
+
+    result = await runner._handle_branch_command(_make_event("/fork ozon", thread_id="17585"))
+
+    assert "Branched to" in result
+    assert "ozon" in result
+    branch_id = runner.async_session_store.switch_session.await_args.args[1]
+    runner.async_session_store.switch_session.assert_awaited_once_with(topic_key, branch_id)
+    binding = session_db.get_telegram_topic_binding(
+        chat_id="208214988", thread_id="17585",
+    )
+    assert binding is not None
+    assert binding["session_id"] == branch_id
+    assert binding["session_id"] != "parent-topic-session"
+    runner.adapters[Platform.TELEGRAM].rename_dm_topic.assert_awaited_once_with(
+        chat_id="208214988",
+        thread_id="17585",
+        name="ozon",
+    )
+
+
+@pytest.mark.asyncio
+async def test_resume_inside_telegram_topic_rewrites_binding(tmp_path):
+    """Regression: /resume must persist the selected session for the topic lane."""
+    session_db = SessionDB(db_path=tmp_path / "state.db")
+    session_db.enable_telegram_topic_mode(chat_id="208214988", user_id="208214988")
+    session_db.create_session(
+        session_id="current-topic-session",
+        source="telegram",
+        user_id="208214988",
+        chat_id="208214988",
+        chat_type="dm",
+        thread_id="17585",
+    )
+    session_db.create_session(
+        session_id="target-topic-session",
+        source="telegram",
+        user_id="208214988",
+        chat_id="208214988",
+        chat_type="dm",
+        thread_id="17585",
+    )
+    session_db.set_session_title("target-topic-session", "Earlier Topic Work")
+
+    topic_source = _make_source(thread_id="17585")
+    topic_key = build_session_key(topic_source)
+    session_db.bind_telegram_topic(
+        chat_id="208214988",
+        thread_id="17585",
+        user_id="208214988",
+        session_key=topic_key,
+        session_id="current-topic-session",
+    )
+
+    current_entry = SessionEntry(
+        session_key=topic_key,
+        session_id="current-topic-session",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        origin=topic_source,
+    )
+    target_entry = SessionEntry(
+        session_key=topic_key,
+        session_id="target-topic-session",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        origin=topic_source,
+    )
+    runner = _make_runner(session_db=session_db)
+    runner._async_session_store = SimpleNamespace(
+        _store=runner.session_store,
+        get_or_create_session=AsyncMock(return_value=current_entry),
+        load_transcript=AsyncMock(return_value=[]),
+        switch_session=AsyncMock(return_value=target_entry),
+    )
+
+    result = await runner._handle_resume_command(
+        _make_event("/resume Earlier Topic Work", thread_id="17585")
+    )
+
+    assert "Resumed" in result
+    runner.async_session_store.switch_session.assert_awaited_once_with(
+        topic_key, "target-topic-session"
+    )
+    binding = session_db.get_telegram_topic_binding(
+        chat_id="208214988", thread_id="17585",
+    )
+    assert binding is not None
+    assert binding["session_id"] == "target-topic-session"
+
+
 @pytest.mark.asyncio
 async def test_topic_binding_follows_compression_tip_on_read(tmp_path, monkeypatch):
     """Stale topic bindings auto-heal to the compression child on next inbound.