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 <steipete@macos.shared>

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<ConstructorParameters<typeof TelegramPollingSession>[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<ChannelAccountSnapshot, "accountId">) => void;
   isolatedIngress?: ConstructorParameters<typeof TelegramPollingSession>[0]["isolatedIngress"];
+  botInfo?: ConstructorParameters<typeof TelegramPollingSession>[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<ConstructorParameters<typeof TelegramPollingSession>[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<void>((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();

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) {

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<Chat, "id"> & Partial<Pick<Chat, "type" | "is_forum">>): 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<typeof getTelegramSequentialKey>[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: {

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}`;
   }

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<TelegramSpooledUpdatePayload>;
+  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 } : {}),

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<void>((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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        from: { ...callback.from, is_bot: true },
+      }),
+    },
+    {
+      name: "an invalid callback sender",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        from: { ...callback.from, id: 0 },
+      }),
+    },
+    {
+      name: "a foreign bot-authored message",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: {
+          ...callback.message,
+          from: { ...callback.message.from, id: 999 },
+        },
+      }),
+    },
+    {
+      name: "an inaccessible callback message",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: { ...callback.message, date: 0 },
+      }),
+    },
+    {
+      name: "an independent business chat",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: { ...callback.message, business_connection_id: "business-1234" },
+      }),
+    },
+    {
+      name: "an independent guest chat",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: { ...callback.message, guest_query_id: "guest-1234" },
+      }),
+    },
+    {
+      name: "a message sent by another chat",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: { ...callback.message, sender_chat: { id: -1234, type: "channel" } },
+      }),
+    },
+    {
+      name: "a direct-messages topic from another surface",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        message: { ...callback.message, direct_messages_topic: { topic_id: 42 } },
+      }),
+    },
+    {
+      name: "an inline callback message",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        inline_message_id: "inline-message-141",
+      }),
+    },
+    {
+      name: "an oversized callback payload",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        data: "x".repeat(65),
+      }),
+    },
+    {
+      name: "a missing chat instance",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        chat_instance: "",
+      }),
+    },
+    {
+      name: "a reserved question callback",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        data: "tgq1:ask_0123456789abcdef0123456789abcdef:1",
+      }),
+    },
+    {
+      name: "a reserved approval callback",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        data: "/approve exec:def456 deny",
+      }),
+    },
+    {
+      name: "a malformed signed approval decision",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        data: "tgq1:ask_0123456789abcdef0123456789abcdef:9",
+      }),
+    },
+    {
+      name: "a reserved question callback without a topic",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...callback,
+        data: "tgq1:ask_0123456789abcdef0123456789abcdef:1",
+        message: { ...callback.message, message_thread_id: undefined },
+      }),
+    },
+    {
+      name: "an ordinary callback without a topic",
+      mutate: (callback: ReturnType<typeof createTelegramPrivateTopicCallback>) => ({
+        ...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<void>((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) => {

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<typeof getTelegramSequentialKey>[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.

src/channels/message/ingress-drain.ts

@@ -101,6 +101,11 @@ export type CreateChannelIngressDrainOptions<
     pendingEvent: ChannelIngressQueueClaim<TPayload, TMetadata>,
   ) => boolean | Promise<boolean>;
   deriveLaneKey?: (record: ChannelIngressQueueRecord<TPayload, TMetadata>) => string | undefined;
+  reconcileStoredLaneKey?: (
+    record: ChannelIngressQueueRecord<TPayload, TMetadata>,
+    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;

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;

src/channels/message/ingress-queue.ts

@@ -204,6 +204,12 @@ export type ChannelIngressQueue<TPayload, TMetadata = unknown, TCompletedMetadat
     scanLimit?: number;
     candidateIds?: Iterable<string>;
     deriveLaneKey?: (record: ChannelIngressQueueRecord<TPayload, TMetadata>) => string | undefined;
+    /** Authorize a changed durable lane before the atomic pending-to-claimed transition. */
+    reconcileStoredLaneKey?: (
+      record: ChannelIngressQueueRecord<TPayload, TMetadata>,
+      storedLaneKey: string,
+      derivedLaneKey: string,
+    ) => boolean;
   }): Promise<ChannelIngressQueueClaim<TPayload, TMetadata> | null>;
   claim(
     id: string,
@@ -769,6 +775,26 @@ export function createChannelIngressQueue<
     if (candidateIds?.length === 0) {
       return null;
     }
+    const resolveClaimLaneKey = (
+      record: ChannelIngressQueueRecord<TPayload, TMetadata>,
+    ): 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<TPayload, TMetadata>(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(