From b5876a10862ced9a2ffed191d7324d57839ef2fb Mon Sep 17 00:00:00 2001 From: vovapetry <144446973+vovapetry@users.noreply.github.com> Date: Tue, 4 Aug 2026 16:19:57 +0300 Subject: [PATCH] fix(telegram): route reactions to their originating forum topics Recover Telegram forum reaction topics from provider-observed message context before authorization, and drop unknown topics instead of routing them to General. Serialize each forum message with its reaction update to avoid cache races. Thanks @vovapetry. --- docs/channels/telegram.md | 2 +- extensions/telegram/src/bot-core.ts | 4 +- ...t-handlers.message-context.runtime.test.ts | 75 +++++ .../bot-handlers.message-context.runtime.ts | 18 ++ .../src/bot-handlers.message.runtime.ts | 2 + .../src/bot-handlers.reaction.runtime.test.ts | 257 ++++++++++++++++++ .../src/bot-handlers.reaction.runtime.ts | 119 +++++--- .../telegram/src/bot-handlers.runtime.ts | 2 +- .../bot.create-telegram-bot.test-harness.ts | 4 +- .../src/bot.create-telegram-bot.test.ts | 23 +- extensions/telegram/src/bot.test.ts | 34 +-- extensions/telegram/src/message-cache.test.ts | 19 ++ extensions/telegram/src/message-cache.ts | 7 + .../telegram/src/sequential-key.test.ts | 38 ++- extensions/telegram/src/sequential-key.ts | 45 ++- 15 files changed, 565 insertions(+), 84 deletions(-) create mode 100644 extensions/telegram/src/bot-handlers.message-context.runtime.test.ts create mode 100644 extensions/telegram/src/bot-handlers.reaction.runtime.test.ts diff --git a/docs/channels/telegram.md b/docs/channels/telegram.md index e1bdc080fdf6..aea1079597c1 100644 --- a/docs/channels/telegram.md +++ b/docs/channels/telegram.md @@ -718,7 +718,7 @@ curl "https://api.telegram.org/bot/getUpdates" `own` means user reactions to bot-sent messages only (best-effort via a sent-message cache). Reaction events still respect Telegram access controls (`dmPolicy`, `allowFrom`, `groupPolicy`, `groupAllowFrom`); unauthorized senders are dropped. - Telegram does not provide thread IDs in reaction updates: non-forum groups route to the group chat session; forum groups route to the general-topic session (`:topic:1`), not the exact originating topic. + Telegram does not provide thread IDs in reaction updates. Non-forum groups route to the group chat session. Forum groups recover the originating topic from OpenClaw's bounded message cache (keyed by account, chat, and message ID), so the reaction routes to that topic's session, including its topic agent and conversation bindings. When the reacted-to message is no longer cached the topic is unknown, so OpenClaw skips the reaction notification and logs a warning instead of attributing it to General (`:topic:1`). `allowed_updates` for polling/webhook include `message_reaction` automatically. diff --git a/extensions/telegram/src/bot-core.ts b/extensions/telegram/src/bot-core.ts index 22f348f7b2f0..e689d0eb0eff 100644 --- a/extensions/telegram/src/bot-core.ts +++ b/extensions/telegram/src/bot-core.ts @@ -64,7 +64,7 @@ import { import { registerTelegramOutboundGroupHistoryRecorder } from "./outbound-message-context.js"; import { formatTelegramRawUpdateForLog } from "./raw-update-log.js"; import { createTelegramSendChatActionHandler } from "./sendchataction-401-backoff.js"; -import { getTelegramSequentialKey } from "./sequential-key.js"; +import { getTelegramSequentialConstraints } from "./sequential-key.js"; import { createTelegramThreadBindingManager } from "./thread-bindings.js"; type TelegramBotRuntime = { @@ -233,7 +233,7 @@ export function createTelegramBotCore( await next(); }); - bot.use(botRuntime.sequentialize(getTelegramSequentialKey)); + bot.use(botRuntime.sequentialize(getTelegramSequentialConstraints)); const rawUpdateLogger = createSubsystemLogger("gateway/channels/telegram/raw-update"); diff --git a/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts b/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts new file mode 100644 index 000000000000..eeb4b261b9d0 --- /dev/null +++ b/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts @@ -0,0 +1,75 @@ +// Telegram tests cover forum topic recovery from the real message cache. +import type { Message } from "grammy/types"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { beforeEach, describe, expect, it } from "vitest"; +import { createTelegramMessageContextRuntime } from "./bot-handlers.message-context.runtime.js"; +import type { RegisterTelegramHandlerParams } from "./bot-native-commands.js"; +import { resetTelegramMessageCacheForTest } from "./runtime.test-support.js"; + +const CHAT_ID = 5678; +const TOPIC_ID = 77; + +let storeScopeId = 0; + +/** + * Builds the runtime against the real message cache so the reaction path's topic + * recovery is proven through the cache it actually reads, not a stub. + */ +function createRuntime() { + storeScopeId += 1; + const cfg: OpenClawConfig = {}; + return createTelegramMessageContextRuntime({ + cfg, + accountId: "default", + opts: { token: "test" }, + telegramCfg: {}, + telegramDeps: { + resolveStorePath: () => `/tmp/openclaw-telegram-thread-recovery-${storeScopeId}/store.json`, + } as RegisterTelegramHandlerParams["telegramDeps"], + }); +} + +function forumMessage(messageId: number, threadId?: number): Message { + return { + chat: { id: CHAT_ID, type: "supergroup", title: "Forum", is_forum: true }, + message_id: messageId, + date: 1736380800, + text: "topic message", + from: { id: 10, is_bot: false, first_name: "Bob" }, + ...(threadId === undefined ? {} : { message_thread_id: threadId }), + } as Message; +} + +describe("resolveCachedMessageThreadId", () => { + beforeEach(() => { + resetTelegramMessageCacheForTest(); + }); + + it("recovers the topic of a recorded forum message", async () => { + const runtime = createRuntime(); + await runtime.recordMessageForReplyChain(forumMessage(100, TOPIC_ID), TOPIC_ID); + + await expect( + runtime.resolveCachedMessageThreadId({ chatId: CHAT_ID, messageId: 100 }), + ).resolves.toBe(TOPIC_ID); + }); + + it("returns undefined for a message that is not in the cache", async () => { + const runtime = createRuntime(); + + // Cache miss must stay unknown; the reaction handler drops rather than + // attributing the reaction to the General topic. + await expect( + runtime.resolveCachedMessageThreadId({ chatId: CHAT_ID, messageId: 404 }), + ).resolves.toBeUndefined(); + }); + + it("returns undefined for a recorded message that carries no topic", async () => { + const runtime = createRuntime(); + await runtime.recordMessageForReplyChain(forumMessage(101)); + + await expect( + runtime.resolveCachedMessageThreadId({ chatId: CHAT_ID, messageId: 101 }), + ).resolves.toBeUndefined(); + }); +}); diff --git a/extensions/telegram/src/bot-handlers.message-context.runtime.ts b/extensions/telegram/src/bot-handlers.message-context.runtime.ts index 767dfed8032e..12c34415e76e 100644 --- a/extensions/telegram/src/bot-handlers.message-context.runtime.ts +++ b/extensions/telegram/src/bot-handlers.message-context.runtime.ts @@ -23,6 +23,7 @@ import { buildTelegramReplyChain, createTelegramMessageCache, isTelegramMessageFromCurrentBot, + resolveProviderObservedTelegramThreadId, type TelegramCachedMessageNode, type TelegramReplyChainEntry, } from "./message-cache.js"; @@ -92,6 +93,22 @@ export function createTelegramMessageContextRuntime({ ...(threadId != null ? { threadId } : {}), }); + // `MessageReactionUpdated` carries no `message_thread_id`, so the reaction handler + // recovers the originating topic from the same bounded cache that records inbound + // and outbound messages. `undefined` means "thread unknown", never "General": the + // caller must not substitute a topic id. + const resolveCachedMessageThreadId = async (params: { + chatId: number | string; + messageId: number | string; + }): Promise => { + const node = await messageCache.get({ + accountId, + chatId: params.chatId, + messageId: String(params.messageId), + }); + return resolveProviderObservedTelegramThreadId(node); + }; + const buildReplyChainForMessage = (msg: Message) => buildTelegramReplyChain({ cache: messageCache, accountId, chatId: msg.chat.id, msg }); @@ -255,6 +272,7 @@ export function createTelegramMessageContextRuntime({ return { recordMessageForReplyChain, + resolveCachedMessageThreadId, buildReplyChainForMessage, toReplyChainEntry, buildPromptContextForMessage, diff --git a/extensions/telegram/src/bot-handlers.message.runtime.ts b/extensions/telegram/src/bot-handlers.message.runtime.ts index 2ee61ff2444d..7aee7988a13c 100644 --- a/extensions/telegram/src/bot-handlers.message.runtime.ts +++ b/extensions/telegram/src/bot-handlers.message.runtime.ts @@ -70,6 +70,7 @@ export function createTelegramHandlerMessageRuntime({ const { resolveTelegramSessionState, resolvePromptContextAmbientWatermark } = sessionRuntime; const { recordMessageForReplyChain, + resolveCachedMessageThreadId, buildReplyChainForMessage, toReplyChainEntry, buildPromptContextForMessage, @@ -424,6 +425,7 @@ export function createTelegramHandlerMessageRuntime({ resolveTelegramSessionState, resolvePromptContextAmbientWatermark, recordMessageForReplyChain, + resolveCachedMessageThreadId, processMessageWithReplyChain, }; } diff --git a/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts b/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts new file mode 100644 index 000000000000..abe2a869e45d --- /dev/null +++ b/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts @@ -0,0 +1,257 @@ +// Telegram tests cover forum reaction topic recovery before authorization and routing. +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { getChildLogger } from "openclaw/plugin-sdk/runtime-env"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { defaultTelegramBotDeps } from "./bot-deps.js"; +import { createTelegramHandlerAuthorizationRuntime } from "./bot-handlers.authorization.runtime.js"; +import { registerTelegramReactionHandler } from "./bot-handlers.reaction.runtime.js"; +import type { RegisterTelegramHandlerParams } from "./bot-native-commands.js"; + +const FIRE_EMOJI = "\u{1F525}"; +const FORUM_CHAT_ID = 5678; +const FORUM_TOPIC_ID = 77; +const REACTED_MESSAGE_ID = 100; + +type ReactionHandler = (ctx: Record) => Promise; + +const enqueueSystemEvent = vi.fn(); +const runtimeLog = vi.fn(); +const runtimeError = vi.fn(); +const resolveCachedMessageThreadId = vi.fn< + (params: { chatId: number | string; messageId: number | string }) => Promise +>(async () => undefined); + +function buildTelegramConfig(overrides?: { + topics?: Record; +}): OpenClawConfig { + return { + channels: { + telegram: { + dmPolicy: "open", + allowFrom: ["*"], + reactionNotifications: "all", + groupPolicy: "open", + groups: { + [String(FORUM_CHAT_ID)]: { + enabled: true, + ...(overrides?.topics ? { topics: overrides.topics } : {}), + }, + }, + }, + }, + } as OpenClawConfig; +} + +/** + * Registers the real reaction handler against the real authorization runtime so + * the test proves topic-scoped config lookup, not just the handler's own branch. + */ +function registerHandler(cfg: OpenClawConfig): ReactionHandler { + const handlers = new Map(); + const params: RegisterTelegramHandlerParams = { + accountId: "default", + bot: { + on: (name: string, handler: ReactionHandler) => { + handlers.set(name, handler); + }, + } as RegisterTelegramHandlerParams["bot"], + cfg, + mediaMaxBytes: 1, + opts: { token: "tok" }, + telegramCfg: {}, + logger: getChildLogger({ module: "telegram/reaction-test" }), + runtime: { log: runtimeLog, error: runtimeError, exit: vi.fn() }, + shouldSkipUpdate: () => false, + resolveGroupPolicy: () => ({ allowlistEnabled: false, allowed: true }), + resolveGroupActivation: () => undefined, + resolveGroupRequireMention: () => false, + resolveTelegramGroupConfig: ( + chatId: string | number, + messageThreadId: number | undefined, + config: OpenClawConfig, + ) => { + const groups = ( + config.channels?.telegram as + | { + groups?: Record< + string, + { + enabled?: boolean; + topics?: Record; + } + >; + } + | undefined + )?.groups; + const groupConfig = groups?.[String(chatId)]; + return { + groupConfig, + topicConfig: + messageThreadId === undefined + ? undefined + : groupConfig?.topics?.[String(messageThreadId)], + }; + }, + processMessage: vi.fn(), + telegramDeps: { + ...defaultTelegramBotDeps, + getRuntimeConfig: () => cfg, + wasSentByBot: () => true, + enqueueSystemEvent, + readChannelAllowFromStore: async () => [], + }, + }; + + registerTelegramReactionHandler( + params, + { resolveCachedMessageThreadId }, + createTelegramHandlerAuthorizationRuntime(params), + ); + const handler = handlers.get("message_reaction"); + if (!handler) { + throw new Error("expected message_reaction handler"); + } + return handler; +} + +function forumReactionContext(overrides?: { + oldReaction?: Array<{ type: string; emoji: string }>; + newReaction?: Array<{ type: string; emoji: string }>; + isForum?: boolean; + chatType?: string; +}) { + return { + update: { update_id: 900 }, + messageReaction: { + chat: { + id: FORUM_CHAT_ID, + type: overrides?.chatType ?? "supergroup", + ...(overrides?.isForum === false ? {} : { is_forum: true }), + }, + message_id: REACTED_MESSAGE_ID, + user: { id: 10, first_name: "Bob", username: "bob_user" }, + date: 1736380800, + old_reaction: overrides?.oldReaction ?? [], + new_reaction: overrides?.newReaction ?? [{ type: "emoji", emoji: FIRE_EMOJI }], + }, + }; +} + +function systemEventOptions(): { sessionKey?: string; contextKey?: string } { + return (enqueueSystemEvent.mock.calls[0]?.[1] ?? {}) as { + sessionKey?: string; + contextKey?: string; + }; +} + +describe("registerTelegramReactionHandler forum topic recovery", () => { + beforeEach(() => { + enqueueSystemEvent.mockClear(); + runtimeLog.mockClear(); + runtimeError.mockClear(); + resolveCachedMessageThreadId.mockReset(); + resolveCachedMessageThreadId.mockResolvedValue(undefined); + }); + + it("recovers the cached topic before authorization and routes to that topic", async () => { + resolveCachedMessageThreadId.mockResolvedValue(FORUM_TOPIC_ID); + const handler = registerHandler( + buildTelegramConfig({ topics: { [String(FORUM_TOPIC_ID)]: { enabled: true } } }), + ); + + await handler(forumReactionContext()); + + expect(resolveCachedMessageThreadId).toHaveBeenCalledWith({ + chatId: FORUM_CHAT_ID, + messageId: REACTED_MESSAGE_ID, + }); + expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); + expect(String(systemEventOptions().sessionKey)).toContain( + `telegram:group:${FORUM_CHAT_ID}:topic:${FORUM_TOPIC_ID}`, + ); + }); + + it("routes a recovered topic through its configured topic agent", async () => { + resolveCachedMessageThreadId.mockResolvedValue(FORUM_TOPIC_ID); + const handler = registerHandler( + buildTelegramConfig({ + topics: { [String(FORUM_TOPIC_ID)]: { enabled: true, agentId: "topicbot" } }, + }), + ); + + await handler(forumReactionContext()); + + expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); + expect(String(systemEventOptions().sessionKey)).toContain("topicbot"); + }); + + it("applies the recovered topic's disabled config instead of the General topic's", async () => { + resolveCachedMessageThreadId.mockResolvedValue(FORUM_TOPIC_ID); + const handler = registerHandler( + buildTelegramConfig({ + topics: { "1": { enabled: true }, [String(FORUM_TOPIC_ID)]: { enabled: false } }, + }), + ); + + await handler(forumReactionContext()); + + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + }); + + it("drops a forum reaction with an unknown topic instead of guessing General", async () => { + resolveCachedMessageThreadId.mockResolvedValue(undefined); + const handler = registerHandler(buildTelegramConfig({ topics: { "1": { enabled: true } } })); + + await handler(forumReactionContext()); + + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + expect(runtimeLog).toHaveBeenCalledTimes(1); + const logged = String(runtimeLog.mock.calls[0]?.[0] ?? ""); + expect(logged).toContain("thread-context-unavailable"); + expect(logged).toContain(`chat=${FORUM_CHAT_ID}`); + expect(logged).toContain(`message=${REACTED_MESSAGE_ID}`); + // Bounded degradation: route ids only, never message content or display names. + expect(logged).not.toContain("bob_user"); + expect(logged).not.toContain(FIRE_EMOJI); + }); + + it("never consults the message cache for non-forum groups", async () => { + const handler = registerHandler(buildTelegramConfig()); + + await handler(forumReactionContext({ isForum: false })); + + expect(resolveCachedMessageThreadId).not.toHaveBeenCalled(); + expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); + expect(String(systemEventOptions().sessionKey)).not.toContain(":topic:"); + }); + + it("never consults the message cache for direct chats", async () => { + const handler = registerHandler(buildTelegramConfig()); + + await handler(forumReactionContext({ isForum: false, chatType: "private" })); + + expect(resolveCachedMessageThreadId).not.toHaveBeenCalled(); + expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); + expect(String(systemEventOptions().sessionKey)).not.toContain(":topic:"); + expect(String(systemEventOptions().sessionKey)).not.toContain(":group:"); + }); + + it("skips the cache lookup entirely when no reaction was added", async () => { + const handler = registerHandler( + buildTelegramConfig({ topics: { [String(FORUM_TOPIC_ID)]: { enabled: true } } }), + ); + + // A removal-only update enqueues nothing, so it must not spend a cache lookup + // or log an unresolved-topic warning. + await handler( + forumReactionContext({ + oldReaction: [{ type: "emoji", emoji: FIRE_EMOJI }], + newReaction: [], + }), + ); + + expect(resolveCachedMessageThreadId).not.toHaveBeenCalled(); + expect(enqueueSystemEvent).not.toHaveBeenCalled(); + expect(runtimeLog).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/telegram/src/bot-handlers.reaction.runtime.ts b/extensions/telegram/src/bot-handlers.reaction.runtime.ts index 950fe3f1e3cb..93132af2bf50 100644 --- a/extensions/telegram/src/bot-handlers.reaction.runtime.ts +++ b/extensions/telegram/src/bot-handlers.reaction.runtime.ts @@ -1,18 +1,26 @@ // Telegram reaction handler registration. import type { ReactionTypeEmoji } from "grammy/types"; import { resolveAgentRoute } from "openclaw/plugin-sdk/routing"; -import { danger, logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { danger, logVerbose, warn } from "openclaw/plugin-sdk/runtime-env"; import { resolveTelegramAccount } from "./accounts.js"; import type { TelegramHandlerAuthorizationRuntime } from "./bot-handlers.authorization.runtime.js"; +import type { TelegramHandlerMessageRuntime } from "./bot-handlers.message.runtime.js"; import type { RegisterTelegramHandlerParams } from "./bot-native-commands.js"; -import { - buildTelegramGroupPeerId, - buildTelegramParentPeer, - resolveTelegramForumThreadId, -} from "./bot/helpers.js"; +import { buildTelegramGroupPeerId, buildTelegramParentPeer } from "./bot/helpers.js"; +import { resolveTelegramConversationRoute } from "./conversation-route.js"; + +/** Stable operator-facing reason for a forum reaction dropped without a known topic. */ +const TELEGRAM_REACTION_THREAD_UNRESOLVED_REASON = "thread-context-unavailable"; + +/** Only the message-cache lookup this handler needs, so tests can supply it directly. */ +type TelegramReactionThreadRecovery = Pick< + TelegramHandlerMessageRuntime, + "resolveCachedMessageThreadId" +>; export function registerTelegramReactionHandler( { accountId, bot, runtime, telegramDeps, shouldSkipUpdate }: RegisterTelegramHandlerParams, + threadRecovery: TelegramReactionThreadRecovery, authorizationRuntime: TelegramHandlerAuthorizationRuntime, ) { const { resolveTelegramEventAuthorizationContext, authorizeTelegramEventSender } = @@ -58,12 +66,51 @@ export function registerTelegramReactionHandler( ); return; } + // Detect added reactions. This runs before topic recovery so a reaction that + // enqueues nothing never spends a cache lookup or logs an unresolved-topic warning. + const oldEmojis = new Set( + reaction.old_reaction + .filter((r): r is ReactionTypeEmoji => r.type === "emoji") + .map((r) => r.emoji), + ); + const addedReactions = reaction.new_reaction + .filter((r): r is ReactionTypeEmoji => r.type === "emoji") + .filter((r) => !oldEmojis.has(r.emoji)); + + if (addedReactions.length === 0) { + return; + } + + // `MessageReactionUpdated` omits `message_thread_id`, so a forum reaction only has a + // topic if the reacted-to message is still in the bounded message cache. Recover it + // before authorization: topic allowlists and topic agents are both keyed by thread + // id, so an assumed thread here authorizes and routes the wrong topic. + let cachedForumThreadId: number | undefined; + if (isForum) { + cachedForumThreadId = await threadRecovery.resolveCachedMessageThreadId({ + chatId, + messageId, + }); + if (cachedForumThreadId === undefined) { + // Never fall back to General: that would authorize and enqueue this reaction + // against a topic the user did not react in. Degrade to one bounded warning + // carrying route ids only. + runtime.log?.( + warn( + `telegram: skipped forum reaction account=${accountId} chat=${chatId} message=${messageId} reason=${TELEGRAM_REACTION_THREAD_UNRESOLVED_REASON}`, + ), + ); + return; + } + } + const eventAuthContext = await resolveTelegramEventAuthorizationContext({ cfg: authorizationCfg, chatId, isGroup, isForum, senderId, + ...(cachedForumThreadId === undefined ? {} : { messageThreadId: cachedForumThreadId }), }); const senderAuthorization = await authorizeTelegramEventSender({ chatId, @@ -93,18 +140,36 @@ export function registerTelegramReactionHandler( } } - // Detect added reactions. - const oldEmojis = new Set( - reaction.old_reaction - .filter((r): r is ReactionTypeEmoji => r.type === "emoji") - .map((r) => r.emoji), - ); - const addedReactions = reaction.new_reaction - .filter((r): r is ReactionTypeEmoji => r.type === "emoji") - .filter((r) => !oldEmojis.has(r.emoji)); - - if (addedReactions.length === 0) { - return; + const resolvedThreadId = eventAuthContext.resolvedThreadId; + let sessionKey: string; + if (isForum) { + // Forum topics carry topic agents and conversation bindings, so the recovered + // topic goes through the canonical route resolver instead of a bare peer route. + sessionKey = resolveTelegramConversationRoute({ + cfg: eventAuthContext.cfg, + accountId, + chatId, + isGroup, + resolvedThreadId, + replyThreadId: resolvedThreadId, + senderId, + topicAgentId: eventAuthContext.topicConfig?.agentId, + }).route.sessionKey; + } else { + // Direct chats and non-forum groups have no topic to recover; keep their + // established peer route so reaction sessions stay where they already are. + const peerId = isGroup + ? buildTelegramGroupPeerId(chatId, resolvedThreadId) + : String(chatId); + const parentPeer = buildTelegramParentPeer({ isGroup, resolvedThreadId, chatId }); + // Fresh config for bindings lookup; other routing inputs are payload-derived. + sessionKey = resolveAgentRoute({ + cfg: eventAuthContext.cfg, + channel: "telegram", + accountId, + peer: { kind: isGroup ? "group" : "direct", id: peerId }, + parentPeer, + }).sessionKey; } // Build sender label. @@ -123,24 +188,6 @@ export function registerTelegramReactionHandler( } senderLabel = senderLabel || "unknown"; - // Reactions target a specific message_id; the Telegram Bot API does not include - // message_thread_id on MessageReactionUpdated, so we route to the chat-level - // session (forum topic routing is not available for reactions). - const resolvedThreadId = isForum - ? resolveTelegramForumThreadId({ isForum, messageThreadId: undefined }) - : undefined; - const peerId = isGroup ? buildTelegramGroupPeerId(chatId, resolvedThreadId) : String(chatId); - const parentPeer = buildTelegramParentPeer({ isGroup, resolvedThreadId, chatId }); - // Fresh config for bindings lookup; other routing inputs are payload-derived. - const route = resolveAgentRoute({ - cfg: eventAuthContext.cfg, - channel: "telegram", - accountId, - peer: { kind: isGroup ? "group" : "direct", id: peerId }, - parentPeer, - }); - const sessionKey = route.sessionKey; - // Enqueue system event for each added reaction. for (const r of addedReactions) { const emoji = r.emoji; diff --git a/extensions/telegram/src/bot-handlers.runtime.ts b/extensions/telegram/src/bot-handlers.runtime.ts index b43f803b1165..35bb2a8e4e1d 100644 --- a/extensions/telegram/src/bot-handlers.runtime.ts +++ b/extensions/telegram/src/bot-handlers.runtime.ts @@ -13,7 +13,7 @@ export const registerTelegramHandlers = (params: RegisterTelegramHandlerParams) const authorizationRuntime = createTelegramHandlerAuthorizationRuntime(params); const inboundRuntime = createTelegramHandlerInboundRuntime(params, messageRuntime); - registerTelegramReactionHandler(params, authorizationRuntime); + registerTelegramReactionHandler(params, messageRuntime, authorizationRuntime); registerTelegramCallbackQueryHandler(params, messageRuntime, authorizationRuntime); registerTelegramMigrationHandler(params); registerTelegramMessageHandlers(params, messageRuntime, authorizationRuntime, inboundRuntime); diff --git a/extensions/telegram/src/bot.create-telegram-bot.test-harness.ts b/extensions/telegram/src/bot.create-telegram-bot.test-harness.ts index 86bbb6b52f48..8cfe4d6e4348 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.test-harness.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.test-harness.ts @@ -442,7 +442,7 @@ const runnerHoisted = vi.hoisted(() => ({ throttlerSpy: vi.fn(() => "throttler"), })); export const sequentializeSpy: AnyMock = runnerHoisted.sequentializeSpy; -export let sequentializeKey: ((ctx: unknown) => string) | undefined; +export let sequentializeKey: ((ctx: unknown) => string | string[] | undefined) | undefined; export const throttlerSpy: AnyMock = runnerHoisted.throttlerSpy; const telegramBotRuntimeForTest = { Bot: class { @@ -492,7 +492,7 @@ const telegramBotRuntimeForTest = { ); } } as unknown as TelegramBotRuntimeForTest["Bot"], - sequentialize: ((keyFn: (ctx: unknown) => string) => { + sequentialize: ((keyFn: (ctx: unknown) => string | string[] | undefined) => { sequentializeKey = keyFn; return ( runnerHoisted.sequentializeSpy as unknown as () => ReturnType< diff --git a/extensions/telegram/src/bot.create-telegram-bot.test.ts b/extensions/telegram/src/bot.create-telegram-bot.test.ts index 72f7991823b3..a9847a43a5a3 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.test.ts @@ -79,7 +79,7 @@ type BuildModelsProviderDataMock = ReturnType< const { resolveTelegramFetch } = await import("./fetch.js"); const messageDispatchDedupe = await import("./message-dispatch-dedupe.js"); const { createTelegramBotCore: createTelegramBotBase } = await import("./bot-core.js"); -const { getTelegramSequentialKey } = await import("./sequential-key.js"); +const { getTelegramSequentialConstraints } = await import("./sequential-key.js"); const { createTelegramSpooledReplayDeferredParticipant, recordTelegramMessageProcessingResult, @@ -309,21 +309,24 @@ function installPerKeySequentializer(): void { sequentializeSpy.mockImplementationOnce(() => { const lanes = new Map>(); return async (ctx: TelegramMiddlewareTestContext, next: () => Promise) => { - const key = harness.sequentializeKey?.(ctx) ?? "default"; - const previous = lanes.get(key) ?? Promise.resolve(); + const constraint = harness.sequentializeKey?.(ctx) ?? "default"; + const keys = Array.isArray(constraint) ? constraint : [constraint]; + const previous = Promise.all(keys.map((key) => lanes.get(key) ?? Promise.resolve())); const current = previous.then(async () => { await next(); }); - lanes.set( - key, - current.catch(() => undefined), - ); + const tracked = current.catch(() => undefined); + for (const key of keys) { + lanes.set(key, tracked); + } try { await current; } finally { - if (lanes.get(key) === current) { - lanes.delete(key); + for (const key of keys) { + if (lanes.get(key) === tracked) { + lanes.delete(key); + } } } }; @@ -590,7 +593,7 @@ describe("createTelegramBot", () => { createTelegramBot({ token: "tok" }); expect(sequentializeSpy).toHaveBeenCalledTimes(1); expect(middlewareUseSpy).toHaveBeenCalledWith(sequentializeSpy.mock.results[0]?.value); - expect(harness.sequentializeKey).toBe(getTelegramSequentialKey); + expect(harness.sequentializeKey).toBe(getTelegramSequentialConstraints); }); it("answers callback queries before same-chat sequentialize delays handlers", async () => { diff --git a/extensions/telegram/src/bot.test.ts b/extensions/telegram/src/bot.test.ts index 297c6e61efa2..e3cb131c7633 100644 --- a/extensions/telegram/src/bot.test.ts +++ b/extensions/telegram/src/bot.test.ts @@ -127,7 +127,6 @@ const CHECK_MARK_EMOJI = "\u{2705}"; const THUMBS_UP_EMOJI = "\u{1F44D}"; const FIRE_EMOJI = "\u{1F525}"; const PARTY_EMOJI = "\u{1F389}"; -const EYES_EMOJI = "\u{1F440}"; const HEART_EMOJI = "\u{2764}\u{FE0F}"; type TelegramChannelConfig = NonNullable["telegram"]>; @@ -5803,9 +5802,10 @@ describe("createTelegramBot", () => { ]); }); - it("routes forum group reactions to the general topic (thread id not available on reactions)", async () => { - // MessageReactionUpdated does not include message_thread_id in the Bot API, - // so forum reactions always route to the general topic (1). + it("drops forum reactions whose topic is no longer in the message cache", async () => { + // MessageReactionUpdated carries no message_thread_id, and this message was never + // cached, so the topic is unknown. Guessing General would authorize and enqueue + // against a topic the user did not react in. await dispatchTelegramReaction({ updateId: 505, channelConfig: { dmPolicy: "open", reactionNotifications: "all" }, @@ -5817,31 +5817,7 @@ describe("createTelegramBot", () => { }, }); - expect(enqueueSystemEventSpy).toHaveBeenCalledTimes(1); - expect(firstSystemEventArg(0)).toBe( - `Telegram reaction added: ${FIRE_EMOJI} by Bob (@bob_user) on msg 100`, - ); - expect(String(systemEventOptions().sessionKey)).toContain("telegram:group:5678:topic:1"); - expect(String(systemEventOptions().contextKey)).toContain("telegram:reaction:add:5678:100:10"); - }); - - it("uses correct session key for forum group reactions in general topic", async () => { - await dispatchTelegramReaction({ - updateId: 506, - channelConfig: { dmPolicy: "open", reactionNotifications: "all" }, - reaction: { - chat: { id: 5678, type: "supergroup", is_forum: true }, - message_id: 101, - // No message_thread_id - should default to general topic (1) - user: { id: 10, first_name: "Bob" }, - new_reaction: [{ type: "emoji", emoji: EYES_EMOJI }], - }, - }); - - expect(enqueueSystemEventSpy).toHaveBeenCalledTimes(1); - expect(firstSystemEventArg(0)).toBe(`Telegram reaction added: ${EYES_EMOJI} by Bob on msg 101`); - expect(String(systemEventOptions().sessionKey)).toContain("telegram:group:5678:topic:1"); - expect(String(systemEventOptions().contextKey)).toContain("telegram:reaction:add:5678:101:10"); + expect(enqueueSystemEventSpy).not.toHaveBeenCalled(); }); it("uses correct session key for regular group reactions without topic", async () => { diff --git a/extensions/telegram/src/message-cache.test.ts b/extensions/telegram/src/message-cache.test.ts index a94d5f70ac06..d2cb1da45064 100644 --- a/extensions/telegram/src/message-cache.test.ts +++ b/extensions/telegram/src/message-cache.test.ts @@ -9,6 +9,7 @@ import { buildTelegramReplyChain, createTelegramMessageCache, hasProviderObservedTelegramThreadBinding, + resolveProviderObservedTelegramThreadId, } from "./message-cache.js"; import { resetTelegramMessageCacheForTest as resetCache } from "./runtime.test-support.js"; @@ -178,9 +179,26 @@ describe("telegram message cache", () => { for (const messageId of ["901", "902"]) { const node = await get(reloaded, messageId, { chatId: -1001 }); expect(hasProviderObservedTelegramThreadBinding(node, 77)).toBe(true); + expect(resolveProviderObservedTelegramThreadId(node)).toBe(77); } }); + it("does not resolve caller-only topic metadata as a provider-observed binding", async () => { + const cache = createTelegramMessageCache(); + await record( + cache, + message(903, "Ada", { + chat: { id: -1001, type: "supergroup", title: "QA", is_forum: true }, + message_thread_id: 77, + is_topic_message: true, + }), + { chatId: -1001, threadId: 77 }, + ); + + const node = await get(cache, "903", { chatId: -1001 }); + expect(resolveProviderObservedTelegramThreadId(node)).toBeUndefined(); + }); + it("hydrates reply chains from persisted cached messages", async () => { const { bucketKey, store } = createMemoryStore(); const photoMessage = message(9000, "Kesava", { @@ -546,6 +564,7 @@ describe("telegram message cache", () => { }); expect(reloaded?.promptContextProjectionMarker).toBeUndefined(); expect(hasProviderObservedTelegramThreadBinding(reloaded, 77)).toBe(false); + expect(resolveProviderObservedTelegramThreadId(reloaded)).toBeUndefined(); }); it("rejects unknown future persisted cache versions", async () => { diff --git a/extensions/telegram/src/message-cache.ts b/extensions/telegram/src/message-cache.ts index 1aa11ac79a8e..5befe048f02c 100644 --- a/extensions/telegram/src/message-cache.ts +++ b/extensions/telegram/src/message-cache.ts @@ -260,6 +260,13 @@ export function hasProviderObservedTelegramThreadBinding( return normalizeTelegramMessageThreadBinding(node?.threadBinding, threadId) !== undefined; } +export function resolveProviderObservedTelegramThreadId( + node: TelegramCachedMessageNode | null | undefined, +): number | undefined { + const threadId = parseTelegramMessageThreadId(node?.threadId); + return hasProviderObservedTelegramThreadBinding(node, threadId) ? threadId : undefined; +} + function normalizeMessageNodes( msg: Message, params: { diff --git a/extensions/telegram/src/sequential-key.test.ts b/extensions/telegram/src/sequential-key.test.ts index a837c1d215bb..7704d0ccf8f8 100644 --- a/extensions/telegram/src/sequential-key.test.ts +++ b/extensions/telegram/src/sequential-key.test.ts @@ -3,7 +3,7 @@ import type { Chat, Message } from "grammy/types"; import { describe, expect, it } from "vitest"; import { buildTelegramApprovalCallbackData } from "./approval-callback-data.js"; import { buildTelegramQuestionCallbackData } from "./question-callback-data.js"; -import { getTelegramSequentialKey } from "./sequential-key.js"; +import { getTelegramSequentialConstraints, getTelegramSequentialKey } from "./sequential-key.js"; const mockChat = (chat: Pick & Partial>): Chat => chat as Chat; @@ -381,6 +381,40 @@ describe("getTelegramSequentialKey", () => { "telegram:123", ], ])("resolves key %#", (input, expected) => { - expect(getTelegramSequentialKey(input)).toBe(expected); + expect(getTelegramSequentialKey(input)).toEqual(expected); + }); +}); + +describe("getTelegramSequentialConstraints", () => { + it("bridges a forum message update with its reaction update", () => { + const message = mockMessage({ + chat: mockChat({ id: -1001, type: "supergroup", is_forum: true }), + message_id: 77, + message_thread_id: 9, + is_topic_message: true, + }); + const expected = "telegram:-1001:message:77"; + const reaction = { + update: { + message_reaction: { + chat: { id: -1001, type: "supergroup", is_forum: true }, + message_id: 77, + }, + }, + }; + + expect(getTelegramSequentialConstraints({ message })).toEqual([ + "telegram:-1001:topic:9", + expected, + ]); + expect(getTelegramSequentialConstraints(reaction)).toEqual(["telegram:-1001", expected]); + }); + + it("does not add a bridge lane outside forum chats", () => { + expect( + getTelegramSequentialConstraints({ + message: mockMessage({ chat: mockChat({ id: 123, type: "private" }) }), + }), + ).toBe("telegram:123"); }); }); diff --git a/extensions/telegram/src/sequential-key.ts b/extensions/telegram/src/sequential-key.ts index 8bf769b15191..31f3298aa064 100644 --- a/extensions/telegram/src/sequential-key.ts +++ b/extensions/telegram/src/sequential-key.ts @@ -44,10 +44,43 @@ type TelegramSequentialKeyContext = { channel_post?: Message; edited_channel_post?: Message; callback_query?: { message?: Message; data?: string }; - message_reaction?: { chat?: { id?: number } }; + message_reaction?: { + chat?: { id?: number; type?: string; is_forum?: boolean }; + message_id?: number; + }; }; }; +function getTelegramMessageReactionSequentialKey( + ctx: TelegramSequentialKeyContext, +): string | undefined { + const reaction = ctx.update?.message_reaction; + if ( + reaction?.chat?.is_forum === true && + typeof reaction.chat.id === "number" && + typeof reaction.message_id === "number" + ) { + return `telegram:${reaction.chat.id}:message:${reaction.message_id}`; + } + const msg = + ctx.message ?? + ctx.channelPost ?? + ctx.editedMessage ?? + ctx.editedChannelPost ?? + ctx.update?.message ?? + ctx.update?.edited_message ?? + ctx.update?.channel_post ?? + ctx.update?.edited_channel_post; + const isForum = resolveTelegramMessageForumFlagHint({ + chatType: msg?.chat?.type, + isForum: msg?.chat?.is_forum, + isTopicMessage: msg?.is_topic_message, + }); + return isForum && typeof msg?.chat.id === "number" && typeof msg.message_id === "number" + ? `telegram:${msg.chat.id}:message:${msg.message_id}` + : undefined; +} + export function isTelegramReadOnlyControlLaneText(params: { rawText?: string; botUsername?: string; @@ -218,3 +251,13 @@ export function getTelegramSequentialKey(ctx: TelegramSequentialKeyContext): str } return "telegram:unknown"; } + +export function getTelegramSequentialConstraints( + ctx: TelegramSequentialKeyContext, +): string | string[] { + const key = getTelegramSequentialKey(ctx); + const messageKey = getTelegramMessageReactionSequentialKey(ctx); + // A forum reaction reads the topic fact recorded by the message update it targets. + // Bridge that exact message without serializing unrelated forum topics. + return messageKey ? [key, messageKey] : key; +}