gateway/telegram: prune stale DM topic binding on Thread-not-found (#31501)

Both fallback sites that currently log "Thread X not found,
retrying without message_thread_id" now also drop the
``telegram_dm_topic_bindings`` row keyed on
``(chat_id, thread_id)``:

* The streaming send loop (``send`` body) — fires on the
  second failure, after the same-thread one-shot retry confirms
  the thread really is gone (the first attempt is left alone
  because Bot API has been observed to return a transient
  "Thread not found" that recovers on immediate retry).
* The control-message helper ``_send_message_with_thread_fallback``
  (approval prompts, model picker, update prompts) — single-shot
  retry, prune unconditionally on the BadRequest match.

Without this prune, a user who deletes a Telegram DM topic in
the client keeps getting their next inbound message recovered
back to the dead thread by
``_recover_telegram_topic_thread_id`` in ``gateway/run.py``,
which walks the per-user binding list newest-first and treats
the deleted thread as authoritative.  The reproduction in the
bug report is exactly this: tool progress, approvals, activity
messages and replies all land in the wrong place until the user
manually runs DELETE on state.db.

Cleanup is best-effort — we log at INFO when it succeeds, swallow
any exception from the SessionDB call, and the user-facing send
proceeds either way.

Refs #31501

Author: xxxigm

plugins/platforms/telegram/adapter.py

@@ -810,6 +810,47 @@ def _message_thread_id_for_typing(cls, thread_id: Optional[str]) -> Optional[int
     def _is_thread_not_found_error(error: Exception) -> bool:
         return "thread not found" in str(error).lower()
 
+    def _prune_stale_dm_topic_binding(
+        self, chat_id: Any, thread_id: Any,
+    ) -> None:
+        """Drop the stale ``telegram_dm_topic_bindings`` row for a
+        topic Telegram has confirmed deleted.
+
+        Without this prune the recovery logic in
+        ``gateway.run._recover_telegram_topic_thread_id`` keeps
+        steering future inbound messages to the dead thread (the
+        bug behind #31501 — tool progress, approvals, replies all
+        end up in the wrong place even though the user has moved
+        on to a fresh topic).  Best-effort: we never raise from a
+        send-fallback path — a failed cleanup must not turn into a
+        failed user-facing send.
+        """
+        if chat_id is None or thread_id is None:
+            return
+        store = getattr(self, "_session_store", None)
+        if store is None:
+            return
+        db = getattr(store, "_db", None)
+        if db is None or not hasattr(db, "delete_telegram_topic_binding"):
+            return
+        try:
+            removed = db.delete_telegram_topic_binding(
+                chat_id=str(chat_id), thread_id=str(thread_id),
+            )
+        except Exception:
+            logger.debug(
+                "[%s] delete_telegram_topic_binding failed for "
+                "chat=%s thread=%s — skipping prune",
+                self.name, chat_id, thread_id, exc_info=True,
+            )
+            return
+        if removed:
+            logger.info(
+                "[%s] Pruned stale Telegram DM topic binding "
+                "chat=%s thread=%s (Bot API: thread not found)",
+                self.name, chat_id, thread_id,
+            )
+
     @staticmethod
     def _is_bad_request_error(error: Exception) -> bool:
         name = error.__class__.__name__.lower()
@@ -2670,11 +2711,17 @@ async def send(
                                     continue
                                 # Second failure: the thread is genuinely gone.
                                 # Retry without ``message_thread_id`` so the
-                                # message still reaches the chat.
+                                # message still reaches the chat, and prune
+                                # the stale binding so future inbound
+                                # messages aren't redirected back to it
+                                # (#31501).
                                 logger.warning(
                                     "[%s] Thread %s not found, retrying without message_thread_id",
                                     self.name, effective_thread_id,
                                 )
+                                self._prune_stale_dm_topic_binding(
+                                    chat_id, effective_thread_id,
+                                )
                                 used_thread_fallback = True
                                 effective_thread_id = None
                                 thread_kwargs = {"message_thread_id": None}
@@ -3355,6 +3402,13 @@ async def _send_message_with_thread_fallback(self, **kwargs):
                     self.name,
                     message_thread_id,
                 )
+                # Same prune as the streaming send path — the
+                # control-message retry tells us the topic is gone,
+                # so the binding row in state.db must go too
+                # (#31501).
+                self._prune_stale_dm_topic_binding(
+                    kwargs.get("chat_id"), message_thread_id,
+                )
                 retry_kwargs = dict(kwargs)
                 retry_kwargs.pop("message_thread_id", None)
                 return await self._bot.send_message(**retry_kwargs)