// Telegram tests cover provider-observed forum-topic message bindings. import type { Message } from "grammy/types"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createPluginStateKeyedStoreForTests, resetPluginStateStoreForTests, } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { resolveTelegramMessageCacheScope } from "./message-cache-persistence.js"; import { createTelegramMessageCache } from "./message-cache.js"; import { resolveTelegramMessageMutationChatId } from "./message-topic-binding.js"; import { recordOutboundMessageForPromptContext } from "./outbound-message-context.js"; import { setTelegramRuntime } from "./runtime.js"; import { clearTelegramRuntimeForTest, resetTelegramMessageCacheForTest, } from "./runtime.test-support.js"; import type { TelegramRuntime } from "./runtime.types.js"; const cfg = { agents: { entries: { main: { default: true } } }, channels: { telegram: { botToken: "tok" } }, session: { store: "/tmp/openclaw-telegram-topic-binding-test.json" }, } as OpenClawConfig; function installRuntimeStore() { setTelegramRuntime({ state: { openKeyedStore: ((options) => createPluginStateKeyedStoreForTests( "telegram", options, )) as TelegramRuntime["state"]["openKeyedStore"], }, channel: {}, } as TelegramRuntime); } function delegatedContext(overrides?: Record) { return { conversationReadOrigin: "delegated" as const, requesterAccountId: "default", toolContext: { currentChannelProvider: "telegram" as const, currentChannelId: "telegram:-1001:topic:77", currentMessageId: "901", }, ...overrides, }; } function topicMessage(messageId: number, threadId: number): Message { return { chat: { id: -1001, type: "supergroup", title: "QA", is_forum: true }, message_id: messageId, message_thread_id: threadId, is_topic_message: true, date: 1_736_380_700, text: `message-${messageId}`, from: { id: 1, is_bot: false, first_name: "QA" }, } as Message; } async function recordMessage(params: { messageId: number; threadId: number; providerObserved: boolean; accountId?: string; }) { const cache = createTelegramMessageCache({ scope: resolveTelegramMessageCacheScope( resolveStorePath(cfg.session?.store, { agentId: "main" }), ), }); await cache.record({ accountId: params.accountId ?? "default", chatId: -1001, msg: topicMessage(params.messageId, params.threadId), threadId: params.threadId, ...(params.providerObserved ? { providerObservedThread: { scope: "forum" as const, id: params.threadId } } : {}), }); } describe("Telegram message topic binding", () => { beforeEach(() => { resetPluginStateStoreForTests(); resetTelegramMessageCacheForTest(); installRuntimeStore(); }); afterEach(() => { clearTelegramRuntimeForTest(); resetTelegramMessageCacheForTest(); resetPluginStateStoreForTests(); }); it("allows the trusted current message without a cache lookup", async () => { await expect( resolveTelegramMessageMutationChatId({ chatId: "telegram:-1001:topic:77", messageId: 901, cfg, accountId: "default", context: delegatedContext(), }), ).resolves.toBe("-1001"); }); it("binds a delegated base-chat spelling to the trusted current topic", async () => { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", messageId: 901, cfg, accountId: "default", context: delegatedContext(), }), ).resolves.toBe("-1001"); }); it("binds General-topic mutations using trusted thread context", async () => { await recordMessage({ messageId: 900, threadId: 1, providerObserved: true }); await recordMessage({ messageId: 899, threadId: 2, providerObserved: true }); const context = delegatedContext({ toolContext: { currentChannelProvider: "telegram", currentChannelId: "telegram:-1001", currentMessageId: "901", currentThreadTs: "1", }, }); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", messageId: 901, cfg, accountId: "default", context, }), ).resolves.toBe("-1001"); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", messageId: 900, cfg, accountId: "default", context, }), ).resolves.toBe("-1001"); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", messageId: 899, cfg, accountId: "default", context, }), ).rejects.toThrow("provider-observed binding"); }); it("rejects a topicless different-chat target during a trusted topic turn", async () => { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1002", messageId: 901, cfg, accountId: "default", context: delegatedContext(), }), ).rejects.toThrow("provider-observed binding"); }); it.each([ { name: "different chat", chatId: "-1002", context: delegatedContext({ toolContext: { currentChannelProvider: "telegram", currentChannelId: "telegram:-1001", currentMessageId: "901", }, }), }, { name: "different provider", chatId: "-1001", context: delegatedContext({ toolContext: { currentChannelProvider: "slack", currentChannelId: "telegram:-1001", currentMessageId: "901", }, }), }, { name: "different account", chatId: "-1001", context: delegatedContext({ requesterAccountId: "work", toolContext: { currentChannelProvider: "telegram", currentChannelId: "telegram:-1001", currentMessageId: "901", }, }), }, ])("rejects a threadless $name mutation", async ({ chatId, context }) => { await expect( resolveTelegramMessageMutationChatId({ chatId, messageId: 901, cfg, accountId: "default", context, }), ).rejects.toThrow("provider-observed binding"); }); it("allows a threadless exact-current mutation with a matching account", async () => { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001", messageId: 901, cfg, accountId: "default", context: delegatedContext({ toolContext: { currentChannelProvider: "telegram", currentChannelId: "telegram:-1001", currentMessageId: "901", }, }), }), ).resolves.toBe("-1001"); }); it("allows an earlier provider-observed same-topic message after restart", async () => { await recordMessage({ messageId: 900, threadId: 77, providerObserved: true }); resetTelegramMessageCacheForTest(); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 900, cfg, accountId: "default", context: delegatedContext(), }), ).resolves.toBe("-1001"); }); it("reads an outbound topic binding from the routed account owner's cache after restart", async () => { const multiAgentCfg = { agents: { ownership: "explicit", entries: { main: {}, ops: {}, research: {} }, }, channels: { telegram: { accounts: { alerts: { botToken: "123456:alerts" } } }, }, bindings: [{ agentId: "ops", match: { channel: "telegram", accountId: "alerts" } }], session: { store: "/tmp/openclaw-telegram-topic-owner/{agentId}/sessions.json" }, } as OpenClawConfig; await recordOutboundMessageForPromptContext({ cfg: multiAgentCfg, account: { accountId: "alerts", name: "Alerts" }, chatId: -1001, message: topicMessage(902, 77), messageId: 902, successfulSendThread: { scope: "forum", id: 77 }, }); resetTelegramMessageCacheForTest(); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 902, cfg: multiAgentCfg, accountId: "alerts", context: delegatedContext({ requesterAccountId: "alerts" }), }), ).resolves.toBe("-1001"); }); it("rejects legacy and wrong-topic cache entries", async () => { await recordMessage({ messageId: 899, threadId: 77, providerObserved: false }); await recordMessage({ messageId: 900, threadId: 88, providerObserved: true }); resetTelegramMessageCacheForTest(); for (const messageId of [899, 900]) { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId, cfg, accountId: "default", context: delegatedContext(), }), ).rejects.toThrow("provider-observed binding"); } }); it("does not borrow a binding from another account", async () => { await recordMessage({ messageId: 900, threadId: 77, providerObserved: true }); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 900, cfg, accountId: "work", context: delegatedContext({ requesterAccountId: "work" }), }), ).rejects.toThrow("provider-observed binding"); await recordMessage({ messageId: 900, threadId: 77, providerObserved: true, accountId: "work", }); await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 900, cfg, accountId: "work", context: delegatedContext({ requesterAccountId: "work" }), }), ).resolves.toBe("-1001"); }); it.each([ { name: "missing origin", messageId: 800, context: { requesterAccountId: "default", toolContext: delegatedContext().toolContext, }, }, { name: "unknown origin", messageId: 800, context: delegatedContext({ conversationReadOrigin: "forged-direct-operator" }), }, { name: "wrong account", context: delegatedContext({ requesterAccountId: "work" }), }, { name: "invalid account", accountId: " !!! ", context: delegatedContext({ requesterAccountId: " !!! " }), }, { name: "wrong provider", context: delegatedContext({ toolContext: { ...delegatedContext().toolContext, currentChannelProvider: "slack", }, }), }, { name: "stale topic", context: delegatedContext({ toolContext: { ...delegatedContext().toolContext, currentChannelId: "telegram:-1001:topic:88", }, }), }, { name: "conflicting trusted thread context", context: delegatedContext({ toolContext: { ...delegatedContext().toolContext, currentThreadTs: "88", }, }), }, { name: "conflicting target forms", context: delegatedContext({ toolContext: { ...delegatedContext().toolContext, currentMessagingTarget: "telegram:-1002:topic:77", }, }), }, ])("rejects $name before provider execution", async ({ accountId, context, messageId }) => { await expect( resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: messageId ?? 901, cfg, accountId: accountId ?? "default", context: context as never, }), ).rejects.toThrow("provider-observed binding"); }); it("keeps direct and delegated authority operation-local under mixed ordering", async () => { const direct = () => resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 800, cfg, accountId: "default", context: { conversationReadOrigin: "direct-operator" }, }); const delegated = () => resolveTelegramMessageMutationChatId({ chatId: "-1001:topic:77", messageId: 800, cfg, accountId: "default", context: delegatedContext(), }); await expect(direct()).resolves.toBe("-1001"); await expect(delegated()).rejects.toThrow("provider-observed binding"); await expect(delegated()).rejects.toThrow("provider-observed binding"); await expect(direct()).resolves.toBe("-1001"); const concurrent = await Promise.allSettled([direct(), delegated()]); expect(concurrent.map((result) => result.status)).toEqual(["fulfilled", "rejected"]); }); });