fix(telegram): retry DM thread sends without message_thread_id [AI-assisted]

liuxiaopai-ai · steipete · commit e6e3a7b497f1 · 2026-03-02T02:58:15.000Z
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -178,6 +178,7 @@ Docs: https://docs.openclaw.ai
 - Security/External content marker folding: expand Unicode angle-bracket homoglyph normalization in marker sanitization so additional guillemet, double-angle, tortoise-shell, flattened-parenthesis, and ornamental variants are folded before boundary replacement. (#30951) Thanks @benediktjohannes.
 - Docs/Slack manifest scopes: add missing DM/group-DM bot scopes (`im:read`, `im:write`, `mpim:read`, `mpim:write`) to the Slack app manifest example so DM setup guidance is complete. (#29999) Thanks @JcMinarro.
 - Slack/Onboarding token help: update setup text to include the “From manifest” app-creation path and current install wording for obtaining the `xoxb-` bot token. (#30846) Thanks @yzhong52.
+- Telegram/Thread fallback safety: when Telegram returns `message thread not found`, retry without `message_thread_id` only for DM-thread sends (not forum topics), and suppress first-attempt danger logs when retry succeeds. Landed from contributor PR #30892 by @liuxiaopai-ai. Thanks @liuxiaopai-ai.
 - Slack/Bot attachment-only messages: when `allowBots: true`, bot messages with empty `text` now include non-forwarded attachment `text`/`fallback` content so webhook alerts are not silently dropped. (#27616) Thanks @lailoo.
 - Slack/Inbound media auth + HTML guard: keep Slack auth headers on forwarded shared attachment image downloads, and reject login/error HTML payloads (while allowing expected `.html` uploads) when resolving Slack media so auth failures do not silently pass as files. (#18642) Thanks @tumf.
 - Slack/Security ingress mismatch guard: drop slash-command and interaction payloads when app/team identifiers do not match the active Slack account context (including nested `team.id` interaction payloads), preventing cross-app or cross-workspace payload injection into system-event handling. (#29091) Thanks @Solvely-Colin.
diff --git a/src/telegram/bot/delivery.test.ts b/src/telegram/bot/delivery.test.ts
@@ -77,6 +77,12 @@ function createVoiceMessagesForbiddenError() {
   );
 }
 
+function createThreadNotFoundError(operation = "sendMessage") {
+  return new Error(
+    `GrammyError: Call to '${operation}' failed! (400: Bad Request: message thread not found)`,
+  );
+}
+
 function createVoiceFailureHarness(params: {
   voiceError: Error;
   sendMessageResult?: { message_id: number; chat: { id: string } };
@@ -225,6 +231,82 @@ describe("deliverReplies", () => {
     );
   });
 
+  it("retries DM topic sends without message_thread_id when thread is missing", async () => {
+    const runtime = createRuntime();
+    const sendMessage = vi
+      .fn()
+      .mockRejectedValueOnce(createThreadNotFoundError("sendMessage"))
+      .mockResolvedValueOnce({
+        message_id: 7,
+        chat: { id: "123" },
+      });
+    const bot = createBot({ sendMessage });
+
+    await deliverWith({
+      replies: [{ text: "hello" }],
+      runtime,
+      bot,
+      thread: { id: 42, scope: "dm" },
+    });
+
+    expect(sendMessage).toHaveBeenCalledTimes(2);
+    expect(sendMessage.mock.calls[0]?.[2]).toEqual(
+      expect.objectContaining({
+        message_thread_id: 42,
+      }),
+    );
+    expect(sendMessage.mock.calls[1]?.[2]).not.toHaveProperty("message_thread_id");
+    expect(runtime.error).not.toHaveBeenCalled();
+  });
+
+  it("does not retry forum sends without message_thread_id", async () => {
+    const runtime = createRuntime();
+    const sendMessage = vi.fn().mockRejectedValue(createThreadNotFoundError("sendMessage"));
+    const bot = createBot({ sendMessage });
+
+    await expect(
+      deliverWith({
+        replies: [{ text: "hello" }],
+        runtime,
+        bot,
+        thread: { id: 42, scope: "forum" },
+      }),
+    ).rejects.toThrow("message thread not found");
+
+    expect(sendMessage).toHaveBeenCalledTimes(1);
+    expect(runtime.error).toHaveBeenCalledTimes(1);
+  });
+
+  it("retries media sends without message_thread_id for DM topics", async () => {
+    const runtime = createRuntime();
+    const sendPhoto = vi
+      .fn()
+      .mockRejectedValueOnce(createThreadNotFoundError("sendPhoto"))
+      .mockResolvedValueOnce({
+        message_id: 8,
+        chat: { id: "123" },
+      });
+    const bot = createBot({ sendPhoto });
+
+    mockMediaLoad("photo.jpg", "image/jpeg", "image");
+
+    await deliverWith({
+      replies: [{ mediaUrl: "https://example.com/photo.jpg", text: "caption" }],
+      runtime,
+      bot,
+      thread: { id: 42, scope: "dm" },
+    });
+
+    expect(sendPhoto).toHaveBeenCalledTimes(2);
+    expect(sendPhoto.mock.calls[0]?.[2]).toEqual(
+      expect.objectContaining({
+        message_thread_id: 42,
+      }),
+    );
+    expect(sendPhoto.mock.calls[1]?.[2]).not.toHaveProperty("message_thread_id");
+    expect(runtime.error).not.toHaveBeenCalled();
+  });
+
   it("does not include link_preview_options when linkPreview is true", async () => {
     const { runtime, sendMessage, bot } = createSendMessageHarness();
 
diff --git a/src/telegram/bot/delivery.ts b/src/telegram/bot/delivery.ts
@@ -37,6 +37,7 @@ const EMPTY_TEXT_ERR_RE = /message text is empty/i;
 const VOICE_FORBIDDEN_RE = /VOICE_MESSAGES_FORBIDDEN/;
 const CAPTION_TOO_LONG_RE = /caption is too long/i;
 const FILE_TOO_BIG_RE = /file is too big/i;
+const THREAD_NOT_FOUND_RE = /message thread not found/i;
 const TELEGRAM_MEDIA_SSRF_POLICY = {
   // Telegram file downloads should trust api.telegram.org even when DNS/proxy
   // resolution maps to private/internal ranges in restricted networks.
@@ -191,24 +192,30 @@ export async function deliverReplies(params: {
         }),
       };
       if (isGif) {
-        await withTelegramApiErrorLogging({
+        await sendTelegramWithThreadFallback({
           operation: "sendAnimation",
           runtime,
-          fn: () => bot.api.sendAnimation(chatId, file, { ...mediaParams }),
+          thread,
+          requestParams: mediaParams,
+          send: (effectiveParams) => bot.api.sendAnimation(chatId, file, { ...effectiveParams }),
         });
         markDelivered();
       } else if (kind === "image") {
-        await withTelegramApiErrorLogging({
+        await sendTelegramWithThreadFallback({
           operation: "sendPhoto",
           runtime,
-          fn: () => bot.api.sendPhoto(chatId, file, { ...mediaParams }),
+          thread,
+          requestParams: mediaParams,
+          send: (effectiveParams) => bot.api.sendPhoto(chatId, file, { ...effectiveParams }),
         });
         markDelivered();
       } else if (kind === "video") {
-        await withTelegramApiErrorLogging({
+        await sendTelegramWithThreadFallback({
           operation: "sendVideo",
           runtime,
-          fn: () => bot.api.sendVideo(chatId, file, { ...mediaParams }),
+          thread,
+          requestParams: mediaParams,
+          send: (effectiveParams) => bot.api.sendVideo(chatId, file, { ...effectiveParams }),
         });
         markDelivered();
       } else if (kind === "audio") {
@@ -223,11 +230,13 @@ export async function deliverReplies(params: {
           // Switch typing indicator to record_voice before sending.
           await params.onVoiceRecording?.();
           try {
-            await withTelegramApiErrorLogging({
+            await sendTelegramWithThreadFallback({
               operation: "sendVoice",
               runtime,
+              thread,
+              requestParams: mediaParams,
               shouldLog: (err) => !isVoiceMessagesForbidden(err),
-              fn: () => bot.api.sendVoice(chatId, file, { ...mediaParams }),
+              send: (effectiveParams) => bot.api.sendVoice(chatId, file, { ...effectiveParams }),
             });
             markDelivered();
           } catch (voiceErr) {
@@ -294,18 +303,22 @@ export async function deliverReplies(params: {
           }
         } else {
           // Audio file - displays with metadata (title, duration) - DEFAULT
-          await withTelegramApiErrorLogging({
+          await sendTelegramWithThreadFallback({
             operation: "sendAudio",
             runtime,
-            fn: () => bot.api.sendAudio(chatId, file, { ...mediaParams }),
+            thread,
+            requestParams: mediaParams,
+            send: (effectiveParams) => bot.api.sendAudio(chatId, file, { ...effectiveParams }),
           });
           markDelivered();
         }
       } else {
-        await withTelegramApiErrorLogging({
+        await sendTelegramWithThreadFallback({
           operation: "sendDocument",
           runtime,
-          fn: () => bot.api.sendDocument(chatId, file, { ...mediaParams }),
+          thread,
+          requestParams: mediaParams,
+          send: (effectiveParams) => bot.api.sendDocument(chatId, file, { ...effectiveParams }),
         });
         markDelivered();
       }
@@ -559,6 +572,69 @@ async function sendTelegramVoiceFallbackText(opts: {
   }
 }
 
+function isTelegramThreadNotFoundError(err: unknown): boolean {
+  if (err instanceof GrammyError) {
+    return THREAD_NOT_FOUND_RE.test(err.description);
+  }
+  return THREAD_NOT_FOUND_RE.test(formatErrorMessage(err));
+}
+
+function hasMessageThreadIdParam(params: Record<string, unknown> | undefined): boolean {
+  if (!params) {
+    return false;
+  }
+  return typeof params.message_thread_id === "number";
+}
+
+function removeMessageThreadIdParam(
+  params: Record<string, unknown> | undefined,
+): Record<string, unknown> {
+  if (!params) {
+    return {};
+  }
+  const { message_thread_id: _ignored, ...rest } = params;
+  return rest;
+}
+
+async function sendTelegramWithThreadFallback<T>(params: {
+  operation: string;
+  runtime: RuntimeEnv;
+  thread?: TelegramThreadSpec | null;
+  requestParams: Record<string, unknown>;
+  send: (effectiveParams: Record<string, unknown>) => Promise<T>;
+  shouldLog?: (err: unknown) => boolean;
+}): Promise<T> {
+  const allowThreadlessRetry = params.thread?.scope === "dm";
+  const hasThreadId = hasMessageThreadIdParam(params.requestParams);
+  const shouldSuppressFirstErrorLog = (err: unknown) =>
+    allowThreadlessRetry && hasThreadId && isTelegramThreadNotFoundError(err);
+  const mergedShouldLog = params.shouldLog
+    ? (err: unknown) => params.shouldLog!(err) && !shouldSuppressFirstErrorLog(err)
+    : (err: unknown) => !shouldSuppressFirstErrorLog(err);
+
+  try {
+    return await withTelegramApiErrorLogging({
+      operation: params.operation,
+      runtime: params.runtime,
+      shouldLog: mergedShouldLog,
+      fn: () => params.send(params.requestParams),
+    });
+  } catch (err) {
+    if (!allowThreadlessRetry || !hasThreadId || !isTelegramThreadNotFoundError(err)) {
+      throw err;
+    }
+    const retryParams = removeMessageThreadIdParam(params.requestParams);
+    params.runtime.log?.(
+      `telegram ${params.operation}: message thread not found; retrying without message_thread_id`,
+    );
+    return await withTelegramApiErrorLogging({
+      operation: `${params.operation} (threadless retry)`,
+      runtime: params.runtime,
+      fn: () => params.send(retryParams),
+    });
+  }
+}
+
 function buildTelegramSendParams(opts?: {
   replyToMessageId?: number;
   thread?: TelegramThreadSpec | null;
@@ -601,14 +677,16 @@ async function sendTelegramText(
   const fallbackText = opts?.plainText ?? text;
   const hasFallbackText = fallbackText.trim().length > 0;
   const sendPlainFallback = async () => {
-    const res = await withTelegramApiErrorLogging({
+    const res = await sendTelegramWithThreadFallback({
       operation: "sendMessage",
       runtime,
-      fn: () =>
+      thread: opts?.thread,
+      requestParams: baseParams,
+      send: (effectiveParams) =>
         bot.api.sendMessage(chatId, fallbackText, {
           ...(linkPreviewOptions ? { link_preview_options: linkPreviewOptions } : {}),
           ...(opts?.replyMarkup ? { reply_markup: opts.replyMarkup } : {}),
-          ...baseParams,
+          ...effectiveParams,
         }),
     });
     runtime.log?.(`telegram sendMessage ok chat=${chatId} message=${res.message_id} (plain)`);
@@ -623,19 +701,21 @@ async function sendTelegramText(
     return await sendPlainFallback();
   }
   try {
-    const res = await withTelegramApiErrorLogging({
+    const res = await sendTelegramWithThreadFallback({
       operation: "sendMessage",
       runtime,
+      thread: opts?.thread,
+      requestParams: baseParams,
       shouldLog: (err) => {
         const errText = formatErrorMessage(err);
         return !PARSE_ERR_RE.test(errText) && !EMPTY_TEXT_ERR_RE.test(errText);
       },
-      fn: () =>
+      send: (effectiveParams) =>
         bot.api.sendMessage(chatId, htmlText, {
           parse_mode: "HTML",
           ...(linkPreviewOptions ? { link_preview_options: linkPreviewOptions } : {}),
           ...(opts?.replyMarkup ? { reply_markup: opts.replyMarkup } : {}),
-          ...baseParams,
+          ...effectiveParams,
         }),
     });
     runtime.log?.(`telegram sendMessage ok chat=${chatId} message=${res.message_id}`);
