diff --git a/extensions/telegram/src/bot-handlers.inbound-media-group.runtime.ts b/extensions/telegram/src/bot-handlers.inbound-media-group.runtime.ts index 49c9244b3907..2e333e58d0d6 100644 --- a/extensions/telegram/src/bot-handlers.inbound-media-group.runtime.ts +++ b/extensions/telegram/src/bot-handlers.inbound-media-group.runtime.ts @@ -29,6 +29,7 @@ import type { TelegramSpooledReplayDeferredParticipant } from "./bot-processing- import { MEDIA_GROUP_TIMEOUT_MS, type MediaGroupEntry } from "./bot-updates.js"; import { resolveMedia } from "./bot/delivery.resolve-media.js"; import { + buildTelegramGroupPeerId, buildTelegramThreadParams, getTelegramTextParts, hasBotMention, @@ -37,6 +38,7 @@ import { } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; import { isTelegramForumServiceMessage } from "./forum-service-message.js"; +import { resolveTelegramGroupIngestEnabled } from "./group-config-helpers.js"; import { resolveTelegramCommandIngressAuthorization } from "./ingress.js"; import type { TelegramMessageDispatchReplayClaim } from "./message-dispatch-dedupe.js"; @@ -68,6 +70,8 @@ type BufferedMediaGroupEntry = MediaGroupEntry & spooledReplayParticipants: TelegramSpooledReplayDeferredParticipant[]; }; +type TelegramGroupMediaDisposition = "process" | "skip" | "silent-ingest"; + export function createTelegramInboundMediaGroupRuntime( params: Pick< RegisterTelegramHandlerParams, @@ -114,9 +118,9 @@ export function createTelegramInboundMediaGroupRuntime( const buffer = new Map(); const queue = new KeyedAsyncQueue(); - const shouldSkipMediaDownloadForUnaddressedMentionGroup = async ( + const resolveUnaddressedGroupMediaDisposition = async ( authorization: MediaAuthorization & { ctx: TelegramContext; msg: Message }, - ): Promise => { + ): Promise => { const { ctx, msg, chatId, isGroup, isForum, resolvedThreadId, dmThreadId, senderId } = authorization; const textParts = getTelegramTextParts(msg); @@ -129,7 +133,7 @@ export function createTelegramInboundMediaGroupRuntime( // history, fires ingest hooks, and settles an explicit skipped result; // consuming them here tombstones the ingress row without any trace. if (!isGroup || !hasInboundMedia(msg) || mayNeedDownload) { - return false; + return "process"; } const sessionState = resolveTelegramSessionState({ chatId, @@ -154,12 +158,18 @@ export function createTelegramInboundMediaGroupRuntime( resolveGroupRequireMention(chatId, authorization.authorizationCfg), ); if (!requireMention) { - return false; + return "process"; } const botUsername = ctx.me?.username?.trim().toLowerCase(); const mentionRegexes = buildMentionRegexes( authorization.authorizationCfg, sessionState.agentId, + { + provider: "telegram", + conversationId: buildTelegramGroupPeerId(chatId, resolvedThreadId), + providerPolicy: + authorization.authorizationCfg.channels?.telegram?.accounts?.[accountId]?.mentionPatterns, + }, ); const hasAnyMention = textParts.entities.some((entity) => entity.type === "mention"); const explicitlyMentioned = botUsername ? hasBotMention(msg, botUsername) : false; @@ -215,10 +225,20 @@ export function createTelegramInboundMediaGroupRuntime( }, }); if (decision.shouldSkip) { + if ( + resolveTelegramGroupIngestEnabled({ + cfg: authorization.authorizationCfg, + chatId, + accountId, + topicConfig: authorization.topicConfig, + }) + ) { + return "silent-ingest"; + } logger.info({ chatId, reason: "no-mention" }, "skipping group media before download"); - return true; + return "skip"; } - return false; + return "process"; }; const processMediaGroup = async (entry: BufferedMediaGroupEntry) => { @@ -275,7 +295,11 @@ export function createTelegramInboundMediaGroupRuntime( }); primary = { ctx: combinedContext, msg: combinedMessage }; } - if (await shouldSkipMediaDownloadForUnaddressedMentionGroup({ ...entry, ...primary })) { + const mediaDisposition = await resolveUnaddressedGroupMediaDisposition({ + ...entry, + ...primary, + }); + if (mediaDisposition === "skip") { releaseDispatchDedupeClaims(entry.dispatchDedupeClaims); settleSpooledReplayParticipants(entry.spooledReplayParticipants, { kind: "skipped" }); return; @@ -324,7 +348,7 @@ export function createTelegramInboundMediaGroupRuntime( skippedCount++; } } - if (skippedCount > 0) { + if (skippedCount > 0 && mediaDisposition !== "silent-ingest") { const verb = skippedCount === 1 ? "was" : "were"; await withTelegramApiErrorLogging({ operation: "sendMessage", @@ -432,5 +456,5 @@ export function createTelegramInboundMediaGroupRuntime( return true; }; - return { handleMediaGroup, shouldSkipMediaDownloadForUnaddressedMentionGroup }; + return { handleMediaGroup, resolveUnaddressedGroupMediaDisposition }; } diff --git a/extensions/telegram/src/bot-handlers.inbound.runtime.ts b/extensions/telegram/src/bot-handlers.inbound.runtime.ts index c4fb54ce2cc1..32aa26a4177d 100644 --- a/extensions/telegram/src/bot-handlers.inbound.runtime.ts +++ b/extensions/telegram/src/bot-handlers.inbound.runtime.ts @@ -69,7 +69,7 @@ export function createTelegramHandlerInboundRuntime( resolveTelegramDebounceLane, } = createTelegramInboundDebounceRuntime({ cfg, bot, runtime }, messageRuntime); - const { handleMediaGroup, shouldSkipMediaDownloadForUnaddressedMentionGroup } = + const { handleMediaGroup, resolveUnaddressedGroupMediaDisposition } = createTelegramInboundMediaGroupRuntime( { accountId, @@ -204,23 +204,22 @@ export function createTelegramHandlerInboundRuntime( return; } - if ( - await shouldSkipMediaDownloadForUnaddressedMentionGroup({ - authorizationCfg, - ctx, - msg, - chatId, - isGroup, - isForum, - resolvedThreadId, - dmThreadId, - senderId, - effectiveGroupAllow, - effectiveDmAllow, - groupConfig, - topicConfig, - }) - ) { + const mediaDisposition = await resolveUnaddressedGroupMediaDisposition({ + authorizationCfg, + ctx, + msg, + chatId, + isGroup, + isForum, + resolvedThreadId, + dmThreadId, + senderId, + effectiveGroupAllow, + effectiveDmAllow, + groupConfig, + topicConfig, + }); + if (mediaDisposition === "skip") { releaseDispatchDedupeClaims(dispatchDedupeClaims); return; } @@ -254,7 +253,7 @@ export function createTelegramHandlerInboundRuntime( return; } if (isMediaSizeLimitError(mediaErr)) { - if (sendOversizeWarning) { + if (sendOversizeWarning && mediaDisposition !== "silent-ingest") { const limitMb = mediaErr instanceof TelegramBotApiFileTooLargeError ? Math.min(mediaErr.limitMb, Math.round(mediaMaxBytes / (1024 * 1024))) @@ -281,18 +280,20 @@ export function createTelegramHandlerInboundRuntime( releaseDispatchDedupeClaims(dispatchDedupeClaims, mediaErr); return; } - await withTelegramApiErrorLogging({ - operation: "sendMessage", - runtime, - fn: () => - bot.api.sendMessage(chatId, "⚠️ Failed to download media. Please try again.", { - ...warningThreadParams, - reply_parameters: { - message_id: msg.message_id, - allow_sending_without_reply: true, - }, - }), - }).catch(() => {}); + if (mediaDisposition !== "silent-ingest") { + await withTelegramApiErrorLogging({ + operation: "sendMessage", + runtime, + fn: () => + bot.api.sendMessage(chatId, "⚠️ Failed to download media. Please try again.", { + ...warningThreadParams, + reply_parameters: { + message_id: msg.message_id, + allow_sending_without_reply: true, + }, + }), + }).catch(() => {}); + } } } diff --git a/extensions/telegram/src/bot-message-context.body.ts b/extensions/telegram/src/bot-message-context.body.ts index b2b1f4763dc6..1d3d11d81f33 100644 --- a/extensions/telegram/src/bot-message-context.body.ts +++ b/extensions/telegram/src/bot-message-context.body.ts @@ -14,7 +14,6 @@ import { type InboundEventKind, type NormalizedLocation, } from "openclaw/plugin-sdk/channel-inbound"; -import { resolveChannelGroupPolicy } from "openclaw/plugin-sdk/channel-policy"; import { hasControlCommand } from "openclaw/plugin-sdk/command-detection"; import { isAbortRequestText } from "openclaw/plugin-sdk/command-primitives-runtime"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; @@ -55,6 +54,7 @@ import { import { buildTelegramGroupPeerId, buildTelegramInboundOriginTarget } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; import { isTelegramForumServiceMessage } from "./forum-service-message.js"; +import { resolveTelegramGroupIngestEnabled } from "./group-config-helpers.js"; import { recordTelegramGroupHistoryEntry } from "./group-history-window.js"; import { resolveTelegramCommandIngressAuthorization } from "./ingress.js"; type TelegramMentionFacts = NonNullable< @@ -388,17 +388,7 @@ export async function resolveTelegramInboundBody(params: { messageId: typeof msg.message_id === "number" ? String(msg.message_id) : undefined, }, }); - const telegramGroupPolicy = resolveChannelGroupPolicy({ - cfg, - channel: "telegram", - groupId: String(chatId), - accountId, - }); - const ingestEnabled = - topicConfig?.ingest ?? - telegramGroupPolicy.groupConfig?.ingest ?? - telegramGroupPolicy.defaultConfig?.ingest; - if (ingestEnabled === true && sessionKey) { + if (sessionKey && resolveTelegramGroupIngestEnabled({ cfg, chatId, accountId, topicConfig })) { fireAndForgetHook( triggerInternalHook( createInternalHookEvent( @@ -408,7 +398,7 @@ export async function resolveTelegramInboundBody(params: { toInternalMessageReceivedContext({ from: `telegram:group:${historyKey ?? chatId}`, to: originatingTo, - content: rawBody, + content: historyBody, timestamp: msg.date ? msg.date * 1000 : undefined, channelId: "telegram", accountId, @@ -424,6 +414,12 @@ export async function resolveTelegramInboundBody(params: { originatingTo, isGroup: true, groupId: `telegram:${chatId}`, + media: materializedMedia.map(({ path, contentType, kind, sourceMessageId }) => ({ + path, + contentType, + kind, + messageId: sourceMessageId ?? String(msg.message_id), + })), }), ), ), diff --git a/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts b/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts index c229bcd0a061..bd94e9165334 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.channel-post-media.test.ts @@ -1,13 +1,33 @@ // Telegram tests cover bot.create telegram bot.channel post media plugin behavior. import { setTimeout as delay } from "node:timers/promises"; +import { + createPluginStateKeyedStoreForTests, + createPluginStateSyncKeyedStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { telegramBotInfoForTest } from "./bot.create-telegram-bot.test-support.js"; +import { resetTelegramForumFlagCacheForTest } from "./bot/helpers.js"; +import { setTelegramRuntime } from "./runtime.js"; +import type { TelegramRuntime } from "./runtime.types.js"; const saveRemoteMedia = vi.fn(); const saveMediaBuffer = vi.fn(); const readRemoteMediaBuffer = vi.fn(); const rootRead = vi.fn(); +const { triggerInternalHookMock } = vi.hoisted(() => ({ + triggerInternalHookMock: vi.fn<(event: unknown) => Promise>(async () => undefined), +})); + +vi.mock("openclaw/plugin-sdk/hook-runtime", async () => { + const actual = await vi.importActual( + "openclaw/plugin-sdk/hook-runtime", + ); + return { + ...actual, + triggerInternalHook: triggerInternalHookMock, + }; +}); vi.mock("openclaw/plugin-sdk/file-access-runtime", () => ({ root: async (rootDir: string) => ({ @@ -76,6 +96,7 @@ const TELEGRAM_TEST_TIMINGS = { textFragmentGapMs: 30, } as const; const TEXT_FRAGMENT_COALESCE_TEST_GAP_MS = 5_000; +const TELEGRAM_TEST_TOPIC = "-100456:topic:42"; async function withTelegramSpooledReplayUpdate( update: object, @@ -89,12 +110,7 @@ function setOpenChannelPostConfig() { channels: { telegram: { groupPolicy: "open", - groups: { - "-100777111222": { - enabled: true, - requireMention: false, - }, - }, + groups: { "-100777111222": { enabled: true, requireMention: false } }, }, }, }); @@ -107,10 +123,6 @@ function getChannelPostHandler( return getOnHandler("channel_post") as (ctx: Record) => Promise; } -function resolveFlushTimer(setTimeoutSpy: ReturnType) { - return resolveFlushTimerForDelay(setTimeoutSpy, TELEGRAM_TEST_TIMINGS.mediaGroupFlushMs); -} - function resolveFlushTimerForDelay(setTimeoutSpy: ReturnType, delayMs: number) { const flushTimerCallIndex = setTimeoutSpy.mock.calls.findLastIndex( (call: Parameters) => call[1] === delayMs, @@ -137,10 +149,6 @@ function createImageFetchSpy(params?: { body?: Uint8Array; contentType?: string ); } -async function waitForBufferedProcessing() { - await delay(75); -} - async function waitForMockCalls(mock: { mock: { calls: unknown[] } }, count: number) { for (let index = 0; index < 80; index++) { if (mock.mock.calls.length >= count) { @@ -178,10 +186,13 @@ function createChannelPostContext(params: { } async function flushChannelPostMediaGroup(setTimeoutSpy: ReturnType) { - const flushTimer = resolveFlushTimer(setTimeoutSpy); + const flushTimer = resolveFlushTimerForDelay( + setTimeoutSpy, + TELEGRAM_TEST_TIMINGS.mediaGroupFlushMs, + ); expect(flushTimer).toBeTypeOf("function"); await flushTimer?.(); - await waitForBufferedProcessing(); + await delay(75); } async function flushChannelPostMediaGroupForDelay( @@ -191,7 +202,7 @@ async function flushChannelPostMediaGroupForDelay( const flushTimer = resolveFlushTimerForDelay(setTimeoutSpy, delayMs); expect(flushTimer).toBeTypeOf("function"); await flushTimer?.(); - await waitForBufferedProcessing(); + await delay(75); } async function queueChannelPostAlbum( @@ -206,25 +217,24 @@ async function queueChannelPostAlbum( secondGetFileResult?: Record; }, ) { - const first = handler( - createChannelPostContext({ - messageId: params.firstMessageId, - caption: params.caption, - date: 1736380800, - mediaGroupId: params.mediaGroupId, - photoFileId: params.firstPhotoFileId ?? "p1", - }), + await Promise.all( + [ + { + messageId: params.firstMessageId, + caption: params.caption, + date: 1736380800, + photoFileId: params.firstPhotoFileId ?? "p1", + }, + { + messageId: params.secondMessageId, + date: 1736380801, + photoFileId: params.secondPhotoFileId ?? "p2", + getFileResult: params.secondGetFileResult, + }, + ].map((message) => + handler(createChannelPostContext({ ...message, mediaGroupId: params.mediaGroupId })), + ), ); - const second = handler( - createChannelPostContext({ - messageId: params.secondMessageId, - date: 1736380801, - mediaGroupId: params.mediaGroupId, - photoFileId: params.secondPhotoFileId ?? "p2", - getFileResult: params.secondGetFileResult, - }), - ); - await Promise.all([first, second]); } function replyPayload(): Record { @@ -247,17 +257,218 @@ function expectTypeOnlyMediaPayload(kind: string, rawBody = "") { expect(media[0]?.path).toBeUndefined(); } +type TelegramMentionPolicyForTest = { + mode: "allow" | "deny"; + allowIn?: string[]; + denyIn?: string[]; +}; + +type TelegramIngestGroupForTest = { + requireMention: boolean; + ingest?: boolean; + topics?: Record; +}; + +function telegramIngestGroupForTest( + ingest?: boolean, + topics?: Record, +): TelegramIngestGroupForTest { + return { + requireMention: true, + ...(ingest === undefined ? {} : { ingest }), + ...(topics ? { topics } : {}), + }; +} + +type TelegramMentionCaseForTest = [ + string, + TelegramMentionPolicyForTest, + TelegramMentionPolicyForTest | undefined, + number | undefined, + boolean, + number, +]; + +function setTelegramIngestGroupConfig( + params: { + groups?: Record; + providerPolicy?: TelegramMentionPolicyForTest; + accountPolicy?: TelegramMentionPolicyForTest; + customMentionPatterns?: boolean; + } = {}, +) { + loadConfig.mockReturnValue({ + ...(params.customMentionPatterns + ? { messages: { groupChat: { mentionPatterns: ["\\bbert\\b"] } } } + : {}), + channels: { + telegram: { + groupPolicy: "open", + ...(params.providerPolicy ? { mentionPatterns: params.providerPolicy } : {}), + groups: params.groups ?? { "-100456": { requireMention: true, ingest: true } }, + ...(params.accountPolicy + ? { accounts: { work: { mentionPatterns: params.accountPolicy } } } + : {}), + }, + }, + }); +} + +async function dispatchTelegramGroupPhoto(params: { + messageId: number; + topicId?: number; + albumId?: string; + caption?: string; + extraMessage?: Record; + getFile?: () => Promise<{ file_path: string }>; +}) { + const handler = getOnHandler("message") as (ctx: Record) => Promise; + await handler({ + message: { + chat: { + id: -100456, + type: "supergroup", + title: "Ops Chat", + ...(params.topicId ? { is_forum: true } : {}), + }, + message_id: params.messageId, + date: 1736380800, + ...(params.topicId ? { message_thread_id: params.topicId, is_topic_message: true } : {}), + ...(params.albumId ? { media_group_id: params.albumId } : {}), + ...(params.caption ? { caption: params.caption } : {}), + ...params.extraMessage, + photo: [{ file_id: `photo-${params.messageId}` }], + from: { id: 55, is_bot: false, first_name: "u" }, + }, + me: { id: 999, username: "openclaw_bot" }, + getFile: params.getFile ?? (async () => ({ file_path: `photos/${params.messageId}.jpg` })), + }); +} + +function expectTelegramIngestHook( + messageIds: number[], + params: { content?: string; expectedCalls?: number } = {}, +) { + const expectedCalls = params.expectedCalls ?? 1; + const event = triggerInternalHookMock.mock.calls[0]?.[0] as + | { type: string; action: string; context: { content: string; media?: unknown[] } } + | undefined; + expect(triggerInternalHookMock).toHaveBeenCalledTimes(expectedCalls); + expect(event?.type).toEqual(expectedCalls ? "message" : undefined); + expect(event?.action).toEqual(expectedCalls ? "received" : undefined); + expect(event?.context.content).toEqual( + expectedCalls ? (params.content ?? expect.stringMatching(/\S/u)) : undefined, + ); + expect(event?.context.media).toEqual( + expectedCalls && messageIds.length + ? messageIds.map((messageId) => + expect.objectContaining({ + path: "/tmp/telegram-media.bin", + contentType: "image/png", + kind: "image", + messageId: String(messageId), + }), + ) + : undefined, + ); +} + +function setOpenTelegramDirectConfig(mediaMaxMb?: number) { + loadConfig.mockReturnValue({ + channels: { + telegram: { + dmPolicy: "open", + allowFrom: ["*"], + ...(mediaMaxMb === undefined ? {} : { mediaMaxMb }), + }, + }, + }); +} + +function createTelegramPrivateMediaContext(params: { + messageId: number; + fileId: string; + fileName?: string; + update?: { update_id: number }; + getFile?: () => Promise<{ file_path: string }>; +}) { + return { + ...(params.update ? { update: params.update } : {}), + message: { + chat: { id: 1234, type: "private" }, + message_id: params.messageId, + date: 1736380800, + ...(params.fileName + ? { document: { file_id: params.fileId, file_name: params.fileName } } + : { photo: [{ file_id: params.fileId }] }), + from: { id: 55, is_bot: false, first_name: "u" }, + }, + me: { username: "openclaw_bot" }, + getFile: params.getFile ?? (async () => ({ file_path: `documents/${params.fileId}` })), + }; +} + +function expectTelegramDownloadWarning(messageId: number, warning?: string) { + expect(sendMessageSpy).toHaveBeenCalledWith( + 1234, + warning ?? "⚠️ Failed to download media. Please try again.", + expect.objectContaining({ + reply_parameters: expect.objectContaining({ + message_id: messageId, + allow_sending_without_reply: true, + }), + }), + ); +} + +function rejectFirstTelegramAlbumDownloadWhen(partial: boolean) { + if (partial) { + saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: Failed to fetch media")); + } +} + +async function rejectTelegramAlbumDownload(shutdown: AbortController, abort: boolean) { + if (abort) { + shutdown.abort(); + } + throw abort + ? Object.assign(new Error("aborted"), { name: "AbortError" }) + : new Error("MediaFetchError: Failed to fetch media"); +} + describe("createTelegramBot channel_post media", () => { beforeAll(() => { createTelegramBot = (opts) => createTelegramBotBase({ botInfo: telegramBotInfoForTest, + telegramTransport: { + fetch: globalThis.fetch, + sourceFetch: globalThis.fetch, + close: async () => {}, + }, ...opts, telegramDeps: telegramBotDepsForTest, }); }); beforeEach(() => { + resetTelegramForumFlagCacheForTest(); + setTelegramRuntime({ + state: { + openKeyedStore: ((options) => + createPluginStateKeyedStoreForTests( + "telegram", + options, + )) as TelegramRuntime["state"]["openKeyedStore"], + openSyncKeyedStore: ((options) => + createPluginStateSyncKeyedStoreForTests( + "telegram", + options, + )) as TelegramRuntime["state"]["openSyncKeyedStore"], + }, + channel: {}, + } as TelegramRuntime); + triggerInternalHookMock.mockClear(); saveRemoteMedia.mockReset(); saveRemoteMedia.mockImplementation( async (params: { fetchImpl: typeof fetch; maxBytes: number; url: string }) => { @@ -376,45 +587,15 @@ describe("createTelegramBot channel_post media", () => { }); it("notifies users when media download fails for direct messages", async () => { - loadConfig.mockReturnValue({ - channels: { - telegram: { dmPolicy: "open", allowFrom: ["*"] }, - }, - }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); + setOpenTelegramDirectConfig(); saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: Failed to fetch media")); - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - throw new Error("MediaFetchError: Failed to fetch media"); - }); - + const fetchSpy = vi.spyOn(globalThis, "fetch").mockRejectedValue(new Error("fetch failed")); try { createTelegramBot({ token: "tok" }); const handler = getOnHandler("message") as (ctx: Record) => Promise; - - await handler({ - message: { - chat: { id: 1234, type: "private" }, - message_id: 411, - date: 1736380800, - photo: [{ file_id: "p1" }], - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => ({ file_path: "photos/p1.jpg" }), - }); + await handler(createTelegramPrivateMediaContext({ messageId: 411, fileId: "p1" })); await waitForMockCalls(sendMessageSpy, 1); - - expect(sendMessageSpy).toHaveBeenCalledWith( - 1234, - "⚠️ Failed to download media. Please try again.", - { - reply_parameters: { - message_id: 411, - allow_sending_without_reply: true, - }, - }, - ); + expectTelegramDownloadWarning(411); expect(replySpy).toHaveBeenCalledOnce(); expectTypeOnlyMediaPayload("image"); } finally { @@ -423,37 +604,21 @@ describe("createTelegramBot channel_post media", () => { }); it("warns and dispatches a type-only fact when Telegram getFile fails (#100000)", async () => { - loadConfig.mockReturnValue({ - channels: { telegram: { dmPolicy: "open", allowFrom: ["*"] } }, - }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - + setOpenTelegramDirectConfig(); createTelegramBot({ token: "tok" }); const handler = getOnHandler("message") as (ctx: Record) => Promise; - - await handler({ - message: { - chat: { id: 1234, type: "private" }, - message_id: 100000, - date: 1736380800, - document: { file_id: "doc-100000", file_name: "report.pdf" }, - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => { - throw new Error("Network request for 'getFile' failed!"); - }, - }); - - await waitForMockCalls(sendMessageSpy, 1); - expect(sendMessageSpy).toHaveBeenCalledWith( - 1234, - "⚠️ Failed to download media. Please try again.", - expect.objectContaining({ - reply_parameters: expect.objectContaining({ message_id: 100000 }), + await handler( + createTelegramPrivateMediaContext({ + messageId: 100000, + fileId: "doc-100000", + fileName: "report.pdf", + getFile: async () => { + throw new Error("Network request for 'getFile' failed!"); + }, }), ); + await waitForMockCalls(sendMessageSpy, 1); + expectTelegramDownloadWarning(100000); expect(replySpy).toHaveBeenCalledOnce(); expectTypeOnlyMediaPayload("document"); expect(saveRemoteMedia).not.toHaveBeenCalled(); @@ -465,398 +630,304 @@ describe("createTelegramBot channel_post media", () => { ])( "reports the effective $expectedLimitMb MB limit for Telegram Bot API failures (#100000)", async ({ mediaMaxMb, expectedLimitMb }) => { - loadConfig.mockReturnValue({ - channels: { telegram: { dmPolicy: "open", allowFrom: ["*"], mediaMaxMb } }, - }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - + setOpenTelegramDirectConfig(mediaMaxMb); createTelegramBot({ token: "tok" }); const handler = getOnHandler("message") as (ctx: Record) => Promise; - const messageId = 100001 + expectedLimitMb; - await handler({ - message: { - chat: { id: 1234, type: "private" }, - message_id: messageId, - date: 1736380800, - document: { file_id: "doc-100001", file_name: "large.bin" }, - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => { - throw new Error("Bad Request: file is too big"); - }, - }); - - await waitForMockCalls(sendMessageSpy, 1); - expect(sendMessageSpy).toHaveBeenCalledWith( - 1234, - `⚠️ File too large. Maximum size is ${expectedLimitMb}MB.`, - expect.objectContaining({ - reply_parameters: expect.objectContaining({ message_id: messageId }), + await handler( + createTelegramPrivateMediaContext({ + messageId, + fileId: "doc-100001", + fileName: "large.bin", + getFile: async () => { + throw new Error("Bad Request: file is too big"); + }, }), ); + await waitForMockCalls(sendMessageSpy, 1); + expectTelegramDownloadWarning( + messageId, + `⚠️ File too large. Maximum size is ${expectedLimitMb}MB.`, + ); expect(replySpy).toHaveBeenCalledOnce(); expectTypeOnlyMediaPayload("document"); expect(saveRemoteMedia).not.toHaveBeenCalled(); }, ); - it("durably retries a spooled-replay shutdown-abort document fetch without warning (#98076)", async () => { - loadConfig.mockReturnValue({ - channels: { telegram: { dmPolicy: "open", allowFrom: ["*"] } }, - }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - saveRemoteMedia.mockRejectedValue(Object.assign(new Error("aborted"), { name: "AbortError" })); - + it.each([ + { + name: "retryable shutdown abort", + messageId: 98076, + error: Object.assign(new Error("aborted"), { name: "AbortError" }), + result: { kind: "failed-retryable", error: expect.any(MediaFetchError) }, + warnings: 0, + }, + { + name: "permanent oversized media", + messageId: 98077, + error: new MediaFetchError("max_bytes", "Failed to fetch media: payload exceeds maxBytes 10"), + result: { kind: "completed" }, + warnings: 1, + }, + { + name: "permanent SSRF rejection", + messageId: 98078, + error: new Error("blocked by SSRF guard: private address"), + result: { kind: "completed" }, + warnings: 1, + }, + ])("preserves durable replay handling for $name (#98076)", async (testCase) => { + setOpenTelegramDirectConfig(); + saveRemoteMedia.mockRejectedValue(testCase.error); createTelegramBot({ token: "tok" }); const handler = getOnHandler("message") as (ctx: Record) => Promise; - const update = { update_id: 98076 }; - const ctx = { + const update = { update_id: testCase.messageId }; + const ctx = createTelegramPrivateMediaContext({ + messageId: testCase.messageId, + fileId: `doc-${testCase.messageId}`, + fileName: "document.pdf", update, - message: { - chat: { id: 1234, type: "private" }, - message_id: 98076, - date: 1736380800, - document: { file_id: "doc-1", file_name: "report.pdf" }, - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => ({ file_path: "documents/doc-1" }), - }; - + }); const { result } = await runWithTelegramUpdateProcessingFrame(() => withTelegramSpooledReplayUpdate(update, () => handler(ctx)), ); + expect(result).toEqual(testCase.result); + expect(sendMessageSpy).toHaveBeenCalledTimes(testCase.warnings); + expect(replySpy).toHaveBeenCalledTimes(testCase.warnings); + expect(sendMessageSpy.mock.calls[0]?.[1]).toEqual( + [undefined, "⚠️ Failed to download media. Please try again."][testCase.warnings], + ); + if (testCase.warnings) { + expectTelegramDownloadWarning(testCase.messageId); + expectTypeOnlyMediaPayload("document"); + } + }); - expect(result).toEqual({ kind: "failed-retryable", error: expect.any(MediaFetchError) }); + it.each([ + ["default disabled", undefined, undefined, undefined, false], + ["enabled group", true, undefined, undefined, true], + ["wildcard inherited", undefined, true, undefined, true], + ["group disables wildcard", false, true, undefined, false], + ["topic enables group", false, undefined, true, true], + ["topic disables group", true, undefined, false, false], + ] as Array<[string, boolean | undefined, boolean | undefined, boolean | undefined, boolean]>)( + "honors %s before skipping unmentioned group media (#92067)", + async (_name, groupIngest, wildcardIngest, topicIngest, shouldIngest) => { + const topics = topicIngest === undefined ? undefined : { "42": { ingest: topicIngest } }; + const groups = { + ...(wildcardIngest === undefined + ? {} + : { "*": telegramIngestGroupForTest(wildcardIngest) }), + "-100456": telegramIngestGroupForTest(groupIngest, topics), + }; + setTelegramIngestGroupConfig({ groups }); + const getFile = vi.fn(async () => ({ file_path: "photos/ingested.jpg" })); + const fetchSpy = createImageFetchSpy(); + try { + createTelegramBot({ token: "tok" }); + await dispatchTelegramGroupPhoto({ + messageId: 92067, + topicId: topicIngest === undefined ? undefined : 42, + getFile, + }); + const expectedCalls = Number(shouldIngest); + expect(getFile).toHaveBeenCalledTimes(expectedCalls); + expect(fetchSpy).toHaveBeenCalledTimes(expectedCalls); + expectTelegramIngestHook([92067], { expectedCalls }); + expect(sendMessageSpy).not.toHaveBeenCalled(); + expect(replySpy).not.toHaveBeenCalled(); + } finally { + fetchSpy.mockRestore(); + } + }, + ); + + it.each([ + { failure: "a download error", error: "Network request for 'getFile' failed!" }, + { failure: "an oversized file", error: "Bad Request: file is too big" }, + ])("silently ingests unmentioned group media after $failure (#92067)", async ({ error }) => { + setTelegramIngestGroupConfig(); + createTelegramBot({ token: "tok" }); + await dispatchTelegramGroupPhoto({ + messageId: 92070, + getFile: async () => { + throw new Error(error); + }, + }); expect(sendMessageSpy).not.toHaveBeenCalled(); + expect(replySpy).not.toHaveBeenCalled(); + expect(saveRemoteMedia).not.toHaveBeenCalled(); + expectTelegramIngestHook([]); }); - it("acks and warns a permanent media failure even on spooled replay (#98076)", async () => { - loadConfig.mockReturnValue({ - channels: { telegram: { dmPolicy: "open", allowFrom: ["*"] } }, + it.each([ + { name: "all", messageIds: [92068, 92069], partial: false, deniedMention: false }, + { name: "partial", messageIds: [92071, 92072], partial: true, deniedMention: false }, + { name: "denied mention", messageIds: [92071, 92072], partial: true, deniedMention: true }, + ])("silently ingests group media albums with $name exactly once (#92067)", async (testCase) => { + setTelegramIngestGroupConfig({ + customMentionPatterns: testCase.deniedMention, + ...(testCase.deniedMention ? { providerPolicy: { mode: "deny" } } : {}), }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - saveRemoteMedia.mockRejectedValue( - new MediaFetchError("max_bytes", "Failed to fetch media: payload exceeds maxBytes 10"), - ); - - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - const update = { update_id: 98077 }; - const ctx = { - update, - message: { - chat: { id: 1234, type: "private" }, - message_id: 98077, - date: 1736380800, - document: { file_id: "doc-2", file_name: "huge.pdf" }, - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => ({ file_path: "documents/doc-2" }), - }; - - const { result } = await runWithTelegramUpdateProcessingFrame(() => - withTelegramSpooledReplayUpdate(update, () => handler(ctx)), - ); - - expect(result).toEqual({ kind: "completed" }); - await waitForMockCalls(sendMessageSpy, 1); - expect(sendMessageSpy).toHaveBeenCalledWith( - 1234, - "⚠️ Failed to download media. Please try again.", - expect.objectContaining({ - reply_parameters: expect.objectContaining({ message_id: 98077 }), - }), - ); - expect(replySpy).toHaveBeenCalledOnce(); - expectTypeOnlyMediaPayload("document"); - }); - - it("acks and warns a permanent fetch_failed (guard/SSRF) on spooled replay (#98076)", async () => { - loadConfig.mockReturnValue({ - channels: { telegram: { dmPolicy: "open", allowFrom: ["*"] } }, - }); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - saveRemoteMedia.mockRejectedValue(new Error("blocked by SSRF guard: private address")); - - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - const update = { update_id: 98078 }; - const ctx = { - update, - message: { - chat: { id: 1234, type: "private" }, - message_id: 98078, - date: 1736380800, - document: { file_id: "doc-3", file_name: "blocked.pdf" }, - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { username: "openclaw_bot" }, - getFile: async () => ({ file_path: "documents/doc-3" }), - }; - - const { result } = await runWithTelegramUpdateProcessingFrame(() => - withTelegramSpooledReplayUpdate(update, () => handler(ctx)), - ); - - expect(result).toEqual({ kind: "completed" }); - await waitForMockCalls(sendMessageSpy, 1); - expect(sendMessageSpy).toHaveBeenCalledWith( - 1234, - "⚠️ Failed to download media. Please try again.", - expect.objectContaining({ - reply_parameters: expect.objectContaining({ message_id: 98078 }), - }), - ); - expect(replySpy).toHaveBeenCalledOnce(); - expectTypeOnlyMediaPayload("document"); - }); - - it("skips unmentioned requireMention group media before downloading (#81181)", async () => { - loadConfig.mockReturnValue({ - channels: { - telegram: { - groupPolicy: "open", - groups: { "*": { requireMention: true } }, - }, - }, - }); - const getFile = vi.fn(async () => ({ file_path: "photos/p1.jpg" })); - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - throw new Error("unexpected media download"); - }); - + rejectFirstTelegramAlbumDownloadWhen(testCase.partial); + const fetchSpy = createImageFetchSpy(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const getFile = vi.fn(async () => ({ file_path: "photos/ingested-album.jpg" })); try { - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - - await handler({ - message: { - chat: { id: -100456, type: "supergroup", title: "Ops Chat" }, - message_id: 81181, - date: 1736380800, - photo: [{ file_id: "p1" }], - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { id: 999, username: "openclaw_bot" }, - getFile, - }); - + createTelegramBot({ token: "tok", testTimings: TELEGRAM_TEST_TIMINGS }); + for (const messageId of testCase.messageIds) { + await dispatchTelegramGroupPhoto({ + messageId, + albumId: "ingested-album", + ...(testCase.deniedMention && messageId === testCase.messageIds[0] + ? { caption: "bert, see attachment" } + : {}), + getFile, + }); + } expect(getFile).not.toHaveBeenCalled(); - expect(fetchSpy).not.toHaveBeenCalled(); + await flushChannelPostMediaGroup(setTimeoutSpy); + expect(getFile).toHaveBeenCalledTimes(2); + expect(fetchSpy).toHaveBeenCalledTimes(testCase.partial ? 1 : 2); + expectTelegramIngestHook( + testCase.partial ? testCase.messageIds.slice(1) : testCase.messageIds, + ); expect(sendMessageSpy).not.toHaveBeenCalled(); expect(replySpy).not.toHaveBeenCalled(); - } finally { - fetchSpy.mockRestore(); - } - }); - - it("notifies mentioned requireMention groups when media download fails", async () => { - loadConfig.mockReturnValue({ - channels: { - telegram: { - groupPolicy: "open", - groups: { "*": { requireMention: true } }, - }, - }, - }); - saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: ECONNRESET")); - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - throw new Error("MediaFetchError: ECONNRESET"); - }); - - try { - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - - await handler({ - message: { - chat: { id: -100456, type: "supergroup", title: "Ops Chat" }, - message_id: 81182, - date: 1736380800, - caption: "@openclaw_bot check this", - photo: [{ file_id: "p1" }], - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { id: 999, username: "openclaw_bot" }, - getFile: async () => ({ file_path: "photos/p1.jpg" }), - }); - await waitForMockCalls(sendMessageSpy, 1); - - expect(sendMessageSpy).toHaveBeenCalledWith( - -100456, - "⚠️ Failed to download media. Please try again.", - { - reply_parameters: { - message_id: 81182, - allow_sending_without_reply: true, - }, - }, - ); - expect(replySpy).toHaveBeenCalledOnce(); - expectTypeOnlyMediaPayload("image", "@openclaw_bot check this"); - } finally { - fetchSpy.mockRestore(); - } - }); - - it("treats targeted bot command captions as mentions before media download", async () => { - loadConfig.mockReturnValue({ - channels: { - telegram: { - groupPolicy: "open", - groups: { "*": { requireMention: true } }, - }, - }, - }); - saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: ECONNRESET")); - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - throw new Error("MediaFetchError: ECONNRESET"); - }); - - try { - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - const caption = "/inspect@openclaw_bot"; - - await handler({ - message: { - chat: { id: -100456, type: "supergroup", title: "Ops Chat" }, - message_id: 81184, - date: 1736380800, - caption, - caption_entities: [{ type: "bot_command", offset: 0, length: caption.length }], - photo: [{ file_id: "p1" }], - from: { id: 55, is_bot: false, first_name: "u" }, - }, - me: { id: 999, username: "openclaw_bot" }, - getFile: async () => ({ file_path: "photos/p1.jpg" }), - }); - await waitForMockCalls(sendMessageSpy, 1); - - expect(sendMessageSpy).toHaveBeenCalledWith( - -100456, - "⚠️ Failed to download media. Please try again.", - { - reply_parameters: { - message_id: 81184, - allow_sending_without_reply: true, - }, - }, - ); - expect(replySpy).toHaveBeenCalledOnce(); - expectTypeOnlyMediaPayload("image", caption); - } finally { - fetchSpy.mockRestore(); - } - }); - - it("notifies requireMention group replies to the bot when media download fails", async () => { - loadConfig.mockReturnValue({ - channels: { - telegram: { - groupPolicy: "open", - groups: { "*": { requireMention: true } }, - }, - }, - }); - saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: ECONNRESET")); - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - throw new Error("MediaFetchError: ECONNRESET"); - }); - - try { - createTelegramBot({ token: "tok" }); - const handler = getOnHandler("message") as (ctx: Record) => Promise; - - await handler({ - message: { - chat: { id: -100456, type: "supergroup", title: "Ops Chat" }, - message_id: 81183, - date: 1736380800, - photo: [{ file_id: "p1" }], - from: { id: 55, is_bot: false, first_name: "u" }, - reply_to_message: { - message_id: 99, - text: "previous bot reply", - from: { id: 999, is_bot: true, first_name: "OpenClaw" }, - }, - }, - me: { id: 999, username: "openclaw_bot" }, - getFile: async () => ({ file_path: "photos/p1.jpg" }), - }); - await waitForMockCalls(sendMessageSpy, 1); - - expect(sendMessageSpy).toHaveBeenCalledWith( - -100456, - "⚠️ Failed to download media. Please try again.", - { - reply_parameters: { - message_id: 81183, - allow_sending_without_reply: true, - }, - }, - ); - expect(replySpy).toHaveBeenCalledOnce(); - expectTypeOnlyMediaPayload("image"); - } finally { - fetchSpy.mockRestore(); - } - }); - - it("processes remaining media group photos when one photo download fails", async () => { - replySpy.mockReset(); - setOpenChannelPostConfig(); - - let fetchCallIndex = 0; - const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - fetchCallIndex++; - if (fetchCallIndex === 2) { - throw new Error("MediaFetchError: Failed to fetch media"); - } - return new Response(new Uint8Array([0x89, 0x50, 0x4e, 0x47]), { - status: 200, - headers: { "content-type": "image/png" }, - }); - }); - - const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); - try { - const handler = getChannelPostHandler(); - await queueChannelPostAlbum(handler, { - caption: "partial album", - mediaGroupId: "partial-album-1", - firstMessageId: 401, - secondMessageId: 402, - }); - expect(replySpy).not.toHaveBeenCalled(); - await flushChannelPostMediaGroup(setTimeoutSpy); - await waitForMockCalls(replySpy, 1); - - await vi.waitFor(() => expect(replySpy).toHaveBeenCalledTimes(1)); - const payload = replyPayload() as { Body?: string }; - expect(payload.Body).toContain("partial album"); } finally { setTimeoutSpy.mockRestore(); fetchSpy.mockRestore(); } }); + it.each([ + ["provider deny", { mode: "deny" }, undefined, undefined, false, 92073], + [ + "conversation deny", + { mode: "allow", denyIn: ["-100456"] }, + undefined, + undefined, + false, + 92074, + ], + ["topic deny", { mode: "allow", denyIn: [TELEGRAM_TEST_TOPIC] }, undefined, 42, false, 92075], + ["topic allow", { mode: "deny", allowIn: [TELEGRAM_TEST_TOPIC] }, undefined, 42, true, 92076], + ["account deny", { mode: "allow" }, { mode: "deny" }, undefined, false, 92077], + [ + "account topic allow", + { mode: "deny" }, + { mode: "deny", allowIn: [TELEGRAM_TEST_TOPIC] }, + 42, + true, + 92078, + ], + ] as TelegramMentionCaseForTest[])( + "applies %s before classifying group media mentions (#92067)", + async (_name, providerPolicy, accountPolicy, topicId, shouldWarn, messageId) => { + setTelegramIngestGroupConfig({ + customMentionPatterns: true, + providerPolicy, + accountPolicy, + }); + createTelegramBot({ token: "tok", ...(accountPolicy ? { accountId: "work" } : {}) }); + await dispatchTelegramGroupPhoto({ + messageId, + topicId, + caption: "bert, see attachment", + getFile: async () => { + throw new Error("Network request for 'getFile' failed!"); + }, + }); + const expectedWarnings = Number(shouldWarn); + expect(sendMessageSpy).toHaveBeenCalledTimes(expectedWarnings); + expect(replySpy).toHaveBeenCalledTimes(expectedWarnings); + expect(sendMessageSpy.mock.calls[0]?.[1]).toEqual( + [undefined, "⚠️ Failed to download media. Please try again."][expectedWarnings], + ); + expectTelegramIngestHook([], { + content: "bert, see attachment", + expectedCalls: Number(!shouldWarn), + }); + }, + ); + + it.each([ + { + name: "a native mention", + messageId: 81182, + caption: "@openclaw_bot check this", + ingest: false, + }, + { + name: "a native mention with ingestion", + messageId: 81186, + caption: "@openclaw_bot check this", + ingest: true, + }, + { + name: "a native mention with denied patterns", + messageId: 81185, + caption: "@openclaw_bot check this", + ingest: true, + denyPatterns: true, + }, + { + name: "a targeted bot command", + messageId: 81184, + caption: "/inspect@openclaw_bot", + extraMessage: { caption_entities: [{ type: "bot_command", offset: 0, length: 21 }] }, + ingest: false, + }, + { + name: "a reply to the bot", + messageId: 81183, + extraMessage: { + reply_to_message: { + message_id: 99, + text: "previous bot reply", + from: { id: 999, is_bot: true, first_name: "OpenClaw" }, + }, + }, + ingest: false, + }, + ])("preserves visible media failures for $name (#92067)", async (testCase) => { + setTelegramIngestGroupConfig({ + groups: { "*": { requireMention: true, ...(testCase.ingest ? { ingest: true } : {}) } }, + ...("denyPatterns" in testCase ? { providerPolicy: { mode: "deny" } } : {}), + }); + saveRemoteMedia.mockRejectedValueOnce(new Error("MediaFetchError: ECONNRESET")); + const fetchSpy = vi.spyOn(globalThis, "fetch").mockRejectedValue(new Error("ECONNRESET")); + try { + createTelegramBot({ token: "tok" }); + await dispatchTelegramGroupPhoto({ + messageId: testCase.messageId, + ...("caption" in testCase ? { caption: testCase.caption } : {}), + ...("extraMessage" in testCase ? { extraMessage: testCase.extraMessage } : {}), + }); + await waitForMockCalls(sendMessageSpy, 1); + expect(sendMessageSpy).toHaveBeenCalledWith( + -100456, + "⚠️ Failed to download media. Please try again.", + expect.objectContaining({ + reply_parameters: expect.objectContaining({ + message_id: testCase.messageId, + allow_sending_without_reply: true, + }), + }), + ); + expect(replySpy).toHaveBeenCalledOnce(); + expectTypeOnlyMediaPayload("image", "caption" in testCase ? testCase.caption : ""); + } finally { + fetchSpy.mockRestore(); + } + }); + it.each([ { failure: "shutdown aborts a download", shutdownAbort: true }, { failure: "Telegram temporarily throttles a download", shutdownAbort: false }, ])("durably retries every spooled album update when $failure", async ({ shutdownAbort }) => { setOpenChannelPostConfig(); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - const shutdown = new AbortController(); saveRemoteMedia.mockImplementationOnce(async () => { if (shutdownAbort) { @@ -876,51 +947,29 @@ describe("createTelegramBot channel_post media", () => { const handler = getOnHandler("channel_post") as ( ctx: Record, ) => Promise; - const firstUpdate = { update_id: 98079 }; - const secondUpdate = { update_id: 98080 }; - const first = await runWithTelegramSpooledReplayUpdate(firstUpdate, () => - handler({ - ...createChannelPostContext({ - messageId: 98079, - caption: "shutdown album", - date: 1736380800, - mediaGroupId: "shutdown-album-1", - photoFileId: "p1", - }), - update: firstUpdate, + const runs = await Promise.all( + [98079, 98080].map((messageId, index) => { + const update = { update_id: messageId }; + return runWithTelegramSpooledReplayUpdate(update, () => + handler({ + ...createChannelPostContext({ + messageId, + ...(index === 0 ? { caption: "shutdown album" } : {}), + date: 1736380800 + index, + mediaGroupId: "shutdown-album-1", + photoFileId: `p${index + 1}`, + }), + update, + }), + ); }), ); - const second = await runWithTelegramSpooledReplayUpdate(secondUpdate, () => - handler({ - ...createChannelPostContext({ - messageId: 98080, - date: 1736380801, - mediaGroupId: "shutdown-album-1", - photoFileId: "p2", - }), - update: secondUpdate, - }), - ); - - expect(first.deferredWork).toBeDefined(); - expect(second.deferredWork).toBeDefined(); - if (!first.deferredWork || !second.deferredWork) { - throw new Error("Expected both album updates to register durable replay work"); - } + expect(runs.map(({ deferredWork }) => Boolean(deferredWork))).toEqual([true, true]); await flushChannelPostMediaGroup(setTimeoutSpy); - - const [firstResult, secondResult] = await Promise.all([ - first.deferredWork.task, - second.deferredWork.task, + expect(await Promise.all(runs.map(({ deferredWork }) => deferredWork!.task))).toEqual([ + { kind: "failed-retryable", error: expect.any(MediaFetchError) }, + { kind: "failed-retryable", error: expect.any(MediaFetchError) }, ]); - expect(firstResult).toEqual({ - kind: "failed-retryable", - error: expect.any(MediaFetchError), - }); - expect(secondResult).toEqual({ - kind: "failed-retryable", - error: expect.any(MediaFetchError), - }); expect(sendMessageSpy).not.toHaveBeenCalled(); expect(replySpy).not.toHaveBeenCalled(); } finally { @@ -928,21 +977,16 @@ describe("createTelegramBot channel_post media", () => { } }); - it("keeps live album delivery when classic polling aborts a download", async () => { + it.each([ + { name: "a photo download fails", firstMessageId: 401, abort: false }, + { name: "classic polling aborts a download", firstMessageId: 98081, abort: true }, + ])("keeps live album delivery when $name", async ({ firstMessageId, abort }) => { setOpenChannelPostConfig(); - sendMessageSpy.mockClear(); - replySpy.mockClear(); - const shutdown = new AbortController(); + const mediaPath = "/tmp/live-album-first.jpg"; saveRemoteMedia - .mockImplementationOnce(async () => ({ - path: "/tmp/classic-restart-first.jpg", - contentType: "image/jpeg", - })) - .mockImplementationOnce(async () => { - shutdown.abort(); - throw Object.assign(new Error("aborted"), { name: "AbortError" }); - }); + .mockResolvedValueOnce({ path: mediaPath, contentType: "image/jpeg" }) + .mockImplementationOnce(() => rejectTelegramAlbumDownload(shutdown, abort)); const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); try { createTelegramBot({ @@ -954,19 +998,19 @@ describe("createTelegramBot channel_post media", () => { ctx: Record, ) => Promise; await queueChannelPostAlbum(handler, { - caption: "classic restart album", - mediaGroupId: "classic-restart-album-1", - firstMessageId: 98081, - secondMessageId: 98082, + caption: "live partial album", + mediaGroupId: `live-album-${firstMessageId}`, + firstMessageId, + secondMessageId: firstMessageId + 1, }); await flushChannelPostMediaGroup(setTimeoutSpy); await waitForMockCalls(replySpy, 1); expect(replySpy).toHaveBeenCalledTimes(1); expect(replyPayload()).toMatchObject({ - Body: expect.stringContaining("classic restart album"), + Body: expect.stringContaining("live partial album"), media: [ - expect.objectContaining({ path: "/tmp/classic-restart-first.jpg" }), + expect.objectContaining({ path: mediaPath }), expect.objectContaining({ path: undefined }), ], }); diff --git a/extensions/telegram/src/group-config-helpers.ts b/extensions/telegram/src/group-config-helpers.ts index 49be7d20cf3f..f733e5902e08 100644 --- a/extensions/telegram/src/group-config-helpers.ts +++ b/extensions/telegram/src/group-config-helpers.ts @@ -1,6 +1,7 @@ -import type { ScopeTree } from "openclaw/plugin-sdk/channel-policy"; +import { resolveChannelGroupPolicy, type ScopeTree } from "openclaw/plugin-sdk/channel-policy"; // Telegram helper module supports group config helpers behavior. import type { + OpenClawConfig, TelegramAccountConfig, TelegramDirectConfig, TelegramGroupConfig, @@ -42,6 +43,21 @@ export function resolveTelegramScopedGroupConfig( return { groupConfig, topicConfig }; } +export function resolveTelegramGroupIngestEnabled(params: { + cfg: OpenClawConfig; + chatId: string | number; + accountId?: string; + topicConfig?: TelegramTopicConfig; +}): boolean { + const { groupConfig, defaultConfig } = resolveChannelGroupPolicy({ + cfg: params.cfg, + channel: "telegram", + groupId: String(params.chatId), + accountId: params.accountId, + }); + return (params.topicConfig?.ingest ?? groupConfig?.ingest ?? defaultConfig?.ingest) === true; +} + export function resolveTelegramGroupPromptSettings(params: { groupConfig?: TelegramGroupConfig | TelegramDirectConfig; topicConfig?: TelegramTopicConfig;