From af3b86091f4bdc564c2df005259239c09cbc3e4f Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 19 Aug 2026 18:00:56 -0700 Subject: [PATCH] fix(telegram): preserve direct-message topic routing (#126207) * fix(telegram): preserve direct message topic identity * fix(telegram): break topic routing import cycle --- .../src/bot-handlers.callback-actions.ts | 10 +- .../src/bot-handlers.callback-router.ts | 33 ++-- .../telegram/src/bot-handlers.debounce-key.ts | 7 +- .../src/bot-handlers.event-bindings.ts | 45 +---- .../src/bot-handlers.inbound-authorization.ts | 2 +- ...handlers.inbound-buffer.provenance.test.ts | 1 + .../src/bot-handlers.inbound-buffer.ts | 20 +- .../src/bot-handlers.inbound-media.ts | 26 ++- .../src/bot-handlers.inbound-pipeline.test.ts | 22 ++- .../src/bot-handlers.inbound-pipeline.ts | 9 +- .../src/bot-handlers.inbound-processing.ts | 28 ++- ...t-handlers.message-context.runtime.test.ts | 3 +- .../src/bot-handlers.message-context.ts | 18 +- ...t-handlers.message-session.runtime.test.ts | 3 +- .../src/bot-handlers.reaction.runtime.test.ts | 4 +- .../src/bot-message-context.body.test.ts | 1 + .../telegram/src/bot-message-context.body.ts | 15 +- .../bot-message-context.dm-threads.test.ts | 5 +- .../src/bot-message-context.session.ts | 4 +- ...bot-message-context.thread-binding.test.ts | 12 +- .../telegram/src/bot-message-context.ts | 6 +- .../telegram/src/bot-message-context.types.ts | 2 + .../src/bot-message-dispatch-context.ts | 4 +- .../src/bot-message-dispatch-reply.ts | 2 +- .../src/bot-native-command-dispatch.ts | 22 +-- .../src/bot-native-command-plugins.ts | 2 +- .../src/bot.create-telegram-bot.test.ts | 20 +- extensions/telegram/src/bot.test.ts | 3 + extensions/telegram/src/bot/helpers.ts | 37 ++-- extensions/telegram/src/channel.ts | 187 ++++++++++-------- ...onversation-route.base-session-key.test.ts | 4 +- extensions/telegram/src/conversation-route.ts | 34 ++-- extensions/telegram/src/error-policy.test.ts | 19 +- extensions/telegram/src/error-policy.ts | 9 +- .../telegram/src/location-message-hook.ts | 2 +- .../telegram/src/session-conversation.test.ts | 20 ++ .../telegram/src/session-conversation.ts | 14 +- extensions/telegram/src/session-route.test.ts | 32 +++ .../src/telegram-ingress-supersede-auth.ts | 5 +- extensions/telegram/src/thread-spec.ts | 6 + .../telegram/src/topic-conversation.test.ts | 16 +- extensions/telegram/src/topic-conversation.ts | 98 +++++---- .../telegram/src/transport-payload.test.ts | 81 ++++++-- 43 files changed, 527 insertions(+), 366 deletions(-) create mode 100644 extensions/telegram/src/thread-spec.ts diff --git a/extensions/telegram/src/bot-handlers.callback-actions.ts b/extensions/telegram/src/bot-handlers.callback-actions.ts index 35a55f9481bd..f5e907f6a186 100644 --- a/extensions/telegram/src/bot-handlers.callback-actions.ts +++ b/extensions/telegram/src/bot-handlers.callback-actions.ts @@ -1,7 +1,7 @@ import type { Message } from "grammy/types"; import { questionGatewayRuntime } from "openclaw/plugin-sdk/question-gateway-runtime"; import type { RegisterTelegramHandlerParams } from "./bot-handlers.types.js"; -import { buildTelegramThreadParams, resolveTelegramMessageThreadSpec } from "./bot/helpers.js"; +import { buildTelegramThreadParams, type TelegramThreadSpec } from "./bot/helpers.js"; import type { TelegramQuestionCallback } from "./question-callback-data.js"; import { buildInlineKeyboard } from "./send.js"; @@ -44,9 +44,9 @@ export interface TelegramCallbackMessageActions { export function createTelegramCallbackMessageActions(params: { bot: RegisterTelegramHandlerParams["bot"]; callbackMessage: Message; - isForum: boolean; + threadSpec: TelegramThreadSpec; }): TelegramCallbackMessageActions { - const { bot, callbackMessage, isForum } = params; + const { bot, callbackMessage, threadSpec } = params; const callbackBusinessParams = callbackMessage.business_connection_id !== undefined ? { business_connection_id: callbackMessage.business_connection_id } @@ -89,9 +89,7 @@ export function createTelegramCallbackMessageActions(params: { }; const replyToCallbackChat = async (text: string, replyParams?: TelegramCallbackReplyParams) => { - const threadParams = buildTelegramThreadParams( - resolveTelegramMessageThreadSpec(callbackMessage, isForum), - ); + const threadParams = buildTelegramThreadParams(threadSpec); const mergedParams = callbackBusinessParams || threadParams || replyParams ? { ...replyParams, ...callbackBusinessParams, ...threadParams } diff --git a/extensions/telegram/src/bot-handlers.callback-router.ts b/extensions/telegram/src/bot-handlers.callback-router.ts index 772e52e8d273..aec16116ee95 100644 --- a/extensions/telegram/src/bot-handlers.callback-router.ts +++ b/extensions/telegram/src/bot-handlers.callback-router.ts @@ -67,6 +67,7 @@ import { parseTelegramQuestionCallbackData, } from "./question-callback-data.js"; import { buildInlineKeyboard } from "./send.js"; +import { buildTelegramConversationId } from "./topic-conversation.js"; export function createTelegramCallbackRouter({ params: { @@ -175,7 +176,6 @@ export function createTelegramCallbackRouter({ return; } - const messageThreadId = callbackMessage.message_thread_id; const isForum = await resolveTelegramForumFlag({ chatId, chatType: callbackMessage.chat.type, @@ -193,7 +193,8 @@ export function createTelegramCallbackRouter({ senderId, threadSpec: resolveTelegramMessageThreadSpec(callbackMessage, isForum), }); - const { resolvedThreadId, dmThreadId, storeAllowFrom, groupConfig } = eventAuthContext; + const threadSpec = eventAuthContext.threadSpec; + const { dmThreadId, storeAllowFrom, groupConfig } = eventAuthContext; const requireTopic = (groupConfig as { requireTopic?: boolean } | undefined)?.requireTopic; if (!isGroup && requireTopic === true && dmThreadId == null) { logVerbose( @@ -204,7 +205,7 @@ export function createTelegramCallbackRouter({ const actions = createTelegramCallbackMessageActions({ bot, callbackMessage, - isForum, + threadSpec, }); const clearRoutedCallbackButtons = async () => { try { @@ -262,9 +263,8 @@ export function createTelegramCallbackRouter({ return; } - const callbackThreadId = resolvedThreadId ?? dmThreadId; - const callbackConversationId = - callbackThreadId != null ? `${chatId}:topic:${callbackThreadId}` : String(chatId); + const callbackConversationId = buildTelegramConversationId({ chatId, thread: threadSpec }); + const callbackThreadId = threadSpec.id; const runtimeCfg = telegramDeps.getRuntimeConfig(); const approvalRuntime = createTelegramCallbackApprovalRuntime({ accountId, @@ -344,9 +344,7 @@ export function createTelegramCallbackRouter({ ctx, chatId, isGroup, - isForum, - messageThreadId, - resolvedThreadId, + threadSpec, senderId, runtimeCfg, telegramDeps, @@ -375,6 +373,7 @@ export function createTelegramCallbackRouter({ allMedia: [], storeAllowFrom, options: { + threadSpec, ...(nativeCallbackCommand ? { commandSource: "native" as const } : {}), forceWasMentioned: true, messageIdOverride: callback.id, @@ -416,9 +415,7 @@ async function handleTelegramModelCallback(params: { ctx: Pick; chatId: number; isGroup: boolean; - isForum: boolean; - messageThreadId?: number; - resolvedThreadId?: number; + threadSpec: ReturnType; senderId: string; runtimeCfg: OpenClawConfig; telegramDeps: RegisterTelegramHandlerParams["telegramDeps"]; @@ -431,9 +428,7 @@ async function handleTelegramModelCallback(params: { ctx, chatId, isGroup, - isForum, - messageThreadId, - resolvedThreadId, + threadSpec, senderId, runtimeCfg, telegramDeps, @@ -465,9 +460,7 @@ async function handleTelegramModelCallback(params: { messageRuntime.resolveTelegramSessionState({ chatId, isGroup, - isForum, - messageThreadId, - resolvedThreadId, + threadSpec, botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(ctx.me), senderId, runtimeCfg, @@ -514,9 +507,7 @@ async function handleTelegramModelCallback(params: { const session = messageRuntime.resolveTelegramSessionState({ chatId, isGroup, - isForum, - messageThreadId, - resolvedThreadId, + threadSpec, botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(ctx.me), senderId, runtimeCfg, diff --git a/extensions/telegram/src/bot-handlers.debounce-key.ts b/extensions/telegram/src/bot-handlers.debounce-key.ts index 13f31974e1ec..f24b5e66aa13 100644 --- a/extensions/telegram/src/bot-handlers.debounce-key.ts +++ b/extensions/telegram/src/bot-handlers.debounce-key.ts @@ -1,4 +1,5 @@ // Telegram plugin module implements bot handlersebounce key behavior. +import { buildTelegramGroupPeerId, type TelegramThreadSpec } from "./bot/helpers.js"; export function buildTelegramInboundDebounceKey(params: { accountId?: string | null; conversationKey: string; @@ -11,9 +12,7 @@ export function buildTelegramInboundDebounceKey(params: { export function buildTelegramInboundDebounceConversationKey(params: { chatId: number | string; - threadId?: number | null; + threadSpec: TelegramThreadSpec; }): string { - return params.threadId != null - ? `${params.chatId}:topic:${params.threadId}` - : String(params.chatId); + return buildTelegramGroupPeerId(params.chatId, params.threadSpec); } diff --git a/extensions/telegram/src/bot-handlers.event-bindings.ts b/extensions/telegram/src/bot-handlers.event-bindings.ts index fcac65a535d6..36c2817772dd 100644 --- a/extensions/telegram/src/bot-handlers.event-bindings.ts +++ b/extensions/telegram/src/bot-handlers.event-bindings.ts @@ -1,7 +1,6 @@ import type { ChatMember, ReactionTypeEmoji } from "grammy/types"; import { resolveChannelConfigWrites } from "openclaw/plugin-sdk/channel-config-helpers"; import { mutateConfigFile } from "openclaw/plugin-sdk/config-mutation"; -import { resolveAgentRoute } from "openclaw/plugin-sdk/routing"; import { danger, logVerbose, warn } from "openclaw/plugin-sdk/runtime-env"; import { resolveTelegramAccount } from "./accounts.js"; import type { TelegramHandlerAuthorization } from "./bot-handlers.inbound-authorization.js"; @@ -11,12 +10,7 @@ import { isTelegramSpooledReplayUpdate, recordTelegramMessageProcessingResult, } from "./bot-processing-outcome.js"; -import { - buildTelegramGroupPeerId, - buildTelegramParentPeer, - resolveTelegramThreadSpec, - type TelegramThreadSpec, -} from "./bot/helpers.js"; +import { resolveTelegramThreadSpec, type TelegramThreadSpec } from "./bot/helpers.js"; import { resolveTelegramConversationRoute } from "./conversation-route.js"; import { migrateTelegramGroupConfig } from "./group-migration.js"; import { getPreparedTelegramPollAnswer } from "./poll-answer-context.js"; @@ -180,34 +174,15 @@ export function createTelegramEventBindings({ } } - const resolvedThreadId = eventAuthContext.resolvedThreadId; - let sessionKey: string; - if (recoveredThreadSpec) { - // Scoped topics must retain topic agents and conversation bindings. - sessionKey = resolveTelegramConversationRoute({ - cfg: eventAuthContext.cfg, - accountId, - chatId, - isGroup, - resolvedThreadId, - replyThreadId: recoveredThreadSpec.id, - senderId, - topicAgentId: eventAuthContext.topicConfig?.agentId, - }).route.sessionKey; - } else { - // Direct chats and non-forum groups retain their established peer route. - const peerId = isGroup - ? buildTelegramGroupPeerId(chatId, resolvedThreadId) - : String(chatId); - const parentPeer = buildTelegramParentPeer({ isGroup, resolvedThreadId, chatId }); - sessionKey = resolveAgentRoute({ - cfg: eventAuthContext.cfg, - channel: "telegram", - accountId, - peer: { kind: isGroup ? "group" : "direct", id: peerId }, - parentPeer, - }).sessionKey; - } + const sessionKey = resolveTelegramConversationRoute({ + cfg: eventAuthContext.cfg, + accountId, + chatId, + isGroup, + threadSpec: recoveredThreadSpec ?? eventAuthContext.threadSpec, + senderId, + topicAgentId: eventAuthContext.topicConfig?.agentId, + }).route.sessionKey; const senderName = user ? [user.first_name, user.last_name].filter(Boolean).join(" ").trim() || user.username diff --git a/extensions/telegram/src/bot-handlers.inbound-authorization.ts b/extensions/telegram/src/bot-handlers.inbound-authorization.ts index 5103346ab5fc..7825fa856d29 100644 --- a/extensions/telegram/src/bot-handlers.inbound-authorization.ts +++ b/extensions/telegram/src/bot-handlers.inbound-authorization.ts @@ -301,7 +301,7 @@ export function createTelegramHandlerAuthorization({ accountId, chatId, isGroup, - resolvedThreadId: context.resolvedThreadId, + threadSpec: context.threadSpec, senderId, senderUsername, }).isAuthorizedSender; diff --git a/extensions/telegram/src/bot-handlers.inbound-buffer.provenance.test.ts b/extensions/telegram/src/bot-handlers.inbound-buffer.provenance.test.ts index fb7191184d9b..12baa8f1f6aa 100644 --- a/extensions/telegram/src/bot-handlers.inbound-buffer.provenance.test.ts +++ b/extensions/telegram/src/bot-handlers.inbound-buffer.provenance.test.ts @@ -72,6 +72,7 @@ describe("Telegram inbound provenance buffering", () => { receivedAtMs: index + 1, debounceKey: "telegram:default:42:42:default", debounceLane: "default", + threadSpec: { scope: "none" }, dispatchDedupeClaims: [], channelIngressResolvers: [channelIngressResolver], }); diff --git a/extensions/telegram/src/bot-handlers.inbound-buffer.ts b/extensions/telegram/src/bot-handlers.inbound-buffer.ts index 9a5b67d5e20b..1d380067238e 100644 --- a/extensions/telegram/src/bot-handlers.inbound-buffer.ts +++ b/extensions/telegram/src/bot-handlers.inbound-buffer.ts @@ -19,7 +19,7 @@ import { buildTelegramThreadParams, getTelegramTextParts, joinTelegramTextParts, - resolveTelegramMessageThreadSpec, + type TelegramThreadSpec, } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; import type { TelegramMessageDispatchReplayClaim } from "./message-dispatch-dedupe.js"; @@ -35,7 +35,7 @@ export type TelegramDebounceEntry = { debounceKey: string | null; debounceLane: TelegramDebounceLane; botUsername?: string; - threadId?: number; + threadSpec: TelegramThreadSpec; promptContextMinTimestampMs?: number; promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark; dispatchDedupeClaims: TelegramMessageDispatchReplayClaim[]; @@ -47,6 +47,7 @@ type TextFragmentEntry = { key: string; storeAllowFrom: string[]; messages: Array<{ msg: Message; ctx: TelegramContext; receivedAtMs: number }>; + threadSpec: TelegramThreadSpec; promptContextMinTimestampMs?: number; promptContextAmbientWatermark?: TelegramAmbientTranscriptWatermark; dispatchDedupeClaims: TelegramMessageDispatchReplayClaim[]; @@ -59,8 +60,7 @@ type TelegramTextFragmentInput = { ctx: TelegramContext; msg: Message; chatId: number; - resolvedThreadId?: number; - dmThreadId?: number; + threadSpec: TelegramThreadSpec; storeAllowFrom: string[]; isAbortControlMessage: boolean; isAuthorizedAbortControlMessage: () => Promise; @@ -164,6 +164,7 @@ export function createTelegramInboundBuffers({ options: { receivedAtMs: last.receivedAtMs, ingressBuffer: "inbound-debounce", + threadSpec: last.threadSpec, ...promptContextBoundaryOptions( last.promptContextMinTimestampMs, last.promptContextAmbientWatermark, @@ -212,6 +213,7 @@ export function createTelegramInboundBuffers({ ), receivedAtMs: first.receivedAtMs, ingressBuffer: "inbound-debounce", + threadSpec: first.threadSpec, bufferedMessages: entries.map((entry) => entry.msg), ...promptContextBoundaryOptions( latestPromptContextMinTimestampMs( @@ -253,10 +255,7 @@ export function createTelegramInboundBuffers({ } const chatId = items[0]?.msg.chat.id; if (chatId != null) { - const firstMessage = items[0]?.msg; - const threadParams = firstMessage - ? buildTelegramThreadParams(resolveTelegramMessageThreadSpec(firstMessage)) - : undefined; + const threadParams = buildTelegramThreadParams(items[0]?.threadSpec); void bot.api .sendMessage( chatId, @@ -326,6 +325,7 @@ export function createTelegramInboundBuffers({ ambientTranscriptBody: formatTelegramAmbientTranscriptBody(bufferedMessages), receivedAtMs: first.receivedAtMs, ingressBuffer: "text-fragment", + threadSpec: entry.threadSpec, bufferedMessages, ...promptContextBoundaryOptions( entry.promptContextMinTimestampMs, @@ -366,8 +366,7 @@ export function createTelegramInboundBuffers({ (entity) => entity.type === "bot_command" && entity.offset === 0, ); const senderId = params.msg.from?.id != null ? String(params.msg.from.id) : "unknown"; - const threadId = params.resolvedThreadId ?? params.dmThreadId; - const key = `text:${params.chatId}:${threadId ?? "main"}:${senderId}`; + const key = `text:${params.chatId}:${params.threadSpec.scope}:${params.threadSpec.id ?? "main"}:${senderId}`; if (text && !isCommand && !params.isAbortControlMessage) { const nowMs = Date.now(); const existing = textBuffer.get(key); @@ -416,6 +415,7 @@ export function createTelegramInboundBuffers({ const entry: TextFragmentEntry = { key, storeAllowFrom: params.storeAllowFrom, + threadSpec: params.threadSpec, messages: [{ msg: params.msg, ctx: params.ctx, receivedAtMs: nowMs }], dispatchDedupeClaims: params.dispatchDedupeClaims, spooledReplayParticipants: participant ? [participant] : [], diff --git a/extensions/telegram/src/bot-handlers.inbound-media.ts b/extensions/telegram/src/bot-handlers.inbound-media.ts index fc87ce3c57a1..cbed53ca68ba 100644 --- a/extensions/telegram/src/bot-handlers.inbound-media.ts +++ b/extensions/telegram/src/bot-handlers.inbound-media.ts @@ -35,8 +35,8 @@ import { buildTelegramThreadParams, getTelegramTextParts, hasBotMention, - resolveTelegramMessageThreadSpec, resolveTelegramPrimaryMedia, + type TelegramThreadSpec, } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; import { isTelegramForumServiceMessage } from "./forum-service-message.js"; @@ -49,8 +49,7 @@ type MediaAuthorization = { chatId: number; isGroup: boolean; isForum: boolean; - resolvedThreadId?: number; - dmThreadId?: number; + threadSpec: TelegramThreadSpec; senderId: string; effectiveGroupAllow: NormalizedAllowFrom; effectiveDmAllow: NormalizedAllowFrom; @@ -135,8 +134,11 @@ export function createTelegramInboundMedia({ const resolveUnaddressedGroupMediaDisposition = async ( authorization: MediaAuthorization & { ctx: TelegramContext; msg: Message }, ): Promise => { - const { ctx, msg, chatId, isGroup, isForum, resolvedThreadId, dmThreadId, senderId } = - authorization; + const { ctx, msg, chatId, isGroup, senderId, threadSpec } = authorization; + const resolvedThreadId = + threadSpec.scope === "forum" || threadSpec.scope === "direct-messages" + ? threadSpec.id + : undefined; const textParts = getTelegramTextParts(msg); const documentMime = msg.document?.mime_type?.split(";")[0]?.trim().toLowerCase(); const mayNeedDownload = @@ -152,9 +154,7 @@ export function createTelegramInboundMedia({ const sessionState = resolveTelegramSessionState({ chatId, isGroup, - isForum, - resolvedThreadId, - messageThreadId: resolvedThreadId ?? dmThreadId, + threadSpec, senderId, runtimeCfg: authorization.authorizationCfg, }); @@ -214,7 +214,7 @@ export function createTelegramInboundMedia({ sessionState.agentId, { provider: "telegram", - conversationId: buildTelegramGroupPeerId(chatId, resolvedThreadId), + conversationId: buildTelegramGroupPeerId(chatId, threadSpec), providerPolicy: authorization.authorizationCfg.channels?.telegram?.accounts?.[accountId]?.mentionPatterns, }, @@ -388,9 +388,7 @@ export function createTelegramInboundMedia({ primary.msg.chat.id, `⚠️ Received ${materializedCount} of ${entry.messages.length} images — ${skippedCount} could not be fetched and ${verb} skipped.`, { - ...buildTelegramThreadParams( - resolveTelegramMessageThreadSpec(primary.msg, entry.isForum), - ), + ...buildTelegramThreadParams(entry.threadSpec), reply_parameters: { message_id: primary.msg.message_id, allow_sending_without_reply: true, @@ -406,6 +404,7 @@ export function createTelegramInboundMedia({ promptContextMessageSelection: selection, storeAllowFrom: entry.storeAllowFrom, options: { + threadSpec: entry.threadSpec, ...(finalIngressMessageId != null ? { messageIdOverride: String(finalIngressMessageId) } : {}), @@ -439,8 +438,7 @@ export function createTelegramInboundMedia({ if (!mediaGroupId) { return false; } - const threadId = input.resolvedThreadId ?? input.dmThreadId; - const key = `media:${input.chatId}:${threadId ?? "main"}:${mediaGroupId}`; + const key = `media:${input.chatId}:${input.threadSpec.scope}:${input.threadSpec.id ?? "main"}:${mediaGroupId}`; const existing = buffer.get(key); const participant = createSpooledReplayParticipantForBufferedWork( `media-group:${key}:${input.msg.message_id}`, diff --git a/extensions/telegram/src/bot-handlers.inbound-pipeline.test.ts b/extensions/telegram/src/bot-handlers.inbound-pipeline.test.ts index 43d2ec2fd596..b5086e28eb91 100644 --- a/extensions/telegram/src/bot-handlers.inbound-pipeline.test.ts +++ b/extensions/telegram/src/bot-handlers.inbound-pipeline.test.ts @@ -28,12 +28,24 @@ describe("buildTelegramInboundDebounceKey", () => { ).toBe("telegram:default:12345:67890:forward"); }); - it("keeps direct topic thread ids in the conversation key", () => { - const topic100 = buildTelegramInboundDebounceConversationKey({ chatId: 7, threadId: 100 }); - const topic200 = buildTelegramInboundDebounceConversationKey({ chatId: 7, threadId: 200 }); + it("keeps scoped topic thread ids in the conversation key", () => { + const topic100 = buildTelegramInboundDebounceConversationKey({ + chatId: 7, + threadSpec: { id: 100, scope: "forum" }, + }); + const topic200 = buildTelegramInboundDebounceConversationKey({ + chatId: 7, + threadSpec: { id: 200, scope: "forum" }, + }); expect(topic100).toBe("7:topic:100"); expect(topic200).toBe("7:topic:200"); + expect( + buildTelegramInboundDebounceConversationKey({ + chatId: 7, + threadSpec: { id: 100, scope: "direct-messages" }, + }), + ).toBe("7:direct-topic:100"); expect( buildTelegramInboundDebounceKey({ accountId: "default", @@ -52,6 +64,8 @@ describe("buildTelegramInboundDebounceKey", () => { }); it("uses the chat id as the conversation key when no thread is present", () => { - expect(buildTelegramInboundDebounceConversationKey({ chatId: 7 })).toBe("7"); + expect( + buildTelegramInboundDebounceConversationKey({ chatId: 7, threadSpec: { scope: "none" } }), + ).toBe("7"); }); }); diff --git a/extensions/telegram/src/bot-handlers.inbound-pipeline.ts b/extensions/telegram/src/bot-handlers.inbound-pipeline.ts index 18c440b8d28c..e0d98948b749 100644 --- a/extensions/telegram/src/bot-handlers.inbound-pipeline.ts +++ b/extensions/telegram/src/bot-handlers.inbound-pipeline.ts @@ -195,19 +195,17 @@ function createTelegramInboundHandlers( const { dmPolicy, resolvedThreadId, - dmThreadId, storeAllowFrom, groupConfig, topicConfig, effectiveGroupAllow, + threadSpec, } = gate.context; const sessionState = resolveTelegramSessionState({ chatId: event.chatId, isGroup: event.isGroup, - isForum: event.isForum, - messageThreadId: event.messageThreadId, - resolvedThreadId, + threadSpec, botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(event.ctx.me), senderId: event.senderId, runtimeCfg: gate.context.cfg, @@ -236,8 +234,7 @@ function createTelegramInboundHandlers( chatId: event.chatId, isGroup: event.isGroup, isForum: event.isForum, - resolvedThreadId, - dmThreadId, + threadSpec, dmPolicy, storeAllowFrom, senderId: event.senderId, diff --git a/extensions/telegram/src/bot-handlers.inbound-processing.ts b/extensions/telegram/src/bot-handlers.inbound-processing.ts index f710a3ec598f..41b8e609d4ee 100644 --- a/extensions/telegram/src/bot-handlers.inbound-processing.ts +++ b/extensions/telegram/src/bot-handlers.inbound-processing.ts @@ -39,7 +39,7 @@ import { resolveMedia } from "./bot/delivery.resolve-media.js"; import { buildTelegramThreadParams, getTelegramTextParts, - resolveTelegramMessageThreadSpec, + type TelegramThreadSpec, resolveTelegramPrimaryMedia, } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; @@ -57,8 +57,7 @@ type TelegramInboundMessage = { chatId: number; isGroup: boolean; isForum: boolean; - resolvedThreadId?: number; - dmThreadId?: number; + threadSpec: TelegramThreadSpec; dmPolicy: DmPolicy; storeAllowFrom: string[]; senderId: string; @@ -129,8 +128,7 @@ export function createTelegramInboundProcessing({ chatId, isGroup, isForum, - resolvedThreadId, - dmThreadId, + threadSpec, dmPolicy, storeAllowFrom, senderId, @@ -145,6 +143,10 @@ export function createTelegramInboundProcessing({ promptContextAmbientWatermark, dispatchDedupeClaims, } = params; + const resolvedThreadId = + threadSpec.scope === "forum" || threadSpec.scope === "direct-messages" + ? threadSpec.id + : undefined; const messageText = getTelegramTextParts(msg).text; const botUsername = ctx.me?.username; @@ -179,8 +181,7 @@ export function createTelegramInboundProcessing({ ctx, msg, chatId, - resolvedThreadId, - dmThreadId, + threadSpec, storeAllowFrom, isAbortControlMessage, isAuthorizedAbortControlMessage, @@ -201,8 +202,7 @@ export function createTelegramInboundProcessing({ chatId, isGroup, isForum, - resolvedThreadId, - dmThreadId, + threadSpec, storeAllowFrom, senderId, effectiveGroupAllow, @@ -225,8 +225,7 @@ export function createTelegramInboundProcessing({ chatId, isGroup, isForum, - resolvedThreadId, - dmThreadId, + threadSpec, senderId, effectiveGroupAllow, effectiveDmAllow, @@ -252,9 +251,7 @@ export function createTelegramInboundProcessing({ } } catch (mediaErr) { const replayingSpooledUpdate = isTelegramSpooledReplayUpdate(ctx.update); - const warningThreadParams = buildTelegramThreadParams( - resolveTelegramMessageThreadSpec(msg, isForum), - ); + const warningThreadParams = buildTelegramThreadParams(threadSpec); if (mediaRuntime.abortSignal?.aborted && isDurablyRetryableInboundMediaError(mediaErr)) { // Abort mid-media-resolution must stay retryable for live updates too; // a clean claim release would settle the update as handled and silently @@ -322,7 +319,7 @@ export function createTelegramInboundProcessing({ : []; const conversationKey = buildTelegramInboundDebounceConversationKey({ chatId, - threadId: resolvedThreadId ?? dmThreadId, + threadSpec, }); const debounceLane = resolveTelegramDebounceLane(msg); const debounceKey = senderId @@ -354,6 +351,7 @@ export function createTelegramInboundProcessing({ debounceKey: isAbortControlMessage ? null : debounceKey, debounceLane, botUsername, + threadSpec, ...promptContextBoundaryOptions(promptContextMinTimestampMs, promptContextAmbientWatermark), dispatchDedupeClaims, channelIngressResolvers: [channelIngressResolver], diff --git a/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts b/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts index b79a162db5ba..1acfec722f2e 100644 --- a/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts +++ b/extensions/telegram/src/bot-handlers.message-context.runtime.test.ts @@ -82,8 +82,7 @@ describe("resolveCachedMessageThreadSpec", () => { const session = sessionRuntime.resolveTelegramSessionState({ chatId: CHAT_ID, isGroup: true, - isForum: true, - messageThreadId: TOPIC_ID, + threadSpec: { id: TOPIC_ID, scope: "forum" }, senderId: 10, runtimeCfg: cfg, }); diff --git a/extensions/telegram/src/bot-handlers.message-context.ts b/extensions/telegram/src/bot-handlers.message-context.ts index 62448a4ec12a..2e73b50fa97e 100644 --- a/extensions/telegram/src/bot-handlers.message-context.ts +++ b/extensions/telegram/src/bot-handlers.message-context.ts @@ -23,7 +23,6 @@ import { buildSenderName, getTelegramTextParts, resolveTelegramPrimaryMedia, - resolveTelegramForumThreadId, type TelegramThreadSpec, } from "./bot/helpers.js"; import type { TelegramContext } from "./bot/types.js"; @@ -81,9 +80,7 @@ export type TelegramSessionState = { export type ResolveTelegramSessionStateParams = { chatId: number | string; isGroup: boolean; - isForum: boolean; - messageThreadId?: number; - resolvedThreadId?: number; + threadSpec: TelegramThreadSpec; botHasTopicsEnabled?: boolean; senderId?: string | number; runtimeCfg: OpenClawConfig; @@ -187,14 +184,8 @@ export function createTelegramMessageSessionRuntime({ const resolveTelegramSessionState = ( params: ResolveTelegramSessionStateParams, ): TelegramSessionState => { - const resolvedThreadId = - params.resolvedThreadId ?? - resolveTelegramForumThreadId({ - isForum: params.isForum, - messageThreadId: params.messageThreadId, - }); - const dmThreadId = !params.isGroup ? params.messageThreadId : undefined; - const topicThreadId = resolvedThreadId ?? dmThreadId; + const dmThreadId = params.threadSpec.scope === "dm" ? params.threadSpec.id : undefined; + const topicThreadId = params.threadSpec.id; const { topicConfig } = resolveTelegramGroupConfig( params.chatId, topicThreadId, @@ -205,8 +196,7 @@ export function createTelegramMessageSessionRuntime({ accountId, chatId: params.chatId, isGroup: params.isGroup, - resolvedThreadId, - replyThreadId: topicThreadId, + threadSpec: params.threadSpec, senderId: params.senderId, topicAgentId: topicConfig?.agentId, }); diff --git a/extensions/telegram/src/bot-handlers.message-session.runtime.test.ts b/extensions/telegram/src/bot-handlers.message-session.runtime.test.ts index 40fa1abe0db7..515dc8837a2a 100644 --- a/extensions/telegram/src/bot-handlers.message-session.runtime.test.ts +++ b/extensions/telegram/src/bot-handlers.message-session.runtime.test.ts @@ -35,8 +35,7 @@ describe("createTelegramMessageSessionRuntime", () => { const state = resolveTelegramSessionState({ chatId: 12345, isGroup: false, - isForum: false, - messageThreadId: 99, + threadSpec: { id: 99, scope: "dm" }, botHasTopicsEnabled: true, senderId: 12345, runtimeCfg: {}, diff --git a/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts b/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts index e81ff396fa7b..ce720fe81b08 100644 --- a/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts +++ b/extensions/telegram/src/bot-handlers.reaction.runtime.test.ts @@ -7,7 +7,7 @@ import { createTelegramEventBindings } from "./bot-handlers.event-bindings.js"; import { createTelegramHandlerAuthorization } from "./bot-handlers.inbound-authorization.js"; import { createTelegramMessagePipeline } from "./bot-handlers.message-pipeline.js"; import type { RegisterTelegramHandlerParams } from "./bot-handlers.types.js"; -import type { TelegramThreadSpec } from "./bot/helpers.js"; +import type { TelegramThreadSpec } from "./thread-spec.js"; const FIRE_EMOJI = "\u{1F525}"; const FORUM_CHAT_ID = 5678; @@ -243,7 +243,7 @@ describe("registerTelegramReactionHandler forum topic recovery", () => { expect(enqueueSystemEvent).toHaveBeenCalledTimes(1); expect(String(systemEventOptions().sessionKey)).toContain("direct-topic-agent"); expect(String(systemEventOptions().sessionKey)).toContain( - `telegram:group:${FORUM_CHAT_ID}:topic:${FORUM_TOPIC_ID}`, + `telegram:group:${FORUM_CHAT_ID}:direct-topic:${FORUM_TOPIC_ID}`, ); }); diff --git a/extensions/telegram/src/bot-message-context.body.test.ts b/extensions/telegram/src/bot-message-context.body.test.ts index ff80e81eef8f..5916a7f20576 100644 --- a/extensions/telegram/src/bot-message-context.body.test.ts +++ b/extensions/telegram/src/bot-message-context.body.test.ts @@ -147,6 +147,7 @@ async function resolveBody(overrides: Partial = {}) { chatId, senderId: String(chatId), senderUsername: "", + threadSpec: { scope: "none" }, effectiveGroupAllow: normalizeAllowFrom([]), effectiveDmAllow: normalizeAllowFrom([]), requireMention: false, diff --git a/extensions/telegram/src/bot-message-context.body.ts b/extensions/telegram/src/bot-message-context.body.ts index fac61e09ea9c..8f0c501a5106 100644 --- a/extensions/telegram/src/bot-message-context.body.ts +++ b/extensions/telegram/src/bot-message-context.body.ts @@ -51,7 +51,11 @@ import { resolveTelegramRichMessagePlaceholder, resolveTelegramRichMessageText, } from "./bot/body-helpers.js"; -import { buildTelegramGroupPeerId, buildTelegramInboundOriginTarget } from "./bot/helpers.js"; +import { + buildTelegramGroupPeerId, + buildTelegramInboundOriginTarget, + type TelegramThreadSpec, +} from "./bot/helpers.js"; import { renderTelegramTextEntities } from "./bot/inbound-text-entities.js"; import type { TelegramContext } from "./bot/types.js"; import { isTelegramForumServiceMessage } from "./forum-service-message.js"; @@ -146,6 +150,7 @@ export async function resolveTelegramInboundBody(params: { sessionKey?: string; resolvedThreadId?: number; replyThreadId?: number; + threadSpec: TelegramThreadSpec; originatingTo?: string; routeAgentId?: string; effectiveGroupAllow: NormalizedAllowFrom; @@ -172,6 +177,7 @@ export async function resolveTelegramInboundBody(params: { sessionKey, resolvedThreadId, replyThreadId, + threadSpec, originatingTo: providedOriginatingTo, routeAgentId, effectiveGroupAllow, @@ -188,7 +194,7 @@ export async function resolveTelegramInboundBody(params: { const botUsername = normalizeOptionalLowercaseString(primaryCtx.me?.username); const mentionRegexes = buildMentionRegexes(cfg, routeAgentId, { provider: "telegram", - conversationId: isGroup ? buildTelegramGroupPeerId(chatId, resolvedThreadId) : String(chatId), + conversationId: isGroup ? buildTelegramGroupPeerId(chatId, threadSpec) : String(chatId), providerPolicy: providerMentionPatterns, }); const messageTextParts = getTelegramTextParts(msg); @@ -224,8 +230,9 @@ export async function resolveTelegramInboundBody(params: { includeDmAllowForGroupCommands: false, }); const commandAuthorized = commandGate.authorized; - const historyKey = isGroup ? buildTelegramGroupPeerId(chatId, resolvedThreadId) : undefined; - const originatingTo = providedOriginatingTo ?? buildTelegramInboundOriginTarget(chatId); + const historyKey = isGroup ? buildTelegramGroupPeerId(chatId, threadSpec) : undefined; + const originatingTo = + providedOriginatingTo ?? buildTelegramInboundOriginTarget(chatId, threadSpec); const primaryMedia = resolveTelegramPrimaryMedia(msg); const nativeMediaFacts = diff --git a/extensions/telegram/src/bot-message-context.dm-threads.test.ts b/extensions/telegram/src/bot-message-context.dm-threads.test.ts index 1999ffec3fff..0bcc8c8eec67 100644 --- a/extensions/telegram/src/bot-message-context.dm-threads.test.ts +++ b/extensions/telegram/src/bot-message-context.dm-threads.test.ts @@ -250,7 +250,10 @@ describe("buildTelegramMessageContext group sessions without forum", () => { expect(ctx?.ctxPayload.MessageThreadId).toBe(77); expect(ctx?.ctxPayload.OriginatingTo).toBe("telegram:-1001234567890:direct-topic:77"); - expect(ctx?.ctxPayload.SessionKey).toBe("agent:main:telegram:group:-1001234567890:topic:77"); + expect(ctx?.ctxPayload.From).toBe("telegram:group:-1001234567890:direct-topic:77"); + expect(ctx?.ctxPayload.SessionKey).toBe( + "agent:main:telegram:group:-1001234567890:direct-topic:77", + ); expect(ctx?.turn.record.updateLastRoute).toMatchObject({ to: "telegram:-1001234567890:direct-topic:77", threadId: "77", diff --git a/extensions/telegram/src/bot-message-context.session.ts b/extensions/telegram/src/bot-message-context.session.ts index 40f3e770ffbc..af2a8602e010 100644 --- a/extensions/telegram/src/bot-message-context.session.ts +++ b/extensions/telegram/src/bot-message-context.session.ts @@ -621,9 +621,7 @@ export async function buildTelegramInboundContextPayload(params: { replyHead?.body ?? visibleReplyTarget?.body ?? (replyTargetMedia ? formatMediaPlaceholderText([replyTargetMedia]) : undefined); - const telegramFrom = isGroup - ? buildTelegramGroupFrom(chatId, resolvedThreadId) - : `telegram:${chatId}`; + const telegramFrom = isGroup ? buildTelegramGroupFrom(chatId, threadSpec) : `telegram:${chatId}`; const telegramTo = buildTelegramInboundOriginTarget(chatId, threadSpec); const locationContext = locationData ? toLocationContext(locationData) : undefined; const telegramUpdate = primaryCtx.update; diff --git a/extensions/telegram/src/bot-message-context.thread-binding.test.ts b/extensions/telegram/src/bot-message-context.thread-binding.test.ts index fee945d31c2e..de7c675d64e9 100644 --- a/extensions/telegram/src/bot-message-context.thread-binding.test.ts +++ b/extensions/telegram/src/bot-message-context.thread-binding.test.ts @@ -105,8 +105,7 @@ describe("buildTelegramMessageContext thread binding override", () => { expect(routeArgs.accountId).toBe("default"); expect(routeArgs.chatId).toBe(-100200300); expect(routeArgs.isGroup).toBe(true); - expect(routeArgs.resolvedThreadId).toBe(77); - expect(routeArgs.replyThreadId).toBe(77); + expect(routeArgs.threadSpec).toEqual({ id: 77, scope: "forum" }); expect(routeArgs.senderId).toBe("42"); expect(ctx?.ctxPayload?.SessionKey).toBe("agent:codex-acp:session-1"); expect(ctx?.turn.record.updateLastRoute).toBeUndefined(); @@ -175,8 +174,7 @@ describe("buildTelegramMessageContext thread binding override", () => { expect(routeArgs.accountId).toBe("work"); expect(routeArgs.chatId).toBe(-100200300); expect(routeArgs.isGroup).toBe(true); - expect(routeArgs.resolvedThreadId).toBe(77); - expect(routeArgs.replyThreadId).toBe(77); + expect(routeArgs.threadSpec).toEqual({ id: 77, scope: "forum" }); expect(routeArgs.senderId).toBe("42"); expect(ctx?.route.accountId).toBe("work"); expect(ctx?.route.matchedBy).toBe("binding.channel"); @@ -207,8 +205,7 @@ describe("buildTelegramMessageContext thread binding override", () => { expect(routeArgs.accountId).toBe("default"); expect(routeArgs.chatId).toBe(1234); expect(routeArgs.isGroup).toBe(false); - expect(routeArgs.resolvedThreadId).toBeUndefined(); - expect(routeArgs.replyThreadId).toBeUndefined(); + expect(routeArgs.threadSpec).toEqual({ scope: "dm" }); expect(routeArgs.senderId).toBe("42"); expect(ctx?.ctxPayload?.SessionKey).toBe("agent:codex-acp:session-dm"); }); @@ -237,8 +234,7 @@ describe("buildTelegramMessageContext thread binding override", () => { const routeArgs = expectRouteArgs(); expect(routeArgs.chatId).toBe(1234); expect(routeArgs.isGroup).toBe(false); - expect(routeArgs.resolvedThreadId).toBeUndefined(); - expect(routeArgs.replyThreadId).toBe(77); + expect(routeArgs.threadSpec).toEqual({ id: 77, scope: "dm" }); expect(ctx?.ctxPayload?.MessageThreadId).toBe(77); }); }); diff --git a/extensions/telegram/src/bot-message-context.ts b/extensions/telegram/src/bot-message-context.ts index bbd6d2186ec3..6d3b50ca858f 100644 --- a/extensions/telegram/src/bot-message-context.ts +++ b/extensions/telegram/src/bot-message-context.ts @@ -171,7 +171,7 @@ export const buildTelegramMessageContext = async ({ isTopicMessage: msg.is_topic_message, getChat: getChatApi, }); - const threadSpec = resolveTelegramMessageThreadSpec(msg, isForum); + const threadSpec = options?.threadSpec ?? resolveTelegramMessageThreadSpec(msg, isForum); const resolvedThreadId = threadSpec.scope === "forum" || threadSpec.scope === "direct-messages" ? threadSpec.id @@ -250,8 +250,7 @@ export const buildTelegramMessageContext = async ({ accountId: account.accountId, chatId, isGroup, - resolvedThreadId, - replyThreadId, + threadSpec, senderId, topicAgentId: topicConfig?.agentId, }); @@ -475,6 +474,7 @@ export const buildTelegramMessageContext = async ({ senderUsername, resolvedThreadId, replyThreadId, + threadSpec, originatingTo, routeAgentId: route.agentId, sessionKey, diff --git a/extensions/telegram/src/bot-message-context.types.ts b/extensions/telegram/src/bot-message-context.types.ts index d85824cd12b7..bb7b0971f94c 100644 --- a/extensions/telegram/src/bot-message-context.types.ts +++ b/extensions/telegram/src/bot-message-context.types.ts @@ -15,6 +15,7 @@ import type { import type { HistoryEntry } from "openclaw/plugin-sdk/reply-history"; import type { MsgContext } from "openclaw/plugin-sdk/reply-runtime"; import type { TelegramMediaKind } from "./bot/body-helpers.js"; +import type { TelegramThreadSpec } from "./bot/helpers.js"; import type { StickerMetadata, TelegramContext } from "./bot/types.js"; import type { TelegramReplyChainEntry } from "./message-cache.js"; import type { TelegramSendChatActionHandler } from "./sendchataction-401-backoff.js"; @@ -32,6 +33,7 @@ export type TelegramChannelIngressResolver = ( ) => Promise; export type TelegramMessageContextOptions = { + threadSpec?: TelegramThreadSpec; commandSource?: "text" | "native"; forceWasMentioned?: boolean; messageIdOverride?: string; diff --git a/extensions/telegram/src/bot-message-dispatch-context.ts b/extensions/telegram/src/bot-message-dispatch-context.ts index a7a1cf818345..ef27685f39e4 100644 --- a/extensions/telegram/src/bot-message-dispatch-context.ts +++ b/extensions/telegram/src/bot-message-dispatch-context.ts @@ -169,7 +169,7 @@ export function resolveDispatchTelegramContext(params: { threadSpec, ); const recoveredFrom = params.context.isGroup - ? buildTelegramGroupFrom(params.context.chatId, threadSpec.id) + ? buildTelegramGroupFrom(params.context.chatId, threadSpec) : params.context.ctxPayload.From; const recoveredUpdateLastRoute = params.context.turn.record.updateLastRoute && threadSpec.id != null @@ -180,7 +180,7 @@ export function resolveDispatchTelegramContext(params: { } : params.context.turn.record.updateLastRoute; const recoveredHistoryKey = params.context.isGroup - ? buildTelegramGroupPeerId(params.context.chatId, threadSpec.id) + ? buildTelegramGroupPeerId(params.context.chatId, threadSpec) : params.context.historyKey; const recoveredHistoryEntries = recoveredHistoryKey && params.context.historyLimit > 0 diff --git a/extensions/telegram/src/bot-message-dispatch-reply.ts b/extensions/telegram/src/bot-message-dispatch-reply.ts index e45117f0d092..d24e2efd35ae 100644 --- a/extensions/telegram/src/bot-message-dispatch-reply.ts +++ b/extensions/telegram/src/bot-message-dispatch-reply.ts @@ -539,7 +539,7 @@ export function handleReplyError( scopeKey: buildTelegramErrorScopeKey({ accountId: turn.context.route.accountId, chatId: turn.context.chatId, - threadId: turn.context.threadSpec.id, + threadSpec: turn.context.threadSpec, }), cooldownMs: errorPolicy.cooldownMs, errorMessage: String(err), diff --git a/extensions/telegram/src/bot-native-command-dispatch.ts b/extensions/telegram/src/bot-native-command-dispatch.ts index d21cca3c9680..2ac8fe605005 100644 --- a/extensions/telegram/src/bot-native-command-dispatch.ts +++ b/extensions/telegram/src/bot-native-command-dispatch.ts @@ -191,6 +191,7 @@ async function resolveTelegramCommandAuth(params: { accountId, chatId, isGroup, + threadSpec, senderId, senderUsername, }) @@ -236,7 +237,7 @@ async function resolveTelegramCommandAuth(params: { accountId, chatId, isGroup, - resolvedThreadId, + threadSpec, senderId, senderUsername, }) @@ -246,7 +247,7 @@ async function resolveTelegramCommandAuth(params: { accountId, chatId, isGroup, - resolvedThreadId, + threadSpec, senderId, senderUsername, }); @@ -359,6 +360,7 @@ async function resolveTelegramCommandAuth(params: { senderUsername, groupConfig, topicConfig, + threadSpec, commandAuthorized, senderIsOwner: ownerAccess.senderIsOwner, }; @@ -395,14 +397,12 @@ export async function prepareTelegramCommandDispatch( if (!auth) { return null; } - const threadSpec = resolveTelegramMessageThreadSpec(params.msg, auth.isForum); const { route, bindingMode } = resolveTelegramConversationRoute({ cfg: runtimeCfg, accountId: params.accountId, chatId: auth.chatId, isGroup: auth.isGroup, - resolvedThreadId: auth.resolvedThreadId, - replyThreadId: threadSpec.id, + threadSpec: auth.threadSpec, senderId: auth.senderId, topicAgentId: auth.topicConfig?.agentId, }); @@ -423,7 +423,7 @@ export async function prepareTelegramCommandDispatch( params.bot.api.sendMessage( auth.chatId, "Configured ACP binding is unavailable right now. Please try again.", - buildTelegramThreadParams(threadSpec) ?? {}, + buildTelegramThreadParams(auth.threadSpec) ?? {}, ), }); return null; @@ -446,7 +446,7 @@ export async function prepareTelegramCommandDispatch( chatId: auth.chatId, isGroup: auth.isGroup, senderId: auth.senderId, - dmThreadId: threadSpec.scope === "dm" ? threadSpec.id : undefined, + dmThreadId: auth.threadSpec.scope === "dm" ? auth.threadSpec.id : undefined, botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(params.botUser), }); const buildDeliveryBaseOptions = (keys?: { @@ -468,7 +468,7 @@ export async function prepareTelegramCommandDispatch( mediaMaxBytes: params.mediaMaxBytes, replyToMode: turnSettings.replyToMode, textLimit: turnSettings.textLimit, - thread: threadSpec, + thread: auth.threadSpec, tableMode, chunkMode, linkPreview: runtimeTelegramCfg.linkPreview, @@ -481,8 +481,8 @@ export async function prepareTelegramCommandDispatch( runtimeTelegramCfg, turnSettings, ...auth, - threadSpec, - threadParams: buildTelegramThreadParams(threadSpec), + threadSpec: auth.threadSpec, + threadParams: buildTelegramThreadParams(auth.threadSpec), route, mediaLocalRoots, targetSessionKey, @@ -541,7 +541,7 @@ export async function dispatchTelegramBuiltinTurn(params: { CommandBody: params.prompt, CommandArgs: params.commandArgs, From: dispatch.isGroup - ? buildTelegramGroupFrom(dispatch.chatId, dispatch.resolvedThreadId) + ? buildTelegramGroupFrom(dispatch.chatId, dispatch.threadSpec) : `telegram:${dispatch.chatId}`, To: `slash:${dispatch.senderId || dispatch.chatId}`, ChatType: dispatch.isGroup ? "group" : "direct", diff --git a/extensions/telegram/src/bot-native-command-plugins.ts b/extensions/telegram/src/bot-native-command-plugins.ts index 232220f43690..5aa93000ec63 100644 --- a/extensions/telegram/src/bot-native-command-plugins.ts +++ b/extensions/telegram/src/bot-native-command-plugins.ts @@ -182,7 +182,7 @@ export async function executeTelegramPluginCommand( sessionKey: dispatch.targetSessionKey, }); const from = dispatch.isGroup - ? buildTelegramGroupFrom(dispatch.chatId, dispatch.threadSpec.id) + ? buildTelegramGroupFrom(dispatch.chatId, dispatch.threadSpec) : `telegram:${dispatch.chatId}`; const to = dispatch.threadSpec.scope === "direct-messages" diff --git a/extensions/telegram/src/bot.create-telegram-bot.test.ts b/extensions/telegram/src/bot.create-telegram-bot.test.ts index e45dabe50c89..7d3797fdac2f 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.test.ts @@ -3910,7 +3910,7 @@ describe("createTelegramBot", () => { expect(payload.MessageThreadId).toBe(77); expect(payload.OriginatingTo).toBe(`telegram:${chatId}:direct-topic:77`); expect(payload.SessionKey).toContain("agent:channel-topic-agent:"); - expect(payload.SessionKey).toContain(":topic:77"); + expect(payload.SessionKey).toContain(":direct-topic:77"); }); it.each([ @@ -4234,7 +4234,7 @@ describe("createTelegramBot", () => { accountId: "default", chatId: -1001234567890, isGroup: true, - resolvedThreadId: 99, + threadSpec: { id: 99, scope: "forum" }, }); expect(result.route.sessionKey).toContain(testCase.expectedSessionKeyFragment); @@ -4701,19 +4701,18 @@ describe("createTelegramBot", () => { isForum: true, messageThreadId, }); - const resolvedThreadId = threadSpec.scope === "forum" ? threadSpec.id : undefined; const route = resolveTelegramConversationRoute({ cfg: {}, accountId: "default", chatId: -1001234567890, isGroup: true, - resolvedThreadId, + threadSpec, }); const expectedGroupFrom = `telegram:group:-1001234567890:topic:${expectedTopicId}`; expect(route.route.sessionKey).toContain(expectedGroupFrom); - expect(buildTelegramGroupFrom(-1001234567890, resolvedThreadId)).toBe(expectedGroupFrom); - expect(buildTypingThreadParams(resolvedThreadId)).toEqual({ + expect(buildTelegramGroupFrom(-1001234567890, threadSpec)).toBe(expectedGroupFrom); + expect(buildTypingThreadParams(threadSpec.id)).toEqual({ message_thread_id: expectedTopicId, }); }); @@ -4738,22 +4737,19 @@ describe("createTelegramBot", () => { isForum, messageThreadId: undefined, }); - const resolvedThreadId = threadSpec.scope === "forum" ? threadSpec.id : undefined; const route = resolveTelegramConversationRoute({ cfg: {}, accountId: "default", chatId, isGroup: true, - resolvedThreadId, + threadSpec, }); expect(getChatSpy).toHaveBeenCalledOnce(); expect(getChatSpy).toHaveBeenCalledWith(chatId); expect(route.route.sessionKey).toContain(`telegram:group:${chatId}:topic:1`); - expect(buildTelegramGroupFrom(chatId, resolvedThreadId)).toBe( - `telegram:group:${chatId}:topic:1`, - ); - expect(buildTypingThreadParams(resolvedThreadId)).toEqual({ message_thread_id: 1 }); + expect(buildTelegramGroupFrom(chatId, threadSpec)).toBe(`telegram:group:${chatId}:topic:1`); + expect(buildTypingThreadParams(threadSpec.id)).toEqual({ message_thread_id: 1 }); }); const allowFromEdgeCases: MessagePolicyCase[] = [ makeMessagePolicyCase({ diff --git a/extensions/telegram/src/bot.test.ts b/extensions/telegram/src/bot.test.ts index 1c15d55c1d2e..64dc85f09d3e 100644 --- a/extensions/telegram/src/bot.test.ts +++ b/extensions/telegram/src/bot.test.ts @@ -539,6 +539,7 @@ async function writeDirectTelegramTranscriptMessages(params: { accountId: "default", chatId: params.chatId, isGroup: false, + threadSpec: { scope: "none" }, senderId: params.senderId, }).route; const sessionKey = resolveTelegramConversationBaseSessionKey({ @@ -2608,6 +2609,7 @@ describe("createTelegramBot", () => { accountId: "default", chatId: 1234, isGroup: false, + threadSpec: { scope: "none" }, senderId: 9, }).route; const sessionKey = resolveTelegramConversationBaseSessionKey({ @@ -2715,6 +2717,7 @@ describe("createTelegramBot", () => { accountId: "default", chatId: 1234, isGroup: false, + threadSpec: { scope: "none" }, senderId: 9, }).route; const sessionKey = resolveTelegramConversationBaseSessionKey({ diff --git a/extensions/telegram/src/bot/helpers.ts b/extensions/telegram/src/bot/helpers.ts index 45ab7ac63416..017609404e51 100644 --- a/extensions/telegram/src/bot/helpers.ts +++ b/extensions/telegram/src/bot/helpers.ts @@ -29,6 +29,8 @@ import { } from "../bot-access.js"; import { normalizeTelegramReplyToMessageId } from "../outbound-params.js"; import { resolveTelegramPreviewStreamMode } from "../preview-streaming.js"; +import type { TelegramThreadSpec } from "../thread-spec.js"; +import { buildTelegramConversationId } from "../topic-conversation.js"; import { buildSenderLabel, buildSenderName, @@ -52,6 +54,7 @@ export type { TelegramMediaKind, TelegramTextEntity, } from "./body-helpers.js"; +export type { TelegramThreadSpec } from "../thread-spec.js"; export { buildSenderLabel, buildSenderName, @@ -97,12 +100,6 @@ function hadUnsafeTelegramText(raw: unknown, sanitized: string): boolean { return typeof raw === "string" && raw.trim().length > 0 && sanitized.trim().length === 0; } -export type TelegramThreadSpec = { - id?: number; - /** dm is the historical bot-private topic scope. */ - scope: "direct-messages" | "dm" | "forum" | "none"; -}; - type TelegramThreadParams = { direct_messages_topic_id?: number; message_thread_id?: number; @@ -497,13 +494,12 @@ export function buildTelegramRoutingTarget( ): string { const base = `telegram:${chatId}`; const threadParams = buildTelegramThreadParams(thread); - if (typeof threadParams?.direct_messages_topic_id === "number") { + if (threadParams?.direct_messages_topic_id != null) { return `${base}:direct-topic:${threadParams.direct_messages_topic_id}`; } - if (typeof threadParams?.message_thread_id !== "number") { - return base; - } - return `${base}:topic:${threadParams.message_thread_id}`; + return threadParams?.message_thread_id != null + ? `${base}:topic:${threadParams.message_thread_id}` + : base; } /** @@ -537,12 +533,19 @@ export function resolveTelegramStreamMode(telegramCfg?: { return resolveTelegramPreviewStreamMode(telegramCfg); } -export function buildTelegramGroupPeerId(chatId: number | string, messageThreadId?: number) { - return messageThreadId != null ? `${chatId}:topic:${messageThreadId}` : String(chatId); +export function buildTelegramGroupPeerId( + chatId: number | string, + thread?: number | TelegramThreadSpec, +) { + const threadSpec = typeof thread === "number" ? { id: thread, scope: "forum" as const } : thread; + return buildTelegramConversationId({ chatId, thread: threadSpec ?? { scope: "none" } }); } -export function buildTelegramGroupFrom(chatId: number | string, messageThreadId?: number) { - return `telegram:group:${buildTelegramGroupPeerId(chatId, messageThreadId)}`; +export function buildTelegramGroupFrom( + chatId: number | string, + thread?: number | TelegramThreadSpec, +) { + return `telegram:group:${buildTelegramGroupPeerId(chatId, thread)}`; } export function isTelegramCommandsAllowFromConfigured(cfg: OpenClawConfig): boolean { @@ -559,7 +562,7 @@ export function resolveTelegramCommandAuthorization(params: { accountId: string; chatId: number; isGroup: boolean; - resolvedThreadId?: number; + threadSpec: TelegramThreadSpec; senderId?: string; senderUsername?: string; }): CommandAuthorization { @@ -571,7 +574,7 @@ export function resolveTelegramCommandAuthorization(params: { AccountId: params.accountId, ChatType: params.isGroup ? "group" : "direct", From: params.isGroup - ? buildTelegramGroupFrom(params.chatId, params.resolvedThreadId) + ? buildTelegramGroupFrom(params.chatId, params.threadSpec) : `telegram:${params.chatId}`, SenderId: params.senderId || undefined, SenderUsername: params.senderUsername || undefined, diff --git a/extensions/telegram/src/channel.ts b/extensions/telegram/src/channel.ts index ae8e19340869..bac3ec34790f 100644 --- a/extensions/telegram/src/channel.ts +++ b/extensions/telegram/src/channel.ts @@ -54,7 +54,7 @@ import { writeCachedTelegramBotInfo, } from "./bot-info-cache.js"; import type { TelegramBotInfo } from "./bot-info.js"; -import { buildTelegramGroupPeerId } from "./bot/helpers.js"; +import { buildTelegramRoutingTarget, type TelegramThreadSpec } from "./bot/helpers.js"; import { telegramMessageActions as telegramMessageActionsImpl } from "./channel-actions.js"; import { findTelegramTokenOwnerAccountId, @@ -95,7 +95,7 @@ import { telegramSetupWizard } from "./setup-surface.js"; import { createTelegramPluginBase } from "./shared.js"; import { withTelegramStartupProbeSlot } from "./startup-probe-limiter.js"; import { collectTelegramStatusIssues } from "./status-issues.js"; -import { parseTelegramTarget } from "./targets.js"; +import { normalizeTelegramChatId, parseTelegramTarget } from "./targets.js"; import { createTelegramThreadBindingManager, setTelegramThreadBindingIdleTimeoutBySessionKey, @@ -103,7 +103,10 @@ import { } from "./thread-bindings.js"; import { buildTelegramThreadingToolContext } from "./threading-tool-context.js"; import { resolveTelegramToken } from "./token.js"; -import { parseTelegramTopicConversation } from "./topic-conversation.js"; +import { + parseTelegramTopicConversation, + buildTelegramConversationId, +} from "./topic-conversation.js"; type TelegramSendFn = typeof import("./send.js").sendMessageTelegram; @@ -375,25 +378,35 @@ function targetsMatchTelegramReplySuppression(params: { }): boolean { const origin = parseTelegramTarget(params.originTarget); const target = parseTelegramTarget(params.targetKey); - const originThreadId = - origin.messageThreadId != null && normalizeOptionalString(String(origin.messageThreadId)) - ? normalizeOptionalString(String(origin.messageThreadId)) - : undefined; - const targetThreadId = - normalizeOptionalString(params.targetThreadId) || - (target.messageThreadId != null && normalizeOptionalString(String(target.messageThreadId)) - ? normalizeOptionalString(String(target.messageThreadId)) - : undefined); - if ( - normalizeOptionalLowercaseString(origin.chatId) !== - normalizeOptionalLowercaseString(target.chatId) - ) { - return false; + const originConversation = buildTelegramConversationId({ + chatId: origin.chatId, + thread: resolveTelegramTargetThread(origin) ?? { scope: "none" }, + }); + const targetConversation = buildTelegramConversationId({ + chatId: target.chatId, + thread: resolveTelegramTargetThread(target, normalizeOptionalString(params.targetThreadId)) ?? { + scope: "none", + }, + }); + return ( + normalizeOptionalLowercaseString(originConversation) === + normalizeOptionalLowercaseString(targetConversation) + ); +} + +function resolveTelegramTargetThread( + target: ReturnType, + fallbackThreadId?: string | number | null, +): TelegramThreadSpec | undefined { + if (target.directMessagesTopicId != null) { + return { id: target.directMessagesTopicId, scope: "direct-messages" }; } - if (originThreadId && targetThreadId) { - return originThreadId === targetThreadId; + const forumThreadId = target.messageThreadId ?? fallbackThreadId; + if (forumThreadId == null) { + return undefined; } - return originThreadId == null && targetThreadId == null; + const id = Number(forumThreadId); + return Number.isFinite(id) ? { id, scope: "forum" } : undefined; } function resolveTelegramCommandConversation(params: { @@ -402,37 +415,46 @@ function resolveTelegramCommandConversation(params: { commandTo?: string; fallbackTo?: string; }) { - const chatId = [params.originatingTo, params.commandTo, params.fallbackTo] - .map((candidate) => { - const trimmed = normalizeOptionalString(candidate) ?? ""; - return trimmed ? (normalizeOptionalString(parseTelegramTarget(trimmed).chatId) ?? "") : ""; - }) - .find((candidate) => candidate.length > 0); - if (!chatId) { - return null; + for (const candidate of [params.originatingTo, params.commandTo, params.fallbackTo]) { + if (!candidate) { + continue; + } + const parsedTarget = parseTelegramTarget(candidate); + const chatId = normalizeTelegramChatId(parsedTarget.chatId); + if (!chatId) { + continue; + } + const thread = resolveTelegramTargetThread(parsedTarget, params.threadId); + const conversationId = buildTelegramConversationId({ + chatId, + thread: thread ?? { scope: "none" }, + }); + if (conversationId !== chatId || !chatId.startsWith("-")) { + return { conversationId, parentConversationId: chatId }; + } } - if (params.threadId) { - return { - conversationId: `${chatId}:topic:${params.threadId}`, - parentConversationId: chatId, - }; - } - if (chatId.startsWith("-")) { - return null; - } - return { - conversationId: chatId, - parentConversationId: chatId, - }; + return null; } function resolveTelegramInboundConversation(params: { to?: string; conversationId?: string; + parentConversationId?: string; threadId?: string | number; }) { - const rawTarget = - normalizeOptionalString(params.to) ?? normalizeOptionalString(params.conversationId) ?? ""; + const parsedConversation = + params.conversationId && + parseTelegramTopicConversation({ + conversationId: params.conversationId, + parentConversationId: params.parentConversationId, + }); + if (parsedConversation) { + return { + conversationId: parsedConversation.canonicalConversationId, + parentConversationId: parsedConversation.chatId, + }; + } + const rawTarget = params.to || params.parentConversationId || params.conversationId || ""; if (!rawTarget) { return null; } @@ -441,27 +463,11 @@ function resolveTelegramInboundConversation(params: { if (!chatId) { return null; } - const threadId = - parsedTarget.messageThreadId != null - ? String(parsedTarget.messageThreadId) - : params.threadId != null - ? normalizeOptionalString(String(params.threadId)) - : undefined; - if (threadId) { - const parsedTopic = parseTelegramTopicConversation({ - conversationId: threadId, - parentConversationId: chatId, - }); - if (!parsedTopic) { - return null; - } - return { - conversationId: parsedTopic.canonicalConversationId, - parentConversationId: parsedTopic.chatId, - }; - } return { - conversationId: chatId, + conversationId: buildTelegramConversationId({ + chatId, + thread: resolveTelegramTargetThread(parsedTarget, params.threadId) ?? { scope: "none" }, + }), parentConversationId: chatId, }; } @@ -476,12 +482,12 @@ function resolveTelegramDeliveryTarget(params: { }); if (parsedTopic) { return { - to: parsedTopic.chatId, - threadId: parsedTopic.topicId, + to: parsedTopic.canonicalConversationId, + threadId: parsedTopic.thread.id == null ? undefined : String(parsedTopic.thread.id), }; } const parsedTarget = parseTelegramTarget( - params.parentConversationId?.trim() || params.conversationId, + params.conversationId.trim() || params.parentConversationId?.trim() || "", ); if (!parsedTarget.chatId.trim()) { return null; @@ -553,7 +559,12 @@ function resolveTelegramOutboundSessionRoute(params: { if (!chatId) { return null; } - const resolvedThreadId = parsed.messageThreadId ?? parseTelegramThreadId(params.threadId); + const thread = resolveTelegramTargetThread(parsed, parseTelegramThreadId(params.threadId)); + const resolvedThreadId = thread ? Number(thread.id) : undefined; + const conversationId = buildTelegramConversationId({ + chatId, + thread: thread ?? { scope: "none" }, + }); const resolvedKind = params.resolvedTarget?.kind; const isGroup = parsed.chatType === "group" || @@ -561,8 +572,7 @@ function resolveTelegramOutboundSessionRoute(params: { // Telegram private chat ids are the sender's stable numeric user id, while // group ids are negative. Usernames remain aliases and cannot key replies. const recipientSessionExact = /^-?\d+$/.test(chatId); - const peerId = - isGroup && resolvedThreadId ? buildTelegramGroupPeerId(chatId, resolvedThreadId) : chatId; + const peerId = isGroup && resolvedThreadId ? conversationId : chatId; const peer: RoutePeer = { kind: isGroup ? "group" : "direct", id: peerId, @@ -579,9 +589,9 @@ function resolveTelegramOutboundSessionRoute(params: { from: isGroup ? `telegram:group:${peerId}` : resolvedThreadId - ? `telegram:${chatId}:topic:${resolvedThreadId}` + ? `telegram:${conversationId}` : `telegram:${chatId}`, - to: `telegram:${chatId}`, + to: `telegram:${conversationId}`, ...(isGroup && resolvedThreadId !== undefined ? { threadId: resolvedThreadId } : {}), }); if (isGroup) { @@ -606,7 +616,11 @@ function resolveTelegramOutboundSessionRoute(params: { }; const canonicalThreadId = resolvedThreadId !== undefined - ? buildTelegramCanonicalTopicThreadId({ chatId, topicId: resolvedThreadId }) + ? buildTelegramCanonicalTopicThreadId({ + chatId, + topicId: resolvedThreadId, + scope: thread?.scope ?? "forum", + }) : undefined; const route = buildThreadAwareOutboundSessionRoute({ route: directBaseRoute, @@ -622,15 +636,27 @@ function resolveTelegramOutboundSessionRoute(params: { ...(routeThreadId !== undefined ? { threadId: routeThreadId } : {}), from: routeThreadId !== undefined - ? `telegram:${chatId}:topic:${routeThreadId}` + ? buildTelegramRoutingTarget(chatId, { + id: Number(routeThreadId), + scope: thread?.scope ?? "forum", + }) : `telegram:${chatId}`, }; } -function buildTelegramCanonicalTopicThreadId(params: { chatId: string; topicId: number }): string { +function buildTelegramCanonicalTopicThreadId(params: { + chatId: string; + topicId: number; + scope: TelegramThreadSpec["scope"]; +}): string { // Core session routing sees one canonical thread id. Telegram topic ids are // chat-scoped, so direct-topic sessions include the chat id to avoid collisions. - return `${params.chatId}:${params.topicId}`; + return params.scope === "direct-messages" + ? buildTelegramConversationId({ + chatId: params.chatId, + thread: { id: params.topicId, scope: params.scope }, + }) + : `${params.chatId}:${params.topicId}`; } function resolveTelegramNativeTopicThreadId( @@ -782,8 +808,9 @@ export const telegramPlugin = createChatChannelPlugin({ threadId, }) => resolveTelegramInboundConversation({ - to: parentConversationId ?? conversationId, + to: conversationId, conversationId, + parentConversationId, threadId: threadId ?? undefined, }), buildBoundReplyPayload: ({ operation, conversation }) => { @@ -875,7 +902,8 @@ export const telegramPlugin = createChatChannelPlugin({ // Same function as the public session-key artifact so the pre-registry // fast path cannot drift from plugin behavior (pinned by contract test). resolveSessionConversation: resolveTelegramSessionConversation, - resolveSessionTarget: ({ kind, id }) => resolveTelegramSessionTarget({ kind, id }), + resolveSessionTarget: ({ kind, id, threadId }) => + resolveTelegramSessionTarget({ kind, id, threadId }), inferTargetChatType: ({ to }) => resolveTelegramRouteTarget(to).chatType, preserveHeartbeatThreadIdForGroupRoute: true, formatTargetDisplay: ({ target, display, kind }) => { @@ -1260,7 +1288,10 @@ export const telegramPlugin = createChatChannelPlugin({ if (threadId == null) { return to; } - return to.includes(":topic:") ? to : `${to}:topic:${threadId}`; + const parsed = parseTelegramTarget(to); + return parsed.messageThreadId != null || parsed.directMessagesTopicId != null + ? to + : `${to}:topic:${threadId}`; }, }, outbound: telegramChannelOutbound, diff --git a/extensions/telegram/src/conversation-route.base-session-key.test.ts b/extensions/telegram/src/conversation-route.base-session-key.test.ts index 6e79f9725adc..ef1a629473ab 100644 --- a/extensions/telegram/src/conversation-route.base-session-key.test.ts +++ b/extensions/telegram/src/conversation-route.base-session-key.test.ts @@ -153,6 +153,7 @@ describe("resolveTelegramConversationBaseSessionKey", () => { accountId: "default", chatId: 12345, isGroup: false, + threadSpec: { scope: "none" }, senderId: 12345, }); @@ -198,8 +199,7 @@ describe("resolveTelegramConversationBaseSessionKey", () => { accountId: "default", chatId: -1001234567890, isGroup: true, - resolvedThreadId: 11, - replyThreadId: 11, + threadSpec: { id: 11, scope: "forum" }, senderId: 12345, }); diff --git a/extensions/telegram/src/conversation-route.ts b/extensions/telegram/src/conversation-route.ts index 1521988868a0..1bd2f523b017 100644 --- a/extensions/telegram/src/conversation-route.ts +++ b/extensions/telegram/src/conversation-route.ts @@ -16,15 +16,13 @@ import { import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; import { resolveDefaultTelegramAccountId } from "./accounts.js"; -import { - buildTelegramGroupPeerId, - buildTelegramParentPeer, - shouldUseTelegramDmThreadSession, -} from "./bot/helpers.js"; +import { buildTelegramParentPeer, shouldUseTelegramDmThreadSession } from "./bot/helpers.js"; import { resolveTelegramDirectPeerId, resolveTelegramNamedAccountBaseSessionKey, } from "./dm-session-key.js"; +import type { TelegramThreadSpec } from "./thread-spec.js"; +import { buildTelegramConversationId } from "./topic-conversation.js"; type TelegramResolvedRoute = ReturnType; type ConfiguredTelegramBinding = NonNullable; @@ -52,17 +50,21 @@ export function resolveTelegramConversationRoute(params: { accountId: string; chatId: number | string; isGroup: boolean; - resolvedThreadId?: number; - replyThreadId?: number; + threadSpec: TelegramThreadSpec; senderId?: string | number | null; topicAgentId?: string | null; }): TelegramConversationRouteResult { + const resolvedThreadId = params.threadSpec.id; + const conversationId = buildTelegramConversationId({ + chatId: params.chatId, + thread: params.threadSpec, + }); const peerId = params.isGroup - ? buildTelegramGroupPeerId(params.chatId, params.resolvedThreadId) + ? conversationId : resolveTelegramDirectPeerId({ chatId: params.chatId, senderId: params.senderId }); const parentPeer = buildTelegramParentPeer({ isGroup: params.isGroup, - resolvedThreadId: params.resolvedThreadId, + resolvedThreadId, chatId: params.chatId, }); let route = resolveAgentRoute({ @@ -110,7 +112,7 @@ export function resolveTelegramConversationRoute(params: { }), }; logVerbose( - `telegram: topic route override: topic=${params.resolvedThreadId ?? params.replyThreadId} agent=${topicAgentId} sessionKey=${route.sessionKey}`, + `telegram: topic route override: topic=${resolvedThreadId} agent=${topicAgentId} sessionKey=${route.sessionKey}`, ); } @@ -120,8 +122,11 @@ export function resolveTelegramConversationRoute(params: { conversation: { channel: "telegram", accountId: params.accountId, - conversationId: peerId, - parentConversationId: params.isGroup ? String(params.chatId) : undefined, + conversationId: params.isGroup ? conversationId : peerId, + parentConversationId: + conversationId !== String(params.chatId) || params.isGroup + ? String(params.chatId) + : undefined, }, }); route = configuredRoute.route; @@ -133,10 +138,7 @@ export function resolveTelegramConversationRoute(params: { } : { kind: "none" }; - const runtimeBindingConversationId = - params.replyThreadId != null - ? `${params.chatId}:topic:${params.replyThreadId}` - : String(params.chatId); + const runtimeBindingConversationId = conversationId; const runtimeRoute = resolveRuntimeConversationBindingRoute({ route, conversation: { diff --git a/extensions/telegram/src/error-policy.test.ts b/extensions/telegram/src/error-policy.test.ts index 1db9fda258a8..7558bdd23768 100644 --- a/extensions/telegram/src/error-policy.test.ts +++ b/extensions/telegram/src/error-policy.test.ts @@ -38,7 +38,7 @@ describe("telegram error policy", () => { const scopeKey = buildTelegramErrorScopeKey({ accountId, chatId: 42, - threadId: 7, + threadSpec: { id: 7, scope: "forum" }, }); expect( @@ -201,7 +201,7 @@ describe("telegram error policy", () => { const workTopic = buildTelegramErrorScopeKey({ accountId, chatId: 42, - threadId: 9, + threadSpec: { id: 9, scope: "forum" }, }); expect( @@ -226,4 +226,19 @@ describe("telegram error policy", () => { }), ).toBe(false); }); + + it("keeps forum and direct-message topics with the same id in separate scopes", () => { + const base = { accountId, chatId: 42 }; + expect( + buildTelegramErrorScopeKey({ + ...base, + threadSpec: { id: 9, scope: "forum" }, + }), + ).not.toBe( + buildTelegramErrorScopeKey({ + ...base, + threadSpec: { id: 9, scope: "direct-messages" }, + }), + ); + }); }); diff --git a/extensions/telegram/src/error-policy.ts b/extensions/telegram/src/error-policy.ts index e3ed066e48b4..f0d3500bd4be 100644 --- a/extensions/telegram/src/error-policy.ts +++ b/extensions/telegram/src/error-policy.ts @@ -10,6 +10,7 @@ import { isFutureDateTimestampMs, resolveExpiresAtMsFromDurationMs, } from "openclaw/plugin-sdk/number-runtime"; +import { buildTelegramGroupPeerId, type TelegramThreadSpec } from "./bot/helpers.js"; type TelegramErrorPolicy = "always" | "once" | "silent"; @@ -57,10 +58,12 @@ export function resolveTelegramErrorPolicy(params: { export function buildTelegramErrorScopeKey(params: { accountId: string; chatId: string | number; - threadId?: string | number | null; + threadSpec?: TelegramThreadSpec; }): string { - const threadId = params.threadId == null ? "main" : String(params.threadId); - return `${params.accountId}:${String(params.chatId)}:${threadId}`; + return `${params.accountId}:${buildTelegramGroupPeerId( + params.chatId, + params.threadSpec ?? { scope: "none" }, + )}`; } export function shouldSuppressTelegramError(params: { diff --git a/extensions/telegram/src/location-message-hook.ts b/extensions/telegram/src/location-message-hook.ts index 800d5222bab8..8c5c27cb2ef0 100644 --- a/extensions/telegram/src/location-message-hook.ts +++ b/extensions/telegram/src/location-message-hook.ts @@ -31,7 +31,7 @@ function buildTelegramLocationMessageHook(params: { const threadSpec = resolveTelegramMessageThreadSpec(msg, params.isForum); const originatingTo = buildTelegramInboundOriginTarget(msg.chat.id, threadSpec); const from = isGroup - ? buildTelegramGroupFrom(msg.chat.id, threadSpec.id) + ? buildTelegramGroupFrom(msg.chat.id, threadSpec) : `telegram:${msg.chat.id}`; const canonical = deriveInboundMessageHookContext({ From: from, diff --git a/extensions/telegram/src/session-conversation.test.ts b/extensions/telegram/src/session-conversation.test.ts index 6f1d6d9cfead..599fb510e889 100644 --- a/extensions/telegram/src/session-conversation.test.ts +++ b/extensions/telegram/src/session-conversation.test.ts @@ -35,12 +35,32 @@ describe("resolveTelegramSessionConversation", () => { rawId: "-1001", }), ).toBeNull(); + expect( + resolveTelegramSessionConversation({ + kind: "group", + rawId: "-1001:direct-topic:77", + }), + ).toMatchObject({ + id: "-1001", + threadId: "direct-topic:77", + baseConversationId: "-1001", + }); }); }); describe("resolveTelegramSessionTarget", () => { it("normalizes group session ids to numeric chat ids", () => { expect(resolveTelegramSessionTarget({ kind: "group", id: "-1001" })).toBe("-1001"); + expect( + resolveTelegramSessionTarget({ + kind: "group", + id: "-1001", + threadId: "direct-topic:77", + }), + ).toBe("-1001:direct-topic:77"); + expect(resolveTelegramSessionTarget({ kind: "group", id: "-1001", threadId: "77" })).toBe( + "-1001:topic:77", + ); }); it("normalizes channel session ids to lookup targets", () => { diff --git a/extensions/telegram/src/session-conversation.ts b/extensions/telegram/src/session-conversation.ts index 42a7b64114ef..396dbc1af837 100644 --- a/extensions/telegram/src/session-conversation.ts +++ b/extensions/telegram/src/session-conversation.ts @@ -12,13 +12,21 @@ export function resolveTelegramSessionConversation(params: { } return { id: parsed.chatId, - threadId: parsed.topicId, + threadId: `${parsed.thread.scope === "direct-messages" ? "direct-topic:" : ""}${parsed.thread.id}`, baseConversationId: parsed.chatId, parentConversationCandidates: [parsed.chatId], }; } -export function resolveTelegramSessionTarget(params: { kind: "group" | "channel"; id: string }) { +export function resolveTelegramSessionTarget(params: { + kind: "group" | "channel"; + id: string; + threadId?: string | null; +}) { const raw = params.kind === "group" ? `telegram:group:${params.id}` : `telegram:${params.id}`; - return normalizeTelegramChatId(raw) ?? normalizeTelegramLookupTarget(raw); + const chatId = normalizeTelegramChatId(raw) ?? normalizeTelegramLookupTarget(raw); + const threadId = params.threadId?.startsWith("direct-topic:") + ? params.threadId + : params.threadId && `topic:${params.threadId}`; + return chatId && threadId ? `${chatId}:${threadId}` : chatId; } diff --git a/extensions/telegram/src/session-route.test.ts b/extensions/telegram/src/session-route.test.ts index 8a1867c35558..6564fa3be585 100644 --- a/extensions/telegram/src/session-route.test.ts +++ b/extensions/telegram/src/session-route.test.ts @@ -131,6 +131,38 @@ describe("telegram session route", () => { expect(route?.recipientSessionExact).toBe(true); }); + it("keeps direct-message and forum topics with the same id in distinct group routes", async () => { + const direct = await telegramPlugin.messaging?.resolveOutboundSessionRoute?.({ + cfg: {}, + agentId: "main", + target: "-100:direct-topic:99", + }); + const forum = await telegramPlugin.messaging?.resolveOutboundSessionRoute?.({ + cfg: {}, + agentId: "main", + target: "-100:topic:99", + }); + + expect(direct?.sessionKey).toBe("agent:main:telegram:group:-100:direct-topic:99"); + expect(direct?.from).toBe("telegram:group:-100:direct-topic:99"); + expect(direct?.to).toBe("telegram:-100:direct-topic:99"); + expect(direct?.threadId).toBe(99); + expect(forum?.sessionKey).toBe("agent:main:telegram:group:-100:topic:99"); + }); + + it("skips unusable command candidates and preserves a later direct topic", () => { + expect( + telegramPlugin.bindings?.resolveCommandConversation?.({ + accountId: "default", + originatingTo: "telegram:-100123", + commandTo: "telegram:-100123:direct-topic:77", + }), + ).toEqual({ + conversationId: "-100123:direct-topic:77", + parentConversationId: "-100123", + }); + }); + it("does not treat directory-resolved usernames as canonical session ids", async () => { const route = await telegramPlugin.messaging?.resolveOutboundSessionRoute?.({ cfg: {}, diff --git a/extensions/telegram/src/telegram-ingress-supersede-auth.ts b/extensions/telegram/src/telegram-ingress-supersede-auth.ts index 59055e72035f..361e41dad8f8 100644 --- a/extensions/telegram/src/telegram-ingress-supersede-auth.ts +++ b/extensions/telegram/src/telegram-ingress-supersede-auth.ts @@ -112,6 +112,7 @@ export async function isTelegramSpooledUpdateSenderAuthorized( const dmPolicy = accountCfg.dmPolicy ?? "pairing"; const allowFrom = accountCfg.allowFrom; const groupAllowFrom = accountCfg.groupAllowFrom ?? accountCfg.allowFrom; + const threadSpec = resolveTelegramMessageThreadSpec(facts.message); const groupAllowContext = await resolveTelegramGroupAllowFromContext({ cfg: auth.cfg, chatId: facts.chatId, @@ -120,7 +121,7 @@ export async function isTelegramSpooledUpdateSenderAuthorized( allowFrom, senderId: facts.senderId, isGroup: facts.isGroup, - threadSpec: resolveTelegramMessageThreadSpec(facts.message), + threadSpec, groupAllowFrom, resolveTelegramGroupConfig: (chatId, messageThreadId, cfg) => { const telegramCfg = mergeTelegramAccountConfig(cfg, auth.accountId); @@ -146,7 +147,7 @@ export async function isTelegramSpooledUpdateSenderAuthorized( accountId: auth.accountId, chatId: facts.chatId, isGroup: facts.isGroup, - ...(resolvedThreadId !== undefined ? { resolvedThreadId } : {}), + threadSpec, senderId: facts.senderId, ...(facts.senderUsername !== undefined ? { senderUsername: facts.senderUsername } : {}), }); diff --git a/extensions/telegram/src/thread-spec.ts b/extensions/telegram/src/thread-spec.ts new file mode 100644 index 000000000000..f9340e691fcd --- /dev/null +++ b/extensions/telegram/src/thread-spec.ts @@ -0,0 +1,6 @@ +// Telegram plugin module defines the canonical thread identity contract. +export type TelegramThreadSpec = { + id?: number; + /** dm is the historical bot-private topic scope. */ + scope: "direct-messages" | "dm" | "forum" | "none"; +}; diff --git a/extensions/telegram/src/topic-conversation.test.ts b/extensions/telegram/src/topic-conversation.test.ts index 77bac4e20796..b1738099b9f8 100644 --- a/extensions/telegram/src/topic-conversation.test.ts +++ b/extensions/telegram/src/topic-conversation.test.ts @@ -10,7 +10,7 @@ describe("parseTelegramTopicConversation", () => { }), ).toEqual({ chatId: "-1001234567890", - topicId: "42", + thread: { id: 42, scope: "forum" }, canonicalConversationId: "-1001234567890:topic:42", }); }); @@ -23,11 +23,23 @@ describe("parseTelegramTopicConversation", () => { }), ).toEqual({ chatId: "-1001234567890", - topicId: "42", + thread: { id: 42, scope: "forum" }, canonicalConversationId: "-1001234567890:topic:42", }); }); + it("keeps direct-message and forum topics with the same numeric id distinct", () => { + expect( + parseTelegramTopicConversation({ + conversationId: "-1001234567890:direct-topic:42", + }), + ).toEqual({ + chatId: "-1001234567890", + thread: { id: 42, scope: "direct-messages" }, + canonicalConversationId: "-1001234567890:direct-topic:42", + }); + }); + it("returns null when a DM binding carries the chat id in both fields", () => { expect( parseTelegramTopicConversation({ diff --git a/extensions/telegram/src/topic-conversation.ts b/extensions/telegram/src/topic-conversation.ts index 48ae14a26066..9ed22e8c9a63 100644 --- a/extensions/telegram/src/topic-conversation.ts +++ b/extensions/telegram/src/topic-conversation.ts @@ -1,64 +1,78 @@ -// Telegram plugin module implements topic conversation behavior. +// Telegram plugin module implements scoped topic conversation serialization. + +import { + normalizeTelegramChatId, + normalizeTelegramLookupTarget, + parseTelegramTarget, + type TelegramTarget, +} from "./targets.js"; +import type { TelegramThreadSpec } from "./thread-spec.js"; + export type ParsedTelegramTopicConversation = { chatId: string; - topicId: string; + thread: TelegramThreadSpec; canonicalConversationId: string; }; -function buildTelegramTopicConversationId(params: { +function threadSpecFromTarget(target: TelegramTarget): TelegramThreadSpec | null { + if (target.directMessagesTopicId != null) { + return { id: target.directMessagesTopicId, scope: "direct-messages" }; + } + return target.messageThreadId == null ? null : { id: target.messageThreadId, scope: "forum" }; +} + +function serializeTelegramTopicConversation(params: { chatId: string; - topicId: string; + thread: TelegramThreadSpec; }): string | null { - const chatId = params.chatId.trim(); - const topicId = params.topicId.trim(); - if (!/^-?\d+$/.test(chatId) || !/^\d+$/.test(topicId)) { + const chatId = + normalizeTelegramChatId(params.chatId) ?? normalizeTelegramLookupTarget(params.chatId); + const id = params.thread.id == null ? undefined : Math.trunc(params.thread.id); + if (!chatId || id == null || !Number.isFinite(id)) { return null; } - return `${chatId}:topic:${topicId}`; + const marker = + params.thread.scope === "direct-messages" && id > 0 + ? "direct-topic" + : params.thread.scope === "forum" && id >= 0 + ? "topic" + : null; + return marker ? `${chatId}:${marker}:${id}` : null; +} + +export function buildTelegramConversationId(params: { + chatId: string | number; + thread: TelegramThreadSpec; +}): string { + const chatId = String(params.chatId).trim(); + return serializeTelegramTopicConversation({ chatId, thread: params.thread }) ?? chatId; } export function parseTelegramTopicConversation(params: { conversationId: string; parentConversationId?: string; }): ParsedTelegramTopicConversation | null { - const conversation = params.conversationId.trim(); - const directMatch = conversation.match(/^(-?\d+):topic:(\d+)$/i); - if (directMatch?.[1] && directMatch[2]) { - const canonicalConversationId = buildTelegramTopicConversationId({ - chatId: directMatch[1], - topicId: directMatch[2], - }); - if (!canonicalConversationId) { - return null; - } - return { - chatId: directMatch[1], - topicId: directMatch[2], - canonicalConversationId, - }; - } - if (!/^\d+$/.test(conversation)) { - return null; + const conversationId = params.conversationId + .trim() + .replace(/:(direct-topic|topic):/i, (_match, marker: string) => `:${marker.toLowerCase()}:`); + const target = parseTelegramTarget(conversationId); + const chatId = + normalizeTelegramChatId(target.chatId) ?? normalizeTelegramLookupTarget(target.chatId); + const thread = threadSpecFromTarget(target); + if (chatId && thread) { + const canonicalConversationId = serializeTelegramTopicConversation({ chatId, thread }); + return canonicalConversationId ? { chatId, thread, canonicalConversationId } : null; } const parent = params.parentConversationId?.trim(); - if (!parent || !/^-?\d+$/.test(parent)) { + if (!/^\d+$/.test(conversationId) || !parent || parent === conversationId) { return null; } - // Telegram DM bindings can carry the chat id in both fields; treat that as - // a direct conversation shape, not a legacy topic binding. - if (parent === conversation) { - return null; - } - const canonicalConversationId = buildTelegramTopicConversationId({ + const parentThread: TelegramThreadSpec = { id: Number(conversationId), scope: "forum" }; + const canonicalConversationId = serializeTelegramTopicConversation({ chatId: parent, - topicId: conversation, + thread: parentThread, }); - if (!canonicalConversationId) { - return null; - } - return { - chatId: parent, - topicId: conversation, - canonicalConversationId, - }; + return canonicalConversationId + ? { chatId: parent, thread: parentThread, canonicalConversationId } + : null; } diff --git a/extensions/telegram/src/transport-payload.test.ts b/extensions/telegram/src/transport-payload.test.ts index 76681b4746e7..b35d491ffbae 100644 --- a/extensions/telegram/src/transport-payload.test.ts +++ b/extensions/telegram/src/transport-payload.test.ts @@ -9,6 +9,8 @@ import { } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { createTelegramCallbackMessageActions } from "./bot-handlers.callback-actions.js"; +import { buildTelegramMessageContextForTest } from "./bot-message-context.test-harness.js"; +import { telegramPlugin } from "./channel.js"; import { asTelegramClientFetch } from "./client-fetch.js"; import { createTelegramDraftStream } from "./draft-stream.js"; import { setTelegramRuntime } from "./runtime.js"; @@ -35,6 +37,26 @@ const cfg = { session: { store: "/tmp/openclaw-telegram-transport-payload-test.json" }, } satisfies OpenClawConfig; +function directMessagesMessage(overrides: Record = {}): Record { + return { + message_id: 41, + date: 1_700_000_000, + chat: { + id: DIRECT_CHAT_ID, + type: "supergroup", + title: "Channel Direct Messages", + is_direct_messages: true, + }, + direct_messages_topic: { + topic_id: DIRECT_TOPIC_ID, + user: { id: 700, is_bot: false, first_name: "Subscriber" }, + }, + from: { id: 700, is_bot: false, first_name: "Subscriber" }, + text: "button", + ...overrides, + }; +} + function installTelegramStateRuntimeForTest(): void { setTelegramRuntime({ state: { @@ -181,6 +203,46 @@ describe("Telegram topic transport payloads", () => { expect(request && hasMultipartField(request, "document")).toBe(true); }); + it("round-trips direct-topic conversation custody into the canonical Bot API field", async () => { + const inbound = await buildTelegramMessageContextForTest({ + message: directMessagesMessage(), + options: { forceWasMentioned: true }, + resolveGroupActivation: () => true, + }); + expect(inbound?.ctxPayload.SessionKey).toContain( + `telegram:group:${DIRECT_CHAT_ID}:direct-topic:${DIRECT_TOPIC_ID}`, + ); + const directRef = telegramPlugin.messaging?.resolveInboundConversation?.({ + to: inbound?.ctxPayload.OriginatingTo, + isGroup: true, + }); + expect(directRef?.conversationId).toBe(`${DIRECT_CHAT_ID}:direct-topic:${DIRECT_TOPIC_ID}`); + if (!directRef?.conversationId) { + throw new Error("expected direct-topic conversation reference"); + } + const persistedRef = structuredClone({ + conversationId: directRef.conversationId, + parentConversationId: directRef.parentConversationId, + }); + const target = telegramPlugin.messaging?.resolveDeliveryTarget?.(persistedRef); + if (!target?.to) { + throw new Error("expected persisted direct-topic delivery target"); + } + installTelegramStateRuntimeForTest(); + await sendMessageTelegram(target.to, "roundtrip", { + cfg, + token: TOKEN, + api: bot.api, + messageThreadId: target.threadId ? Number(target.threadId) : undefined, + }); + + const request = requests.find((candidate) => candidate.method === "sendMessage"); + expect(request && parseJsonBody(request)).toMatchObject({ + direct_messages_topic_id: DIRECT_TOPIC_ID, + }); + expect(request && parseJsonBody(request)).not.toHaveProperty("message_thread_id"); + }); + it("serializes local rich delivery through the canonical direct topic field", async () => { const richCfg = { ...cfg, @@ -219,26 +281,13 @@ describe("Telegram topic transport payloads", () => { }); it("serializes callback replies through only the canonical direct topic field", async () => { - const callbackMessage = { - message_id: 41, - date: 1_700_000_000, - chat: { - id: DIRECT_CHAT_ID, - type: "supergroup", - title: "Channel Direct Messages", - is_direct_messages: true, - }, + const callbackMessage = directMessagesMessage({ message_thread_id: 999, - direct_messages_topic: { - topic_id: DIRECT_TOPIC_ID, - user: { id: 700, is_bot: false, first_name: "Subscriber" }, - }, - text: "button", - } as Message; + }) as unknown as Message; const actions = createTelegramCallbackMessageActions({ bot, callbackMessage, - isForum: false, + threadSpec: { id: DIRECT_TOPIC_ID, scope: "direct-messages" }, }); await actions.replyToCallbackChat("callback reply");