negodiy/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
liuxiaopai-ai
2026-03-02 02:51:14 +08:00
committed by Peter Steinberger
parent ef9085927b
commit e6e3a7b497
3 changed files with 181 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -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.

82
src/telegram/bot/delivery.test.ts
View File

@@ -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();

116
src/telegram/bot/delivery.ts
View File

@@ -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}`);