From 7068bc2ffe570c517dddbcbd3ae28037f4d35136 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 1 Aug 2026 23:57:31 -0700 Subject: [PATCH] fix(telegram): serialize approval ingress sessions (#117861) * fix(telegram): serialize approval ingress sessions * fix(telegram): narrow approval session tuples * fix(ingress): return lane reconciliation result --------- Co-authored-by: Peter Steinberger --- .../telegram/src/polling-session.test.ts | 103 +++ extensions/telegram/src/polling-session.ts | 9 +- .../telegram/src/sequential-key.test.ts | 70 ++ extensions/telegram/src/sequential-key.ts | 19 +- .../telegram/src/telegram-ingress-drain.ts | 162 ++++ extensions/telegram/src/webhook.test.ts | 800 +++++++++++++++++- extensions/telegram/src/webhook.ts | 12 +- src/channels/message/ingress-drain.ts | 8 + src/channels/message/ingress-queue.test.ts | 108 +++ src/channels/message/ingress-queue.ts | 41 +- 10 files changed, 1306 insertions(+), 26 deletions(-) diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index b3bf1146aa6a..3c2ab058e76b 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { expectDefined } from "@openclaw/normalization-core"; +import { Bot } from "grammy"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS as TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound"; import { @@ -307,6 +308,13 @@ function makeIsolatedBot(params?: { config: { use: vi.fn() }, }, init: vi.fn(params?.init ?? (async () => undefined)), + botInfo: { + id: 123, + is_bot: true, + first_name: "OpenClaw", + username: "openclaw_bot", + has_topics_enabled: false, + } as NonNullable[0]["botInfo"]>, handleUpdate: vi.fn(params?.handleUpdate ?? (async () => undefined)), stop: vi.fn(params?.stop ?? (async () => undefined)), }; @@ -481,6 +489,7 @@ function createPollingSession(params: { stallThresholdMs?: number; setStatus?: (patch: Omit) => void; isolatedIngress?: ConstructorParameters[0]["isolatedIngress"]; + botInfo?: ConstructorParameters[0]["botInfo"]; }) { return new TelegramPollingSession({ token: "tok", @@ -497,6 +506,7 @@ function createPollingSession(params: { stallThresholdMs: params.stallThresholdMs, setStatus: params.setStatus, isolatedIngress: params.isolatedIngress, + ...(params.botInfo ? { botInfo: params.botInfo } : {}), ...(params.createTelegramTransport ? { createTelegramTransport: params.createTelegramTransport } : {}), @@ -1127,6 +1137,99 @@ describe("TelegramPollingSession", () => { }); }); + it.each([ + { + name: "preserves a cached bot without making another getMe request", + seeded: true, + topicsEnabled: false, + expectedGetMeCalls: 0, + expectedLaneKey: "telegram:1234", + }, + { + name: "initializes an uncached bot with exactly one getMe request", + seeded: false, + topicsEnabled: true, + expectedGetMeCalls: 1, + expectedLaneKey: "telegram:1234:topic:42", + }, + ])( + "shares the installed grammY bot capability snapshot: $name", + async ({ seeded, topicsEnabled, expectedGetMeCalls, expectedLaneKey }) => { + await withTempSpool(async (tempDir) => { + const abort = new AbortController(); + const worker = createListeningIngressWorker(); + const botInfo = { + id: 123, + is_bot: true, + first_name: "OpenClaw", + username: "openclaw_bot", + has_topics_enabled: topicsEnabled, + } as NonNullable[0]["botInfo"]>; + const bot = new Bot("tok", seeded ? { botInfo } : undefined); + const getMe = vi.spyOn(bot.api, "getMe").mockResolvedValue(botInfo); + vi.spyOn(bot.api, "deleteWebhook").mockResolvedValue(true); + vi.spyOn(bot, "stop").mockResolvedValue(undefined); + let releaseHandler: (() => void) | undefined; + const handlerCompleted = new Promise((resolve) => { + releaseHandler = resolve; + }); + const handleUpdate = vi.spyOn(bot, "handleUpdate").mockImplementation(async () => { + await handlerCompleted; + }); + createTelegramBotMock.mockReturnValueOnce(bot); + const session = createPollingSession({ + abortSignal: abort.signal, + ...(seeded ? { botInfo } : {}), + isolatedIngress: { + enabled: true, + spoolDir: tempDir, + createWorker: worker.createWorker, + drainIntervalMs: 10, + }, + }); + const runPromise = session.runUntilAbort(); + try { + await waitForTelegramTestState(() => expect(worker.hasListener()).toBe(true)); + worker.emit({ + type: "update", + requestId: "topic-capability-1", + update: { + update_id: 143, + message: { + chat: { id: 1234, type: "private" }, + message_thread_id: 42, + text: "installed bot capability snapshot", + }, + }, + queued: 1, + }); + await waitForTelegramTestState(() => + expect(worker.ackSpooledUpdate).toHaveBeenCalledWith("topic-capability-1", { + ok: true, + updateId: 143, + }), + ); + await waitForTelegramTestState(() => expect(handleUpdate).toHaveBeenCalledOnce()); + const { database, kysely } = openTelegramSpoolTestKysely(tempDir); + const rows = executeSqliteQuerySync( + database.db, + kysely + .selectFrom("channel_ingress_events") + .select(["lane_key", "status"]) + .where("event_id", "=", String(143).padStart(16, "0")), + ).rows; + expect(rows).toMatchObject([{ lane_key: expectedLaneKey, status: "claimed" }]); + expect(getMe).toHaveBeenCalledTimes(expectedGetMeCalls); + expect(bot.botInfo).toBe(botInfo); + } finally { + releaseHandler?.(); + abort.abort(); + await runPromise; + } + }); + }, + ); + it("spools, persists the actual update id, then acknowledges", async () => { await withTempSpool(async (tempDir) => { const abort = new AbortController(); diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts index c114be6de850..d4b432467f12 100644 --- a/extensions/telegram/src/polling-session.ts +++ b/extensions/telegram/src/polling-session.ts @@ -353,6 +353,7 @@ export class TelegramPollingSession { /** Long-lived monitor for this session; stop only when the cycle ends. */ #getOrCreateSpooledMonitor(params: { bot: TelegramBot; + botInfo: TelegramBot["botInfo"]; spoolDir: string; pollIntervalMs: number; abortSignal?: AbortSignal; @@ -365,7 +366,7 @@ export class TelegramPollingSession { bot: params.bot, cfg: this.opts.config, accountId: this.opts.accountId, - botInfo: this.opts.botInfo, + botInfo: params.botInfo, adoptionStallTimeoutMs: this.#spooledUpdateHandlerTimeoutMs, pollIntervalMs: params.pollIntervalMs, ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), @@ -400,6 +401,9 @@ export class TelegramPollingSession { ); return shouldRetry ? "continue" : "exit"; } + // A pre-probed or cached bot may already be initialized; admission and replay + // must share grammY's actual capability snapshot instead of a second source. + const botInfo = bot.botInfo; const spoolDir = ingress.spoolDir ?? resolveTelegramIngressSpoolDir({ accountId: this.opts.accountId }); const workerFactory = ingress.createWorker ?? createTelegramIngressWorker; @@ -456,6 +460,7 @@ export class TelegramPollingSession { : this.opts.abortSignal; const ingressMonitor = this.#getOrCreateSpooledMonitor({ bot, + botInfo, spoolDir, pollIntervalMs: drainIntervalMs, ...(ingressAbortSignal ? { abortSignal: ingressAbortSignal } : {}), @@ -524,7 +529,7 @@ export class TelegramPollingSession { updateId = await writeTelegramSpooledUpdate({ spoolDir, update: message.update, - laneKey: telegramSpooledUpdateLaneKey(message.update, this.opts.botInfo), + laneKey: telegramSpooledUpdateLaneKey(message.update, botInfo), }); this.opts.log(`[telegram][diag] isolated polling update spooled updateId=${updateId}`); } catch (err: unknown) { diff --git a/extensions/telegram/src/sequential-key.test.ts b/extensions/telegram/src/sequential-key.test.ts index 68b36126a62f..a837c1d215bb 100644 --- a/extensions/telegram/src/sequential-key.test.ts +++ b/extensions/telegram/src/sequential-key.test.ts @@ -1,6 +1,8 @@ // Telegram tests cover sequential key plugin behavior. import type { Chat, Message } from "grammy/types"; import { describe, expect, it } from "vitest"; +import { buildTelegramApprovalCallbackData } from "./approval-callback-data.js"; +import { buildTelegramQuestionCallbackData } from "./question-callback-data.js"; import { getTelegramSequentialKey } from "./sequential-key.js"; const mockChat = (chat: Pick & Partial>): Chat => @@ -22,6 +24,26 @@ describe("getTelegramSequentialKey", () => { message_thread_id: 9, }), }, + "telegram:123", + ], + [ + { + me: { has_topics_enabled: false } as never, + message: mockMessage({ + chat: mockChat({ id: 123, type: "private" }), + message_thread_id: 9, + }), + }, + "telegram:123", + ], + [ + { + me: { has_topics_enabled: true } as never, + message: mockMessage({ + chat: mockChat({ id: 123, type: "private" }), + message_thread_id: 9, + }), + }, "telegram:123:topic:9", ], [ @@ -282,6 +304,40 @@ describe("getTelegramSequentialKey", () => { }, "telegram:789:approval", ], + ...(["exec", "plugin"] as const).map( + (approvalKind): [Parameters[0], string] => [ + { + update: { + callback_query: { + message: mockMessage({ chat: mockChat({ id: 654 }) }), + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind, + approvalId: "signed-approval", + decision: "allow-once", + }), + }, + }, + }, + "telegram:654:approval", + ], + ), + [ + { + update: { + callback_query: { + message: mockMessage({ chat: mockChat({ id: 655 }) }), + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + })?.replace(":o:", ":z:"), + }, + }, + }, + "telegram:655:approval", + ], [ { update: { @@ -293,6 +349,20 @@ describe("getTelegramSequentialKey", () => { }, "telegram:321:question", ], + [ + { + update: { + callback_query: { + message: mockMessage({ chat: mockChat({ id: 322 }) }), + data: buildTelegramQuestionCallbackData({ + questionId: "ask_0123456789abcdef0123456789abcdef", + optionIndex: 2, + })?.replace(/:2$/, ":9"), + }, + }, + }, + "telegram:322:question", + ], [ { update: { diff --git a/extensions/telegram/src/sequential-key.ts b/extensions/telegram/src/sequential-key.ts index a54f611da968..8bf769b15191 100644 --- a/extensions/telegram/src/sequential-key.ts +++ b/extensions/telegram/src/sequential-key.ts @@ -10,11 +10,14 @@ import { isAbortRequestText, isBtwRequestText, } from "openclaw/plugin-sdk/command-primitives-runtime"; +import { hasTelegramApprovalCallbackPrefix } from "./approval-callback-data.js"; import { + resolveTelegramBotHasTopicsEnabled, resolveTelegramForumThreadId, resolveTelegramMessageForumFlagHint, + shouldUseTelegramDmThreadSession, } from "./bot/helpers.js"; -import { parseTelegramQuestionCallbackData } from "./question-callback-data.js"; +import { hasTelegramQuestionCallbackPrefix } from "./question-callback-data.js"; const TELEGRAM_READ_ONLY_STATUS_COMMAND_KEYS = new Set([ "commands", @@ -180,13 +183,16 @@ export function getTelegramSequentialKey(ctx: TelegramSequentialKeyContext): str return "telegram:btw"; } const callbackData = ctx.update?.callback_query?.data; - if (parseTelegramQuestionCallbackData(callbackData)) { + if (hasTelegramQuestionCallbackPrefix(callbackData)) { if (typeof chatId === "number") { return `telegram:${chatId}:question`; } return "telegram:question"; } - if (callbackData && parseExecApprovalCommandText(callbackData) !== null) { + if ( + hasTelegramApprovalCallbackPrefix(callbackData) || + (callbackData && parseExecApprovalCommandText(callbackData) !== null) + ) { if (typeof chatId === "number") { return `telegram:${chatId}:approval`; } @@ -201,7 +207,12 @@ export function getTelegramSequentialKey(ctx: TelegramSequentialKeyContext): str }); const threadId = isGroup ? resolveTelegramForumThreadId({ isForum, messageThreadId }) - : messageThreadId; + : shouldUseTelegramDmThreadSession({ + dmThreadId: messageThreadId, + botHasTopicsEnabled: resolveTelegramBotHasTopicsEnabled(ctx.me), + }) + ? messageThreadId + : undefined; if (typeof chatId === "number") { return threadId != null ? `telegram:${chatId}:topic:${threadId}` : `telegram:${chatId}`; } diff --git a/extensions/telegram/src/telegram-ingress-drain.ts b/extensions/telegram/src/telegram-ingress-drain.ts index dadb935ac6fd..c3dda25c8f37 100644 --- a/extensions/telegram/src/telegram-ingress-drain.ts +++ b/extensions/telegram/src/telegram-ingress-drain.ts @@ -4,15 +4,26 @@ import { DEFAULT_INGRESS_ADOPTION_STALL_MS, type ChannelIngressMonitorLifecycle, type ChannelIngressQueue, + type ChannelIngressQueueRecord, } from "openclaw/plugin-sdk/channel-outbound"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { clampPositiveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; +import { + hasTelegramApprovalCallbackPrefix, + parseTelegramApprovalCallbackData, +} from "./approval-callback-data.js"; import type { TelegramBotInfo } from "./bot-info.js"; import { runWithTelegramSpooledReplayUpdate, type TelegramMessageProcessingResult, } from "./bot-processing-outcome.js"; +import { + resolveTelegramForumThreadId, + resolveTelegramMessageForumFlagHint, +} from "./bot/helpers.js"; +import { hasTelegramQuestionCallbackPrefix } from "./question-callback-data.js"; import { getTelegramSequentialKey } from "./sequential-key.js"; +import { normalizeTelegramStateAccountId } from "./state-account-id.js"; import { resolveTelegramIngressNonRetryableFailure } from "./telegram-ingress-non-retryable.js"; import { resolveTelegramUpdateId, telegramQueueEventId } from "./telegram-ingress-spool.js"; import { @@ -62,6 +73,149 @@ function inspectTelegramSpooledUpdate(update: unknown, botInfo?: TelegramBotInfo }; } +function isNonemptyTelegramCallbackValue(value: unknown): value is string { + return typeof value === "string" && value.trim().length > 0; +} + +function isBoundedTelegramCallbackData(value: unknown): value is string { + return isNonemptyTelegramCallbackValue(value) && Buffer.byteLength(value, "utf8") <= 64; +} + +function canReconcileTelegramLegacyLane(params: { + record: ChannelIngressQueueRecord; + storedLaneKey: string; + derivedLaneKey: string; + accountId: string; + botInfo?: TelegramBotInfo; +}): boolean { + if ( + params.record.channelId !== "telegram" || + params.record.accountId !== normalizeTelegramStateAccountId(params.accountId) + ) { + return false; + } + const update = params.record.payload.update; + if (!update || typeof update !== "object") { + return false; + } + type TelegramLaneMessage = { + chat?: { id?: unknown; is_forum?: unknown; type?: unknown }; + business_connection_id?: unknown; + date?: unknown; + direct_messages_topic?: unknown; + from?: { id?: unknown; is_bot?: unknown }; + guest_query_id?: unknown; + message_id?: unknown; + message_thread_id?: unknown; + is_topic_message?: unknown; + sender_chat?: unknown; + }; + type TelegramLaneCallback = { + id?: unknown; + data?: unknown; + chat_instance?: unknown; + inline_message_id?: unknown; + from?: { id?: unknown; is_bot?: unknown }; + message?: TelegramLaneMessage; + }; + const candidate = update as { + message?: TelegramLaneMessage; + edited_message?: TelegramLaneMessage; + callback_query?: TelegramLaneCallback; + }; + const callback = candidate.callback_query; + if (callback !== undefined) { + if (!callback || typeof callback !== "object") { + return false; + } + const senderId = callback.from?.id; + const callbackMessage = callback.message; + if ( + candidate.message !== undefined || + candidate.edited_message !== undefined || + !isNonemptyTelegramCallbackValue(callback.id) || + !isBoundedTelegramCallbackData(callback.data) || + !isNonemptyTelegramCallbackValue(callback.chat_instance) || + callback.inline_message_id !== undefined || + typeof senderId !== "number" || + !Number.isSafeInteger(senderId) || + senderId <= 0 || + callback.from?.is_bot !== false || + !params.botInfo || + callbackMessage?.from?.id !== params.botInfo.id || + callbackMessage.from.is_bot !== true || + callbackMessage.business_connection_id !== undefined || + callbackMessage.guest_query_id !== undefined || + callbackMessage.sender_chat !== undefined || + callbackMessage.direct_messages_topic !== undefined || + typeof callbackMessage.date !== "number" || + !Number.isSafeInteger(callbackMessage.date) || + callbackMessage.date <= 0 || + typeof callbackMessage.message_id !== "number" || + !Number.isSafeInteger(callbackMessage.message_id) || + callbackMessage.message_id <= 0 + ) { + return false; + } + } + const message = candidate.message ?? candidate.edited_message ?? callback?.message; + if (message == null) { + return false; + } + const chatId = message?.chat?.id; + const chatType = message?.chat?.type; + const threadId = message?.message_thread_id; + const callbackData = typeof callback?.data === "string" ? callback.data : undefined; + const typedApproval = parseTelegramApprovalCallbackData(callbackData); + const isPrivateChat = chatType === "private" && typeof chatId === "number" && chatId > 0; + const isGroupChat = + (chatType === "group" || chatType === "supergroup") && typeof chatId === "number" && chatId < 0; + const hasValidThreadId = + typeof threadId === "number" && Number.isSafeInteger(threadId) && threadId > 0; + if ( + typeof chatId !== "number" || + !Number.isSafeInteger(chatId) || + (typedApproval ? !isPrivateChat && !isGroupChat : !isPrivateChat) || + (!typedApproval && !hasValidThreadId) || + (typedApproval && threadId !== undefined && !hasValidThreadId) + ) { + return false; + } + const baseLaneKey = `telegram:${chatId}`; + const legacyThreadId = isGroupChat + ? resolveTelegramForumThreadId({ + isForum: resolveTelegramMessageForumFlagHint({ + chatType, + isForum: typeof message.chat?.is_forum === "boolean" ? message.chat.is_forum : undefined, + isTopicMessage: + typeof message.is_topic_message === "boolean" ? message.is_topic_message : undefined, + }), + messageThreadId: hasValidThreadId ? threadId : undefined, + }) + : hasValidThreadId + ? threadId + : undefined; + const topicLaneKey = legacyThreadId ? `${baseLaneKey}:topic:${legacyThreadId}` : undefined; + const canonicalLaneKey = typedApproval + ? `${baseLaneKey}:approval` + : params.botInfo?.has_topics_enabled === true + ? topicLaneKey + : baseLaneKey; + const previousLaneKey = canonicalLaneKey === baseLaneKey ? topicLaneKey : baseLaneKey; + + // Signed releases stored typed approvals in ordinary chat/forum lanes. Admit only a + // valid, authenticated owner callback into its dedicated privileged lane. + return ( + (typedApproval + ? params.storedLaneKey === baseLaneKey || params.storedLaneKey === topicLaneKey + : !hasTelegramApprovalCallbackPrefix(callbackData) && + !hasTelegramQuestionCallbackPrefix(callbackData) && + params.storedLaneKey === previousLaneKey) && + params.derivedLaneKey === canonicalLaneKey && + telegramSpooledLaneKey(update, params.botInfo) === canonicalLaneKey + ); +} + export type TelegramIngressDrainLifecycle = Omit< ChannelIngressMonitorLifecycle, "admission" | "onFailed" @@ -229,6 +383,14 @@ export function createTelegramIngressMonitor(params: CreateTelegramIngressMonito ...(params.botInfo?.username ? { botUsername: params.botInfo.username } : {}), }), deriveLaneKey: (record) => telegramSpooledLaneKey(record.payload.update, params.botInfo), + reconcileStoredLaneKey: (record, storedLaneKey, derivedLaneKey) => + canReconcileTelegramLegacyLane({ + record, + storedLaneKey, + derivedLaneKey, + accountId: params.accountId, + botInfo: params.botInfo, + }), ...(params.onLog ? { onLog: params.onLog } : {}), }, ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), diff --git a/extensions/telegram/src/webhook.test.ts b/extensions/telegram/src/webhook.test.ts index 397f1241dbaf..592270bccd94 100644 --- a/extensions/telegram/src/webhook.test.ts +++ b/extensions/telegram/src/webhook.test.ts @@ -12,6 +12,7 @@ import { } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { WEBHOOK_RATE_LIMIT_DEFAULTS } from "openclaw/plugin-sdk/webhook-ingress"; import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { buildTelegramApprovalCallbackData } from "./approval-callback-data.js"; import { createTelegramSpooledReplayDeferredParticipant, type TelegramSpooledReplayDeferredParticipant, @@ -20,7 +21,7 @@ import { import { setTelegramRuntime } from "./runtime.js"; import { clearTelegramRuntimeForTest as clearTelegramRuntime } from "./runtime.test-support.js"; import type { TelegramRuntime } from "./runtime.types.js"; -import { writeTelegramSpooledUpdate } from "./telegram-ingress-spool.js"; +import { openTelegramIngressQueue, writeTelegramSpooledUpdate } from "./telegram-ingress-spool.js"; import { listTelegramSpooledUpdateClaims, listTelegramSpooledUpdates, @@ -33,9 +34,17 @@ const setWebhookSpy = vi.hoisted(() => vi.fn()); const deleteWebhookSpy = vi.hoisted(() => vi.fn(async () => true)); const initSpy = vi.hoisted(() => vi.fn(async () => undefined)); const stopSpy = vi.hoisted(() => vi.fn()); +const webhookBotInfo = vi.hoisted(() => ({ + id: 123, + is_bot: true as const, + first_name: "OpenClaw", + username: "openclaw_bot", + has_topics_enabled: false, +})); const createTelegramBotSpy = vi.hoisted(() => vi.fn(() => ({ init: initSpy, + botInfo: webhookBotInfo, handleUpdate: handleUpdateSpy, api: { setWebhook: setWebhookSpy, deleteWebhook: deleteWebhookSpy }, stop: stopSpy, @@ -170,6 +179,22 @@ function requireWebhookSpoolDir(): string { return webhookSpoolDir; } +function createTelegramPrivateTopicCallback(updateId: number) { + return { + id: `callback-${updateId}`, + data: "cmd:option_a", + chat_instance: "telegram-private-chat-1234", + from: { id: 111, is_bot: false as const, first_name: "Ada" }, + message: { + chat: { id: 1234, type: "private" as const }, + date: 1_736_380_800, + from: { id: webhookBotInfo.id, is_bot: true as const, first_name: "OpenClaw" }, + message_id: 10, + message_thread_id: 42, + }, + }; +} + function resetTelegramWebhookMocks(): void { handleUpdateSpy.mockReset(); handleUpdateSpy.mockImplementation((..._args: unknown[]): unknown => undefined); @@ -182,9 +207,11 @@ function resetTelegramWebhookMocks(): void { stopSpy.mockReset(); resolveTelegramTransportSpy.mockClear(); transportCloseSpies.length = 0; + webhookBotInfo.has_topics_enabled = false; createTelegramBotSpy.mockReset(); createTelegramBotSpy.mockImplementation(() => ({ init: initSpy, + botInfo: webhookBotInfo, handleUpdate: handleUpdateSpy, api: { setWebhook: setWebhookSpy, deleteWebhook: deleteWebhookSpy }, stop: stopSpy, @@ -1183,6 +1210,700 @@ describe("startTelegramWebhook", () => { ); }); + it.each([ + { + topicsEnabled: false, + persistedLaneKey: "telegram:1234:topic:42", + canonicalLaneKey: "telegram:1234", + }, + { + topicsEnabled: true, + persistedLaneKey: "telegram:1234", + canonicalLaneKey: "telegram:1234:topic:42", + }, + ])( + "replays acknowledged legacy DM lanes after restart when topic capability is $topicsEnabled", + async ({ topicsEnabled, persistedLaneKey, canonicalLaneKey }) => { + webhookBotInfo.has_topics_enabled = topicsEnabled; + const firstUpdate = { + update_id: 130, + message: { + chat: { id: 1234, type: "private" }, + message_id: 1, + message_thread_id: 42, + text: "accepted before restart", + }, + }; + const secondUpdate = { + update_id: 131, + message: { + chat: { id: 1234, type: "private" }, + message_id: 2, + ...(topicsEnabled ? { message_thread_id: 42 } : {}), + text: "accepted after the first event", + }, + }; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: firstUpdate, + laneKey: persistedLaneKey, + }); + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: secondUpdate, + laneKey: canonicalLaneKey, + }); + expect( + (await openTelegramIngressQueue(requireWebhookSpoolDir()).listPending()).map( + (record) => record.laneKey, + ), + ).toEqual([persistedLaneKey, canonicalLaneKey]); + closeOpenClawStateDatabaseForTest(); + + const seenUpdateIds: number[] = []; + let releaseFirstUpdate: (() => void) | undefined; + const firstUpdateCompleted = new Promise((resolve) => { + releaseFirstUpdate = resolve; + }); + handleUpdateSpy.mockImplementation(async (update: unknown) => { + const updateId = (update as { update_id: number }).update_id; + seenUpdateIds.push(updateId); + if (updateId === firstUpdate.update_id) { + await firstUpdateCompleted; + } + }); + + try { + await withStartedWebhook( + { + accountId: "test", + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async () => { + await waitForWebhookState(() => expect(seenUpdateIds).toEqual([130])); + await sleep(25); + expect(seenUpdateIds).toEqual([130]); + + releaseFirstUpdate?.(); + await waitForWebhookState(() => expect(seenUpdateIds).toEqual([130, 131])); + await waitForWebhookState(async () => + expect( + await listTelegramSpooledUpdates({ spoolDir: requireWebhookSpoolDir() }), + ).toEqual([]), + ); + expect(await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.()).toEqual( + [], + ); + expect(handleUpdateSpy).toHaveBeenCalledTimes(2); + }, + ); + } finally { + releaseFirstUpdate?.(); + } + }, + ); + + it.each([ + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:1234:topic:42", + }, + { + approvalKind: "plugin" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:1234:topic:42", + }, + { approvalKind: "exec" as const, topicsEnabled: true, persistedLaneKey: "telegram:1234" }, + { approvalKind: "plugin" as const, topicsEnabled: true, persistedLaneKey: "telegram:1234" }, + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:1234", + hasThread: false, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:1234", + hasThread: false, + }, + { + approvalKind: "exec" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:1234", + hasThread: false, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:1234", + hasThread: false, + }, + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:-1234", + hasThread: false, + chatId: -1234, + chatType: "group" as const, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:-1234", + hasThread: false, + chatId: -1234, + chatType: "group" as const, + }, + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:-1001234", + hasThread: false, + chatId: -1001234, + chatType: "supergroup" as const, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:-1001234", + hasThread: false, + chatId: -1001234, + chatType: "supergroup" as const, + }, + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:-1001234:topic:42", + chatId: -1001234, + chatType: "supergroup" as const, + isForum: true, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:-1001234:topic:42", + chatId: -1001234, + chatType: "supergroup" as const, + isForum: true, + }, + { + approvalKind: "exec" as const, + topicsEnabled: false, + persistedLaneKey: "telegram:-1001234:topic:1", + hasThread: false, + chatId: -1001234, + chatType: "supergroup" as const, + isForum: true, + }, + { + approvalKind: "plugin" as const, + topicsEnabled: true, + persistedLaneKey: "telegram:-1001234:topic:1", + hasThread: false, + chatId: -1001234, + chatType: "supergroup" as const, + isForum: true, + }, + ])( + "replays acknowledged typed $approvalKind approvals in their privileged lane (chat=$chatType, topics=$topicsEnabled, thread=$hasThread)", + async ({ + approvalKind, + topicsEnabled, + persistedLaneKey, + hasThread, + chatId, + chatType, + isForum, + }) => { + webhookBotInfo.has_topics_enabled = topicsEnabled; + const callback = createTelegramPrivateTopicCallback(142); + const expectedChatId = chatId ?? callback.message.chat.id; + const update = { + update_id: 142, + callback_query: { + ...callback, + message: { + ...callback.message, + chat: { + ...callback.message.chat, + id: expectedChatId, + type: chatType ?? callback.message.chat.type, + ...(isForum ? { is_forum: true } : {}), + }, + ...(hasThread === false ? { message_thread_id: undefined } : {}), + }, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind, + approvalId: "signed-approval", + decision: "allow-once", + }), + }, + }; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update, + laneKey: persistedLaneKey, + }); + closeOpenClawStateDatabaseForTest(); + + handleUpdateSpy.mockImplementationOnce(async () => { + expect(await openTelegramIngressQueue(requireWebhookSpoolDir()).listClaims()).toMatchObject( + [{ laneKey: `telegram:${expectedChatId}:approval` }], + ); + }); + + await withStartedWebhook( + { accountId: "test", secret: TELEGRAM_SECRET, path: TELEGRAM_WEBHOOK_PATH }, + async () => { + await waitForWebhookState(() => expect(handleUpdateSpy).toHaveBeenCalledOnce()); + expect(handleUpdateSpy).toHaveBeenCalledWith(update); + await waitForWebhookState(async () => + expect(await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.()).toEqual( + [], + ), + ); + }, + ); + }, + ); + + it.each([ + { + topicsEnabled: false, + persistedLaneKey: "telegram:1234:topic:42", + }, + { + topicsEnabled: true, + persistedLaneKey: "telegram:1234", + }, + { + topicsEnabled: false, + persistedLaneKey: "telegram:1234:topic:42", + callbackIdentityLength: 128, + }, + ])( + "replays legitimate private callbacks after a topic-capability transition ($topicsEnabled)", + async ({ topicsEnabled, persistedLaneKey, callbackIdentityLength }) => { + webhookBotInfo.has_topics_enabled = topicsEnabled; + const update = { + update_id: 140, + callback_query: { + ...createTelegramPrivateTopicCallback(140), + ...(callbackIdentityLength + ? { + id: "i".repeat(callbackIdentityLength), + chat_instance: "c".repeat(callbackIdentityLength), + } + : {}), + }, + }; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update, + laneKey: persistedLaneKey, + }); + closeOpenClawStateDatabaseForTest(); + + await withStartedWebhook( + { + accountId: "test", + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async () => { + await waitForWebhookState(() => expect(handleUpdateSpy).toHaveBeenCalledOnce()); + expect(handleUpdateSpy).toHaveBeenCalledWith(update); + expect(await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.()).toEqual( + [], + ); + }, + ); + }, + ); + + it.each([ + { + name: "a bot callback sender", + mutate: (callback: ReturnType) => ({ + ...callback, + from: { ...callback.from, is_bot: true }, + }), + }, + { + name: "an invalid callback sender", + mutate: (callback: ReturnType) => ({ + ...callback, + from: { ...callback.from, id: 0 }, + }), + }, + { + name: "a foreign bot-authored message", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { + ...callback.message, + from: { ...callback.message.from, id: 999 }, + }, + }), + }, + { + name: "an inaccessible callback message", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, date: 0 }, + }), + }, + { + name: "an independent business chat", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, business_connection_id: "business-1234" }, + }), + }, + { + name: "an independent guest chat", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, guest_query_id: "guest-1234" }, + }), + }, + { + name: "a message sent by another chat", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, sender_chat: { id: -1234, type: "channel" } }, + }), + }, + { + name: "a direct-messages topic from another surface", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, direct_messages_topic: { topic_id: 42 } }, + }), + }, + { + name: "an inline callback message", + mutate: (callback: ReturnType) => ({ + ...callback, + inline_message_id: "inline-message-141", + }), + }, + { + name: "an oversized callback payload", + mutate: (callback: ReturnType) => ({ + ...callback, + data: "x".repeat(65), + }), + }, + { + name: "a missing chat instance", + mutate: (callback: ReturnType) => ({ + ...callback, + chat_instance: "", + }), + }, + { + name: "a reserved question callback", + mutate: (callback: ReturnType) => ({ + ...callback, + data: "tgq1:ask_0123456789abcdef0123456789abcdef:1", + }), + }, + { + name: "a reserved approval callback", + mutate: (callback: ReturnType) => ({ + ...callback, + data: "/approve exec:def456 deny", + }), + }, + { + name: "a malformed signed approval decision", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + })?.replace(":o:", ":z:"), + }), + }, + { + name: "a malformed signed approval without a topic", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + })?.replace(":o:", ":z:"), + message: { ...callback.message, message_thread_id: undefined }, + }), + }, + { + name: "a malformed signed approval kind", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + })?.replace(":e:", ":x:"), + }), + }, + { + name: "a signed approval missing its canonical identifier", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "plugin", + approvalId: "signed-approval", + decision: "deny", + })?.replace(/signed-approval$/, ""), + }), + }, + { + name: "a signed approval in another business namespace", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + }), + message: { ...callback.message, business_connection_id: "business-1234" }, + }), + }, + { + name: "a signed approval in another guest namespace", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "plugin", + approvalId: "signed-approval", + decision: "deny", + }), + message: { ...callback.message, guest_query_id: "guest-1234" }, + }), + }, + { + name: "a signed approval from a foreign bot", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + }), + message: { ...callback.message, from: { ...callback.message.from, id: 999 } }, + }), + }, + { + name: "a signed approval from an invalid actor", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "plugin", + approvalId: "signed-approval", + decision: "deny", + }), + from: { ...callback.from, id: 0 }, + }), + }, + { + name: "a signed approval for another chat", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + }), + message: { ...callback.message, chat: { ...callback.message.chat, id: 9999 } }, + }), + }, + { + name: "an inline signed approval", + mutate: (callback: ReturnType) => ({ + ...callback, + data: buildTelegramApprovalCallbackData({ + type: "approval", + approvalKind: "exec", + approvalId: "signed-approval", + decision: "allow-once", + }), + inline_message_id: "inline-message-141", + }), + }, + { + name: "a malformed reserved question callback", + mutate: (callback: ReturnType) => ({ + ...callback, + data: "tgq1:ask_0123456789abcdef0123456789abcdef:9", + }), + }, + { + name: "a reserved question callback without a topic", + mutate: (callback: ReturnType) => ({ + ...callback, + data: "tgq1:ask_0123456789abcdef0123456789abcdef:1", + message: { ...callback.message, message_thread_id: undefined }, + }), + }, + { + name: "an ordinary callback without a topic", + mutate: (callback: ReturnType) => ({ + ...callback, + message: { ...callback.message, message_thread_id: undefined }, + }), + }, + ])("does not authorize durable-lane reconciliation for $name", async ({ mutate }) => { + const laneKey = "telegram:1234:topic:42"; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: { update_id: 141, callback_query: mutate(createTelegramPrivateTopicCallback(141)) }, + laneKey, + }); + closeOpenClawStateDatabaseForTest(); + + await withStartedWebhook( + { + accountId: "test", + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async () => { + await waitForWebhookState(async () => + expect( + await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.({ limit: "all" }), + ).toMatchObject([{ reason: "invalid-event", laneKey }]), + ); + expect(handleUpdateSpy).not.toHaveBeenCalled(); + }, + ); + }); + + it.each(["telegram:9999:topic:42", "telegram:1234:topic:99", "telegram:1234:control"])( + "rejects persisted DM lanes outside the signed upgrade contract (%s)", + async (laneKey) => { + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update: { + update_id: 132, + message: { + chat: { id: 1234, type: "private" }, + message_id: 1, + message_thread_id: 42, + text: "reject mismatched durable identity", + }, + }, + laneKey, + }); + closeOpenClawStateDatabaseForTest(); + + await withStartedWebhook( + { + accountId: "test", + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async () => { + await waitForWebhookState(async () => + expect( + await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.({ + limit: "all", + }), + ).toMatchObject([{ reason: "invalid-event", laneKey }]), + ); + expect(handleUpdateSpy).not.toHaveBeenCalled(); + expect(await listTelegramSpooledUpdates({ spoolDir: requireWebhookSpoolDir() })).toEqual( + [], + ); + }, + ); + }, + ); + + it.each([ + { + name: "another account", + accountId: "other", + update: { + update_id: 133, + message: { + chat: { id: 1234, type: "private" }, + message_id: 1, + message_thread_id: 42, + text: "wrong account", + }, + }, + }, + { + name: "a malformed callback query", + accountId: "test", + update: { + update_id: 134, + callback_query: { + data: "unrecognized-callback", + message: { + chat: { id: 1234, type: "private" }, + message_id: 1, + message_thread_id: 42, + }, + }, + }, + }, + { + name: "a group message", + accountId: "test", + update: { + update_id: 135, + message: { + chat: { id: 1234, type: "group" }, + message_id: 1, + message_thread_id: 42, + text: "wrong chat kind", + }, + }, + }, + ])("rejects legacy lane reconciliation for $name", async ({ accountId, update }) => { + const laneKey = "telegram:1234:topic:42"; + await writeTelegramSpooledUpdate({ + spoolDir: requireWebhookSpoolDir(), + update, + laneKey, + }); + closeOpenClawStateDatabaseForTest(); + + await withStartedWebhook( + { + accountId, + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async () => { + await waitForWebhookState(async () => + expect( + await openTelegramIngressQueue(requireWebhookSpoolDir()).listFailed?.({ limit: "all" }), + ).toMatchObject([{ reason: "invalid-event", laneKey }]), + ); + expect(handleUpdateSpy).not.toHaveBeenCalled(); + }, + ); + }); + it("keeps a webhook lane guarded while claimed completion retries", async () => { let completeAttempts = 0; let releaseCompletion: (() => void) | undefined; @@ -1673,6 +2394,83 @@ describe("startTelegramWebhook", () => { ); }); + it.each([ + { topicsEnabled: false, shouldSerialize: true }, + { topicsEnabled: true, shouldSerialize: false }, + ])( + "matches DM session serialization to initialized bot topic capability ($topicsEnabled)", + async ({ topicsEnabled, shouldSerialize }) => { + webhookBotInfo.has_topics_enabled = topicsEnabled; + const seenUpdateIds: number[] = []; + let releaseFirstUpdate: (() => void) | undefined; + const firstUpdateCompleted = new Promise((resolve) => { + releaseFirstUpdate = resolve; + }); + handleUpdateSpy.mockImplementation(async (update: unknown) => { + const updateId = (update as { update_id: number }).update_id; + seenUpdateIds.push(updateId); + if (updateId === 100) { + await firstUpdateCompleted; + } + }); + + await withStartedWebhook( + { + secret: TELEGRAM_SECRET, + path: TELEGRAM_WEBHOOK_PATH, + }, + async ({ port }) => { + const url = webhookUrl(port, TELEGRAM_WEBHOOK_PATH); + const firstUpdate = { + update_id: 100, + message: { + chat: { id: 1234, type: "private" }, + message_id: 1, + text: "first", + }, + }; + const secondUpdate = { + update_id: 101, + message: { + chat: { id: 1234, type: "private" }, + message_id: 2, + message_thread_id: 42, + text: "second", + }, + }; + + try { + const firstResponse = await postWebhookJson({ + url, + payload: JSON.stringify(firstUpdate), + secret: TELEGRAM_SECRET, + }); + expect(firstResponse.status).toBe(200); + await waitForWebhookState(() => expect(seenUpdateIds).toEqual([100])); + + const secondResponse = await postWebhookJson({ + url, + payload: JSON.stringify(secondUpdate), + secret: TELEGRAM_SECRET, + }); + expect(secondResponse.status).toBe(200); + + if (shouldSerialize) { + await sleep(25); + expect(seenUpdateIds).toEqual([100]); + } else { + await waitForWebhookState(() => expect(seenUpdateIds).toEqual([100, 101])); + } + } finally { + releaseFirstUpdate?.(); + } + + await waitForWebhookState(() => expect(seenUpdateIds).toEqual([100, 101])); + }, + ); + }, + ); + it("keeps webhook payload readable across multiple delayed reads", async () => { const seenPayloads: string[] = []; const delayedHandler = async (update: unknown) => { diff --git a/extensions/telegram/src/webhook.ts b/extensions/telegram/src/webhook.ts index 7bb9dd417c47..f58b5acdd1ca 100644 --- a/extensions/telegram/src/webhook.ts +++ b/extensions/telegram/src/webhook.ts @@ -36,10 +36,10 @@ import { withTelegramApiErrorLogging } from "./api-logging.js"; import { createTelegramBot } from "./bot.js"; import { resolveTelegramTransport } from "./fetch.js"; import { isRetryableTelegramApiError } from "./network-errors.js"; -import { getTelegramSequentialKey } from "./sequential-key.js"; import { createTelegramTransportIngressMonitor } from "./telegram-ingress-drain-factory.js"; import { resolveTelegramIngressSpoolDir, + telegramSpooledUpdateLaneKey, writeTelegramSpooledUpdate, } from "./telegram-ingress-spool.js"; import { createTelegramWebhookStatusPublisher } from "./webhook-status.js"; @@ -295,12 +295,6 @@ function resolveTelegramWebhookRateLimitKey( return `${path}:${resolveTelegramWebhookClientIp(req, config)}`; } -function resolveWebhookSpooledUpdateLaneKey(update: unknown): string { - return getTelegramSequentialKey({ - update: update as Parameters[0]["update"], - }); -} - export async function startTelegramWebhook(opts: { token: string; accountId?: string; @@ -376,6 +370,7 @@ export async function startTelegramWebhook(opts: { await closeTransportOnce(); throw err; } + const botInfo = bot.botInfo; const telegramWebhookRateLimiter = createFixedWindowRateLimiter({ windowMs: WEBHOOK_RATE_LIMIT_DEFAULTS.windowMs, maxRequests: WEBHOOK_RATE_LIMIT_DEFAULTS.maxRequests, @@ -400,6 +395,7 @@ export async function startTelegramWebhook(opts: { webhookIngressMonitor = createTelegramTransportIngressMonitor({ spoolDir, bot, + botInfo, cfg: opts.config ?? {}, accountId: opts.accountId ?? "default", // Pre-migration product default: 25m claim→adoption stall for webhook. @@ -483,7 +479,7 @@ export async function startTelegramWebhook(opts: { await writeTelegramSpooledUpdate({ spoolDir, update: body.value, - laneKey: resolveWebhookSpooledUpdateLaneKey(body.value), + laneKey: telegramSpooledUpdateLaneKey(body.value, botInfo), }); // Enqueue duplicate detection makes Telegram webhook retries idempotent: // re-posted update_ids map to the same spool row and still ack fast. diff --git a/src/channels/message/ingress-drain.ts b/src/channels/message/ingress-drain.ts index bdde418731bd..a54cabbd6925 100644 --- a/src/channels/message/ingress-drain.ts +++ b/src/channels/message/ingress-drain.ts @@ -101,6 +101,11 @@ export type CreateChannelIngressDrainOptions< pendingEvent: ChannelIngressQueueClaim, ) => boolean | Promise; deriveLaneKey?: (record: ChannelIngressQueueRecord) => string | undefined; + reconcileStoredLaneKey?: ( + record: ChannelIngressQueueRecord, + storedLaneKey: string, + derivedLaneKey: string, + ) => boolean; ownerId?: string; adoptionStallTimeoutMs?: number; claimLeaseMs?: number; @@ -756,6 +761,9 @@ export function createChannelIngressDrain< scanLimit, candidateIds, deriveLaneKey: options.deriveLaneKey, + ...(options.reconcileStoredLaneKey + ? { reconcileStoredLaneKey: options.reconcileStoredLaneKey } + : {}), }); if (!claimed) { break; diff --git a/src/channels/message/ingress-queue.test.ts b/src/channels/message/ingress-queue.test.ts index a03768132f4e..0e56cffa6a02 100644 --- a/src/channels/message/ingress-queue.test.ts +++ b/src/channels/message/ingress-queue.test.ts @@ -289,6 +289,114 @@ describe("channel ingress queue", () => { }); }); + it("preserves durable lanes when a channel derives ephemeral claim lanes", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createTestIngressQueue<{ text: string }>(stateDir, { now: () => clock++ }); + + await queue.enqueue("message-1", { text: "debounced" }, { laneKey: "chat:123" }); + + const claimed = await queue.claimNext({ + ownerId: "imessage-worker", + deriveLaneKey: (record) => `${record.laneKey ?? "event"}:${record.id}`, + }); + + expect(claimed?.laneKey).toBe("chat:123"); + expect((await queue.listClaims())[0]?.laneKey).toBe("chat:123"); + }); + }); + + it("reconciles opted-in persisted lanes before blocking and claiming", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ }); + + await queue.enqueue( + "a", + { lane: "chat:123" }, + { laneKey: "chat:123:topic:7", receivedAt: 1 }, + ); + await queue.enqueue( + "b", + { lane: "chat:456" }, + { laneKey: "chat:456:topic:9", receivedAt: 2 }, + ); + + const claimed = await queue.claimNext({ + ownerId: "worker", + blockedLaneKeys: ["chat:123"], + deriveLaneKey: (record) => record.payload.lane, + reconcileStoredLaneKey: (_record, storedLaneKey, derivedLaneKey) => + storedLaneKey === `${derivedLaneKey}:topic:7` || + storedLaneKey === `${derivedLaneKey}:topic:9`, + }); + + expect(claimed?.id).toBe("b"); + expect(claimed?.laneKey).toBe("chat:456"); + expect((await queue.listClaims())[0]?.laneKey).toBe("chat:456"); + expect((await queue.listPending())[0]?.laneKey).toBe("chat:123:topic:7"); + }); + }); + + it("blocks opted-in legacy candidate lanes using their canonical owner", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ }); + + await queue.enqueue( + "a", + { lane: "chat:123" }, + { laneKey: "chat:123:topic:7", receivedAt: 1 }, + ); + await queue.enqueue( + "b", + { lane: "chat:123" }, + { laneKey: "chat:123:topic:8", receivedAt: 2 }, + ); + await queue.enqueue( + "c", + { lane: "chat:456" }, + { laneKey: "chat:456:topic:9", receivedAt: 3 }, + ); + await queue.claim("a", { ownerId: "sibling-worker" }); + + const claimed = await queue.claimNext({ + ownerId: "worker", + candidateIds: ["a", "b", "c"], + orderBy: "id", + deriveLaneKey: (record) => record.payload.lane, + reconcileStoredLaneKey: (_record, storedLaneKey, derivedLaneKey) => + storedLaneKey.startsWith(`${derivedLaneKey}:topic:`), + }); + + expect(claimed?.id).toBe("c"); + expect(claimed?.laneKey).toBe("chat:456"); + expect((await queue.listClaims()).find((record) => record.id === "a")?.laneKey).toBe( + "chat:123:topic:7", + ); + expect((await queue.listPending())[0]?.laneKey).toBe("chat:123:topic:8"); + }); + }); + + it("preserves persisted lanes when an owner rejects their reconciliation", async () => { + await withTempState(async (stateDir) => { + let clock = 1; + const queue = createTestIngressQueue<{ lane: string }>(stateDir, { now: () => clock++ }); + + await queue.enqueue("a", { lane: "chat:123" }, { laneKey: "chat:999:topic:7" }); + + const claimed = await queue.claimNext({ + ownerId: "worker", + deriveLaneKey: (record) => record.payload.lane, + reconcileStoredLaneKey: (_record, storedLaneKey, derivedLaneKey) => + storedLaneKey === `${derivedLaneKey}:topic:7`, + }); + + expect(claimed?.laneKey).toBe("chat:999:topic:7"); + expect((await queue.listClaims())[0]?.laneKey).toBe("chat:999:topic:7"); + }); + }); + it("blocks lanes claimed by candidate rows before claiming later candidates", async () => { await withTempState(async (stateDir) => { let clock = 1; diff --git a/src/channels/message/ingress-queue.ts b/src/channels/message/ingress-queue.ts index 61e66086142e..5bd44bb40117 100644 --- a/src/channels/message/ingress-queue.ts +++ b/src/channels/message/ingress-queue.ts @@ -204,6 +204,12 @@ export type ChannelIngressQueue; deriveLaneKey?: (record: ChannelIngressQueueRecord) => string | undefined; + /** Authorize a changed durable lane before the atomic pending-to-claimed transition. */ + reconcileStoredLaneKey?: ( + record: ChannelIngressQueueRecord, + storedLaneKey: string, + derivedLaneKey: string, + ) => boolean; }): Promise | null>; claim( id: string, @@ -769,6 +775,26 @@ export function createChannelIngressQueue< if (candidateIds?.length === 0) { return null; } + const resolveClaimLaneKey = ( + record: ChannelIngressQueueRecord, + ): string | undefined => { + const storedLaneKey = record.laneKey; + if (storedLaneKey === undefined) { + return claimOptions?.deriveLaneKey?.(record); + } + if (!claimOptions?.deriveLaneKey || !claimOptions.reconcileStoredLaneKey) { + return storedLaneKey; + } + const derivedLaneKey = claimOptions.deriveLaneKey(record); + if (!derivedLaneKey || derivedLaneKey === storedLaneKey) { + return storedLaneKey; + } + // Durable identity changes need their channel owner's explicit approval; + // unrelated derivations can intentionally be ephemeral claim lanes. + return claimOptions.reconcileStoredLaneKey(record, storedLaneKey, derivedLaneKey) + ? derivedLaneKey + : storedLaneKey; + }; const database = openStateDatabase(options.stateDir); return runOpenClawStateWriteTransaction( (tx) => { @@ -788,14 +814,11 @@ export function createChannelIngressQueue< ).rows; const claimedCandidateLaneKeys = claimedCandidateRows .map((row) => { - if (row.lane_key) { + if (row.lane_key && !claimOptions?.reconcileStoredLaneKey) { return row.lane_key; } - if (!claimOptions?.deriveLaneKey) { - return undefined; - } const rec = baseRecord(row); - return rec ? claimOptions.deriveLaneKey(rec) : undefined; + return rec ? resolveClaimLaneKey(rec) : (row.lane_key ?? undefined); }) .filter((laneKey): laneKey is string => Boolean(laneKey)); if (claimedCandidateLaneKeys.length > 0) { @@ -847,9 +870,7 @@ export function createChannelIngressQueue< } continue; } - const laneKey = - row.lane_key ?? - (claimOptions?.deriveLaneKey ? claimOptions.deriveLaneKey(rec) : undefined); + const laneKey = resolveClaimLaneKey(rec); if (!laneKey || !effectiveBlocked.has(laneKey)) { selected = { row, record: rec }; break; @@ -866,9 +887,7 @@ export function createChannelIngressQueue< if (!selected) { return null; } - const derivedLaneKey = - selected.row.lane_key ?? - (claimOptions?.deriveLaneKey ? claimOptions.deriveLaneKey(selected.record) : undefined); + const derivedLaneKey = resolveClaimLaneKey(selected.record); const token = randomUUID(); const ownerId = normalizePart(claimOptions?.ownerId, `${process.pid}`); const result = executeSqliteQuerySync(