import type { Message } from "grammy/types"; import { describe, expect, it } from "vitest"; import { resolveTelegramMessageCachePersistentScopeKey, TELEGRAM_MESSAGE_CACHE_PERSISTENT_MAX_MESSAGES, type TelegramResolvedMedia, } from "./message-cache-persistence.js"; import { buildTelegramConversationContext, buildTelegramReplyChain, createTelegramMessageCache, hasProviderObservedTelegramThreadBinding, resolveProviderObservedTelegramThreadSpec, } from "./message-cache.js"; import { resetTelegramMessageCacheForTest as resetCache } from "./runtime.test-support.js"; type PersistentStore = NonNullable< NonNullable[0]>["persistentStore"] >; type Cache = ReturnType; type ReplyChain = Awaited>; type PersistedValue = { version: 1; sourceMessage: Message; botUserId?: number; promptContextProjection?: unknown; resolvedMedia?: TelegramResolvedMedia; threadBinding?: { kind: "provider-observed-v1"; threadSpec: { scope: "direct-messages" | "dm" | "forum"; id: number }; }; threadId?: string; }; let persistentStoreId = 0; function createMemoryStore(maxEntries = TELEGRAM_MESSAGE_CACHE_PERSISTENT_MAX_MESSAGES) { const entries = new Map(); return { bucketKey: `test:${process.pid}:${Date.now()}:${persistentStoreId++}`, entries, store: { async register(key, value) { entries.delete(key); entries.set(key, structuredClone(value)); while (entries.size > maxEntries) { const oldest = entries.keys().next().value; if (oldest === undefined) { break; } entries.delete(oldest); } }, async entries() { return Array.from(entries, ([key, value]) => ({ key, value: structuredClone(value) })); }, } satisfies PersistentStore, }; } const sender = (id: number, first_name: string, is_bot = false) => ({ id, is_bot, first_name }); function message(message_id: number, firstName: string, overrides: Record = {}) { const { chat, date, from, ...rest } = overrides; return { chat: chat ?? { id: 7, type: "private", first_name: firstName }, message_id, date: date ?? 1_736_371_600 + message_id, from: from ?? sender(1, firstName), ...rest, } as Message; } function photo(file_id: string) { return [{ file_id, file_unique_id: `${file_id}-unique`, width: 640, height: 480 }]; } function botMessage(messageId: number, text: string, overrides: Record = {}) { return message(messageId, "OpenClaw", { text, from: sender(999, "OpenClaw", true), ...overrides, }); } function record(cache: Cache, msg: Message, overrides: Record = {}) { return cache.record({ accountId: "default", chatId: 7, msg, ...overrides } as never); } function get(cache: Cache, messageId: string, overrides: Record = {}) { return cache.get({ accountId: "default", chatId: 7, messageId, ...overrides } as never); } function reloadGet(bucketKey: string, store: PersistentStore, messageId: string) { resetCache(); return get(cacheFor(bucketKey, store), messageId); } function recentBefore(cache: Cache, messageId: string, overrides: Record = {}) { return cache.recentBefore({ accountId: "default", chatId: 7, messageId, limit: 10, ...overrides, } as never); } const replyChain = (cache: Cache, msg: Message, chatId = 7) => buildTelegramReplyChain({ cache, accountId: "default", chatId, msg }); const cacheFor = (bucketKey: string, persistentStore: PersistentStore) => createTelegramMessageCache({ bucketKey, persistentStore }); function entryStore(store: PersistentStore, key: string, value: unknown): PersistentStore { return { register: (nextKey, nextValue) => store.register(nextKey, nextValue), entries: async () => [{ key, value }], } as PersistentStore; } function conversationContext(cache: Cache, messageId: string, replyChainNodes: ReplyChain) { return buildTelegramConversationContext({ cache, accountId: "default", chatId: 7, messageId, replyChainNodes, recentLimit: 10, replyTargetWindowSize: 2, }); } function onlyEntry(entries: Map): [string, PersistedValue] { const entry = entries.entries().next().value; if (!entry) { throw new Error("expected persisted Telegram message cache value"); } return entry; } const projection = (transcriptMessageId: string) => ({ transcriptMessageId, partIndex: 0, finalPart: true, }); describe("telegram message cache", () => { it("persists resolved media with its source message and drops it when the media changes", async () => { const { bucketKey, entries, store } = createMemoryStore(); const cache = cacheFor(bucketKey, store); await record(cache, message(9000, "Kesava", { photo: photo("photo-1") })); await cache.recordResolvedMedia({ accountId: "default", chatId: 7, messageId: "9000", media: { id: "saved-photo.png", fileUniqueId: "photo-1-unique", size: 4, savedAt: 1_736_380_700_000, kind: "image", contentType: "image/png", }, }); expect(onlyEntry(entries)[1].resolvedMedia?.id).toBe("saved-photo.png"); const reloaded = await reloadGet(bucketKey, store, "9000"); expect(reloaded?.resolvedMedia).toMatchObject({ id: "saved-photo.png", fileUniqueId: "photo-1-unique", kind: "image", }); const reloadedCache = cacheFor(bucketKey, store); await record(reloadedCache, message(9000, "Kesava", { photo: photo("photo-2") })); expect((await get(reloadedCache, "9000"))?.resolvedMedia).toBeUndefined(); }); it("persists provider-observed topic bindings for messages and same-topic replies", async () => { const { bucketKey, entries, store } = createMemoryStore(); const forum = { id: -1001, type: "supergroup", title: "QA", is_forum: true }; const parent = message(901, "Ada", { chat: forum, date: 1_736_380_701, text: "Parent", from: sender(1, "Ada"), }); const cache = cacheFor(bucketKey, store); await record( cache, message(902, "Grace", { chat: forum, date: 1_736_380_702, text: "Reply", from: sender(2, "Grace"), message_thread_id: 77, is_topic_message: true, reply_to_message: parent, }), { chatId: -1001, threadId: 77, providerObservedThread: { scope: "forum", id: 77 }, }, ); expect(entries.size).toBe(2); expect( Array.from(entries.values()).every( (value) => value.threadBinding?.kind === "provider-observed-v1" && value.threadBinding.threadSpec.scope === "forum" && value.threadBinding.threadSpec.id === 77, ), ).toBe(true); resetCache(); const reloaded = cacheFor(bucketKey, store); for (const messageId of ["901", "902"]) { const node = await get(reloaded, messageId, { chatId: -1001 }); expect(hasProviderObservedTelegramThreadBinding(node, 77)).toBe(true); expect(resolveProviderObservedTelegramThreadSpec(node)).toEqual({ scope: "forum", id: 77 }); } }); it("keeps an authoritative supplied thread ahead of conflicting root message metadata", async () => { const { bucketKey, entries, store } = createMemoryStore(); const forum = { id: -1001, type: "supergroup", title: "QA", is_forum: true }; const ancestor = message(904, "Lin", { chat: forum, text: "Ancestor without its own thread metadata", }); const reply = message(905, "Ada", { chat: forum, text: "Reply with embedded thread metadata", message_thread_id: 88, reply_to_message: ancestor, }); const root = message(906, "Grace", { chat: forum, text: "Authoritative root", message_thread_id: 999, reply_to_message: reply, }); const cache = cacheFor(bucketKey, store); const recorded = await record(cache, root, { chatId: -1001, threadId: 77, providerObservedThread: { scope: "forum", id: 77 }, }); expect(recorded.threadId).toBe("77"); expect(resolveProviderObservedTelegramThreadSpec(recorded)).toEqual({ scope: "forum", id: 77, }); expect((await get(cache, "905", { chatId: -1001 }))?.threadId).toBe("88"); expect((await get(cache, "904", { chatId: -1001 }))?.threadId).toBe("88"); const persistedRoot = Array.from(entries.values()).find( (entry) => entry.sourceMessage.message_id === 906, ); expect(persistedRoot?.threadId).toBe("77"); }); it("persists a channel Direct Messages binding ahead of conflicting raw thread metadata", async () => { const { bucketKey, store } = createMemoryStore(); const cache = cacheFor(bucketKey, store); await record( cache, message(904, "Ada", { chat: { id: -1002, type: "supergroup", title: "Channel replies", is_direct_messages: true, }, direct_messages_topic: { topic_id: 77 }, message_thread_id: 999, is_topic_message: true, }), { chatId: -1002, threadId: 999, providerObservedThread: { scope: "direct-messages", id: 77 }, }, ); resetCache(); const reloaded = await get(cacheFor(bucketKey, store), "904", { chatId: -1002 }); expect(reloaded?.threadId).toBe("77"); expect(resolveProviderObservedTelegramThreadSpec(reloaded)).toEqual({ scope: "direct-messages", id: 77, }); }); it("does not resolve caller-only topic metadata as a provider-observed binding", async () => { const cache = createTelegramMessageCache(); await record( cache, message(903, "Ada", { chat: { id: -1001, type: "supergroup", title: "QA", is_forum: true }, message_thread_id: 77, is_topic_message: true, }), { chatId: -1001, threadId: 77 }, ); const node = await get(cache, "903", { chatId: -1001 }); expect(resolveProviderObservedTelegramThreadSpec(node)).toBeUndefined(); }); it("hydrates reply chains from persisted cached messages", async () => { const { bucketKey, store } = createMemoryStore(); const photoMessage = message(9000, "Kesava", { date: 1_736_380_700, photo: photo("photo-1"), }); const reply = message(9001, "Ada", { date: 1_736_380_750, text: "The cache warmer is the piece I meant", from: sender(2, "Ada"), reply_to_message: photoMessage, }); const firstCache = cacheFor(bucketKey, store); await record(firstCache, photoMessage); await record(firstCache, reply); resetCache(); const secondCache = cacheFor(bucketKey, store); const chain = await replyChain( secondCache, message(9002, "Grace", { text: "Please explain what this reply was about", from: sender(3, "Grace"), reply_to_message: message(9001, "Ada", { date: 1_736_380_750, text: "The cache warmer is the piece I meant", from: sender(2, "Ada"), }), }), ); expect(chain).toEqual([ { messageId: "9001", sender: "Ada", senderId: "2", timestamp: 1736380750000, body: "The cache warmer is the piece I meant", replyToId: "9000", sourceMessage: reply, }, { messageId: "9000", sender: "Kesava", senderId: "1", timestamp: 1736380700000, mediaRef: "telegram:file/photo-1", mediaType: "image", sourceMessage: photoMessage, }, ]); }); it("records embedded reply targets as normal cached messages", async () => { const { bucketKey, store } = createMemoryStore(); const chat = { id: 7, type: "group", title: "Ops" }; const imageReply = message(101, "Bot", { chat, date: 1_736_380_700, text: "Done, here is the image", from: sender(999, "Bot", true), photo: photo("generated-photo-1"), }); const userReply = message(102, "UserB", { chat, date: 1_736_380_750, text: "Why is there a 4th person?", from: sender(2, "UserB"), reply_to_message: imageReply, }); const firstCache = cacheFor(bucketKey, store); await record(firstCache, userReply); resetCache(); const secondCache = cacheFor(bucketKey, store); const chain = await replyChain( secondCache, message(103, "UserA", { chat, date: 1_736_380_800, text: "Explain what went wrong", reply_to_message: message(102, "UserB", { chat, date: 1_736_380_750, text: "Why is there a 4th person?", from: sender(2, "UserB"), }), }), ); const context = await conversationContext(secondCache, "103", chain); expect(chain.map((entry) => entry.messageId)).toEqual(["102", "101"]); expect(chain[1]).toMatchObject({ sender: "Bot", body: "Done, here is the image", mediaRef: "telegram:file/generated-photo-1", }); expect(context.map((entry) => entry.node.messageId)).toEqual(["101", "102"]); expect(context.find((entry) => entry.node.messageId === "101")?.isReplyTarget).toBe(true); }); it("replaces authoritative edited message fields without stale caption carryover", async () => { const cache = createTelegramMessageCache(); const chat = { id: 7, type: "group", title: "Ops" }; const photoFields = { chat, from: sender(999, "Bot", true), photo: photo("generated-photo-2") }; await record( cache, message(104, "Bot", { ...photoFields, date: 1_736_380_900, caption: "old caption" }), ); const updated = await record( cache, message(104, "Bot", { ...photoFields, date: 1_736_380_900, edit_date: 1_736_380_910 }), ); expect(updated).toMatchObject({ messageId: "104", mediaType: "image", mediaRef: "telegram:file/generated-photo-2", }); expect(updated.body).toBeUndefined(); expect(updated?.body).not.toBe("old caption"); }); it("shares one persisted bucket across live cache instances", async () => { const { bucketKey, store } = createMemoryStore(); const [firstCache, secondCache] = [cacheFor(bucketKey, store), cacheFor(bucketKey, store)]; const nora = message(9100, "Nora", { text: "Architecture sketch for the cache warmer" }); const ira = message(9101, "Ira", { text: "The cache warmer is the piece I meant", from: sender(2, "Ira"), reply_to_message: nora, }); await record(firstCache, nora); await record(secondCache, ira); const chain = await replyChain( cacheFor(bucketKey, store), message(9102, "Mina", { text: "Please explain what this reply was about", from: sender(3, "Mina"), reply_to_message: message(9101, "Ira", { text: "The cache warmer is the piece I meant", from: sender(2, "Ira"), }), }), ); expect(chain.map((entry) => entry.messageId)).toEqual(["9101", "9100"]); }); it("persists cached records through the plugin state store", async () => { const { bucketKey, store } = createMemoryStore(3); const cache = cacheFor(bucketKey, store); for (let index = 0; index < 5; index++) { await record( cache, message(9120 + index, "Nora", { date: 1_736_380_700 + index, text: `State message ${index}`, }), ); } resetCache(); const recent = await recentBefore(cacheFor(bucketKey, store), "9125"); expect(recent.map((entry) => entry.messageId)).toEqual(["9122", "9123", "9124"]); }); it("persists prompt-context projection provenance across cache restart", async () => { const { bucketKey, entries, store } = createMemoryStore(); const marker = projection("assistant-projection-restart"); const cache = cacheFor(bucketKey, store); await record(cache, botMessage(9125, "Projection-aware state message"), { promptContextProjection: marker, }); expect(entries.values().next().value).toMatchObject({ version: 1, promptContextProjection: marker, }); resetCache(); const reloadedCache = cacheFor(bucketKey, store); const reloaded = await get(reloadedCache, "9125"); expect(reloaded?.promptContextProjectionMarker).toEqual({ kind: "valid", projection: marker }); const edited = await record( reloadedCache, botMessage(9125, "Edited projection-aware state message", { edit_date: 1_736_380_730 }), ); expect(edited).toMatchObject({ body: "Edited projection-aware state message", promptContextProjectionMarker: { kind: "valid", projection: marker }, }); const editedReloaded = await reloadGet(bucketKey, store, "9125"); expect(editedReloaded).toMatchObject({ body: "Edited projection-aware state message", promptContextProjectionMarker: { kind: "valid", projection: marker }, }); const malformedStore = entryStore(store, entries.keys().next().value!, { ...entries.values().next().value, promptContextProjection: { ...marker, partIndex: -1 }, }); resetCache(); const malformedCache = cacheFor(bucketKey, malformedStore); const malformed = await get(malformedCache, "9125"); expect(malformed?.promptContextProjectionMarker).toEqual({ kind: "invalid", transcriptMessageId: marker.transcriptMessageId, }); await record( malformedCache, botMessage(9125, "Edited malformed projection state message", { edit_date: 1_736_380_731 }), ); expect(entries.values().next().value?.promptContextProjection).toEqual({ transcriptMessageId: marker.transcriptMessageId, }); const malformedReloaded = await reloadGet(bucketKey, store, "9125"); expect(malformedReloaded?.promptContextProjectionMarker).toEqual({ kind: "invalid", transcriptMessageId: marker.transcriptMessageId, }); }); it("recognizes projected messages sent on behalf of a Telegram Business account", async () => { const { bucketKey, entries, store } = createMemoryStore(); const marker = projection("assistant-business-projection"); const businessMessage = message(9128, "Business User", { text: "Business reply", from: sender(700, "Business User"), sender_business_bot: sender(42, "OpenClaw", true), }); const cache = cacheFor(bucketKey, store); const live = await record(cache, businessMessage, { botUserId: 42, promptContextProjection: marker, }); expect(live.promptContextProjectionMarker).toEqual({ kind: "valid", projection: marker }); expect(entries.values().next().value).toMatchObject({ botUserId: 42 }); const reloaded = await reloadGet(bucketKey, store, "9128"); expect(reloaded?.promptContextProjectionMarker).toEqual({ kind: "valid", projection: marker }); const [persistedKey, persistedValue] = onlyEntry(entries); entries.set(persistedKey, { ...persistedValue, botUserId: 99 }); const mismatched = await reloadGet(bucketKey, store, "9128"); expect(mismatched?.promptContextProjectionMarker).toBeUndefined(); }); it("preserves projected message whitespace across cache restart", async () => { const { bucketKey, store } = createMemoryStore(); const marker = projection("assistant-whitespace-projection"); const text = " indented\nnext \n"; const cache = cacheFor(bucketKey, store); const live = await record( cache, message(9132, "OpenClaw", { text, from: sender(42, "OpenClaw", true) }), { botUserId: 42, promptContextProjection: marker, }, ); expect(live.body).toBe(text); const reloaded = await reloadGet(bucketKey, store, "9132"); expect(reloaded?.body).toBe(text); expect(reloaded?.promptContextProjectionMarker).toEqual({ kind: "valid", projection: marker }); }); it("poisons projection provenance when its durable cache write fails", async () => { const bucketKey = `test:${process.pid}:${Date.now()}:${persistentStoreId++}`; const persistentStore: PersistentStore = { async register() { throw new Error("state store unavailable"); }, async entries() { return []; }, }; const cache = cacheFor(bucketKey, persistentStore); await expect( record(cache, message(9126, "Nora", { text: "Markerless context" })), ).resolves.toMatchObject({ messageId: "9126" }); const marker = projection("assistant-persistence-failure"); await expect( record(cache, botMessage(9127, "Projected context"), { promptContextProjection: marker }), ).rejects.toThrow("state store unavailable"); await expect(get(cache, "9127")).resolves.toMatchObject({ promptContextProjectionMarker: { kind: "invalid", transcriptMessageId: marker.transcriptMessageId, }, }); }); it.each([ ["projected row first", ["projected", "parent"]], ["embedding parent first", ["parent", "projected"]], ])("keeps projected bot provenance when hydrating $0", async (_name, order) => { const { bucketKey, entries, store } = createMemoryStore(); const scopeKey = resolveTelegramMessageCachePersistentScopeKey("default"); const marker = projection("assistant-embedded-order"); const bot = botMessage(9130, "Projected answer"); const values: Record = { projected: [ `${scopeKey}:default:7:9130`, { version: 1, sourceMessage: bot, promptContextProjection: marker }, ], parent: [ `${scopeKey}:default:7:9131`, { version: 1, sourceMessage: message(9131, "Nora", { text: "Replying to the answer", reply_to_message: bot, }), }, ], }; for (const name of order) { const [key, value] = values[name]!; entries.set(key, value); } const hydrated = await get(cacheFor(bucketKey, store), "9130"); expect(hydrated?.promptContextProjectionMarker).toEqual({ kind: "valid", projection: marker }); }); it("ignores persisted projection metadata on inbound messages", async () => { const { bucketKey, entries, store } = createMemoryStore(); const scopeKey = resolveTelegramMessageCachePersistentScopeKey("default"); entries.set(`${scopeKey}:default:7:9140`, { version: 1, sourceMessage: message(9140, "Nora", { text: "Inbound text" }), promptContextProjection: projection("must-not-be-trusted"), }); const hydrated = await get(cacheFor(bucketKey, store), "9140"); expect(hydrated?.promptContextProjectionMarker).toBeUndefined(); }); it("hydrates unversioned pre-projection rows without inferring provenance", async () => { const { bucketKey, entries, store } = createMemoryStore(); const cache = cacheFor(bucketKey, store); await record(cache, botMessage(9126, "Pre-projection state message")); const [persistedKey, persistedValue] = onlyEntry(entries); const legacyStore = entryStore(store, persistedKey, { sourceMessage: persistedValue.sourceMessage, promptContextProjection: projection("must-not-be-inferred"), threadBinding: { kind: "provider-observed-v1", threadId: "77" }, threadId: "77", }); resetCache(); const reloaded = await get(cacheFor(bucketKey, legacyStore), "9126"); expect(reloaded).toMatchObject({ body: "Pre-projection state message", messageId: "9126", }); expect(reloaded?.promptContextProjectionMarker).toBeUndefined(); expect(hasProviderObservedTelegramThreadBinding(reloaded, 77)).toBe(false); }); it("rejects unknown future persisted cache versions", async () => { const { bucketKey, store } = createMemoryStore(); const scopeKey = resolveTelegramMessageCachePersistentScopeKey("default"); const futureStore = entryStore(store, `${scopeKey}:default:7:9127`, { version: 2, sourceMessage: message(9127, "Nora", { chat: { id: 7, type: "group", title: "Ops" }, text: "Future state message", }), }); const cache = cacheFor(bucketKey, futureStore); expect(await get(cache, "9127")).toBeNull(); }); it("does not partially parse malformed persisted thread ids", async () => { const { bucketKey, entries, store } = createMemoryStore(); const cache = cacheFor(bucketKey, store); await record( cache, message(9126, "Nora", { chat: { id: 7, type: "supergroup", title: "Ops" }, date: 1_736_389_126, text: "State topic message", }), { threadId: 100 }, ); const [persistedKey, persistedValue] = onlyEntry(entries); expect(persistedValue.threadId).toBe("100"); entries.set(persistedKey, { ...persistedValue, threadId: "0x64" }); resetCache(); const recent = await recentBefore(cacheFor(bucketKey, store), "9127", { threadId: 100 }); expect(recent).toEqual([]); }); it("drops unsafe Telegram thread ids from live messages", async () => { const { bucketKey, entries, store } = createMemoryStore(); const cache = cacheFor(bucketKey, store); await record( cache, message(9127, "Nora", { chat: { id: 7, type: "supergroup", title: "Ops" }, date: 1_736_389_127, message_thread_id: Number.MAX_SAFE_INTEGER + 1, text: "Unsafe topic message", }), ); const [, persistedValue] = onlyEntry(entries); expect(persistedValue.threadId).toBeUndefined(); const topicRecent = await recentBefore(cache, "9128", { threadId: Number.MAX_SAFE_INTEGER + 1, }); const unscopedRecent = await recentBefore(cache, "9128"); expect(topicRecent).toEqual([]); expect(unscopedRecent.map((entry) => entry.messageId)).toEqual(["9127"]); }); it("does not use unsafe message ids as recent-before cutoffs", async () => { const cache = createTelegramMessageCache(); await record(cache, message(9124, "Nora", { date: 1_736_380_700, text: "State message" })); const recent = await recentBefore(cache, "9007199254740992"); expect(recent).toEqual([]); }); });