ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-25 20:05:46 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(telegram): preserve direct-message topic routing (#126207)

Browse Source 
* fix(telegram): preserve direct message topic identity

* fix(telegram): break topic routing import cycle
This commit is contained in:
Peter Steinberger
2026-08-19 18:00:56 -07:00
committed by GitHub
parent 41009e765c
commit af3b86091f
43 changed files with 527 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/telegram/src/bot-handlers.callback-actions.ts
+4 -6
View File
@@ -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 }
extensions/telegram/src/bot-handlers.callback-router.ts
+12 -21
View File
@@ -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<TelegramContext, "me">;
chatId: number;
isGroup: boolean;
isForum: boolean;
messageThreadId?: number;
resolvedThreadId?: number;
threadSpec: ReturnType<typeof resolveTelegramMessageThreadSpec>;
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,
extensions/telegram/src/bot-handlers.debounce-key.ts
+3 -4
View File
@@ -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);
}
extensions/telegram/src/bot-handlers.event-bindings.ts
+10 -35
View File
@@ -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
extensions/telegram/src/bot-handlers.inbound-authorization.ts
+1 -1
View File
@@ -301,7 +301,7 @@ export function createTelegramHandlerAuthorization({
accountId,
chatId,
isGroup,
resolvedThreadId: context.resolvedThreadId,
threadSpec: context.threadSpec,
senderId,
senderUsername,
}).isAuthorizedSender;
extensions/telegram/src/bot-handlers.inbound-buffer.provenance.test.ts
+1
View File
@@ -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],
});
extensions/telegram/src/bot-handlers.inbound-buffer.ts
+10 -10
View File
@@ -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<boolean>;
@@ -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] : [],
extensions/telegram/src/bot-handlers.inbound-media.ts
+12 -14
View File
@@ -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<TelegramGroupMediaDisposition> => {
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}`,
extensions/telegram/src/bot-handlers.inbound-pipeline.test.ts
+18 -4
View File
@@ -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");
});
});
extensions/telegram/src/bot-handlers.inbound-pipeline.ts
+3 -6
View File
@@ -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,
extensions/telegram/src/bot-handlers.inbound-processing.ts
+13 -15
View File
@@ -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],
extensions/telegram/src/bot-handlers.message-context.runtime.test.ts
+1 -2
View File
@@ -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,
});
extensions/telegram/src/bot-handlers.message-context.ts
+4 -14
View File
@@ -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,
});
extensions/telegram/src/bot-handlers.message-session.runtime.test.ts
+1 -2
View File
@@ -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: {},
extensions/telegram/src/bot-handlers.reaction.runtime.test.ts
+2 -2
View File
@@ -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}`,
);
});
extensions/telegram/src/bot-message-context.body.test.ts
+1
View File
@@ -147,6 +147,7 @@ async function resolveBody(overrides: Partial<BodyParams> = {}) {
chatId,
senderId: String(chatId),
senderUsername: "",
threadSpec: { scope: "none" },
effectiveGroupAllow: normalizeAllowFrom([]),
effectiveDmAllow: normalizeAllowFrom([]),
requireMention: false,
extensions/telegram/src/bot-message-context.body.ts
+11 -4
View File
@@ -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 =
extensions/telegram/src/bot-message-context.dm-threads.test.ts
+4 -1
View File
@@ -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",
extensions/telegram/src/bot-message-context.session.ts
+1 -3
View File
@@ -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;
extensions/telegram/src/bot-message-context.thread-binding.test.ts
+4 -8
View File
@@ -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);
});
});
extensions/telegram/src/bot-message-context.ts
+3 -3
View File
@@ -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,
extensions/telegram/src/bot-message-context.types.ts
+2
View File
@@ -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<ResolvedChannelMessageIngress>;
export type TelegramMessageContextOptions = {
threadSpec?: TelegramThreadSpec;
commandSource?: "text" | "native";
forceWasMentioned?: boolean;
messageIdOverride?: string;
extensions/telegram/src/bot-message-dispatch-context.ts
+2 -2
View File
@@ -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
extensions/telegram/src/bot-message-dispatch-reply.ts
+1 -1
View File
@@ -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),
extensions/telegram/src/bot-native-command-dispatch.ts
+11 -11
View File
@@ -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",
extensions/telegram/src/bot-native-command-plugins.ts
+1 -1
View File
@@ -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"
extensions/telegram/src/bot.create-telegram-bot.test.ts
+8 -12
View File
@@ -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({
extensions/telegram/src/bot.test.ts
+3
View File
@@ -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({
extensions/telegram/src/bot/helpers.ts
+20 -17
View File
@@ -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,
extensions/telegram/src/channel.ts
+109 -78
View File
@@ -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<typeof parseTelegramTarget>,
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,
extensions/telegram/src/conversation-route.base-session-key.test.ts
+2 -2
View File
@@ -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,
});
extensions/telegram/src/conversation-route.ts
+18 -16
View File
@@ -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<typeof resolveAgentRoute>;
type ConfiguredTelegramBinding = NonNullable<ConfiguredBindingRouteResult["bindingResolution"]>;
@@ -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: {
extensions/telegram/src/error-policy.test.ts
+17 -2
View File
@@ -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" },
}),
);
});
});
extensions/telegram/src/error-policy.ts
+6 -3
View File
@@ -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: {
extensions/telegram/src/location-message-hook.ts
+1 -1
View File
@@ -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,
extensions/telegram/src/session-conversation.test.ts
+20
View File
@@ -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", () => {
extensions/telegram/src/session-conversation.ts
+11 -3
View File
@@ -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;
}
extensions/telegram/src/session-route.test.ts
+32
View File
@@ -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: {},
extensions/telegram/src/telegram-ingress-supersede-auth.ts
+3 -2
View File
@@ -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 } : {}),
});
extensions/telegram/src/thread-spec.ts
+6
View File
@@ -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";
};
extensions/telegram/src/topic-conversation.test.ts
+14 -2
View File
@@ -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({
extensions/telegram/src/topic-conversation.ts
+56 -42
View File
@@ -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;
}
extensions/telegram/src/transport-payload.test.ts
+65 -16
View File
@@ -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<string, unknown> = {}): Record<string, unknown> {
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");