fix(telegram): prevent durable ingress lane spins (#118357)

* fix(channels): prevent ingress lane reconciliation spins

* fix(ui): restore composer lint budget

CHANGELOG.md

@@ -60,6 +60,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- **Telegram durable ingress:** preserve pre-identity control-lane ownership during replay and attempt each drain snapshot row only once per pass, preventing targeted commands from spinning the spool and blocking polling shutdown.
 - **Control UI operator session permissions:** honor Gateway-advertised operator scopes for new-thread creation, thread management, checkpoints, and sharing controls while preserving read-only navigation and legacy Gateway compatibility. Fixes #117786. Thanks @shakkernerd.
 - **Control UI archived session deletion:** send archive-gated delete requests from Sessions-page row and mixed-selection actions so write-scoped operators can remove archived threads while active-session deletion remains admin-only. Thanks @shakkernerd.
 - **Control UI command recovery:** keep delayed detached and immediate command failures scoped to their submitting session, preserving failed drafts and attachments for that pane without overwriting the active session. Fixes #116846. Thanks @shakkernerd.

extensions/telegram/src/telegram-ingress-drain.ts

@@ -62,14 +62,26 @@ function telegramSpooledLaneKey(update: unknown, botInfo?: TelegramBotInfo): str
   });
 }
 
-function inspectTelegramSpooledUpdate(update: unknown, botInfo?: TelegramBotInfo) {
+function inspectTelegramSpooledUpdate(
+  update: unknown,
+  botInfo?: TelegramBotInfo,
+  claimedLaneKey?: string,
+) {
   const updateId = resolveTelegramUpdateId(update);
   if (updateId === null) {
     throw new TelegramIngressPayloadError("Telegram spooled update is missing numeric update_id.");
   }
+  const derivedLaneKey = telegramSpooledLaneKey(update, botInfo);
+  const preservePreIdentityControlLane =
+    botInfo !== undefined &&
+    claimedLaneKey?.endsWith(":control") === true &&
+    claimedLaneKey !== derivedLaneKey &&
+    claimedLaneKey === telegramSpooledLaneKey(update);
   return {
     eventId: telegramQueueEventId(updateId),
-    laneKey: telegramSpooledLaneKey(update, botInfo),
+    // Admission can precede getMe(). Preserve only the exact control lane that
+    // the same payload derived before identity became available.
+    laneKey: preservePreIdentityControlLane ? claimedLaneKey : derivedLaneKey,
   };
 }
 
@@ -256,7 +268,12 @@ export function createTelegramIngressMonitor(params: CreateTelegramIngressMonito
     TelegramSpooledUpdatePayload
   >({
     queue: params.queue,
-    inspect: (update) => inspectTelegramSpooledUpdate(update, params.botInfo),
+    inspect: (update, context) =>
+      inspectTelegramSpooledUpdate(
+        update,
+        params.botInfo,
+        context.phase === "claim" ? context.claimedLaneKey : undefined,
+      ),
     payload: {
       version: TELEGRAM_SPOOLED_UPDATE_PAYLOAD_VERSION,
       serialize: (update, { receivedAt }) => {

src/channels/message/ingress-drain-lanes.test.ts

@@ -0,0 +1,49 @@
+import { describe, expect, it, vi } from "vitest";
+import { createChannelIngressDrain } from "./ingress-drain.js";
+import {
+  createTestIngressQueue,
+  type IngressDrainTestPayload as Payload,
+  withTempState,
+} from "./ingress-drain.test-helpers.js";
+
+describe("channel ingress drain lanes", () => {
+  it("preserves rejected stored lanes and attempts each snapshot candidate once", async () => {
+    await withTempState(async (stateDir) => {
+      const queue = createTestIngressQueue(stateDir);
+      await queue.enqueue("active", { text: "topic" }, { laneKey: "topic" });
+      let releaseActive: (() => void) | undefined;
+      const activeDone = new Promise<void>((resolve) => {
+        releaseActive = resolve;
+      });
+      const dispatches: string[] = [];
+      const drain = createChannelIngressDrain<Payload>({
+        queue,
+        deriveLaneKey: (record) => record.payload.text,
+        reconcileStoredLaneKey: () => false,
+        dispatchClaimedEvent: async (event, lifecycle) => {
+          dispatches.push(event.id);
+          if (event.id === "active") {
+            await activeDone;
+          }
+          await lifecycle.onAdopted();
+        },
+      });
+
+      expect(await drain.drainOnce()).toEqual({ started: 1 });
+      await vi.waitFor(() => expect(dispatches).toEqual(["active"]));
+      await queue.enqueue("candidate", { text: "topic" }, { laneKey: "control" });
+      const claimNext = vi.spyOn(queue, "claimNext");
+
+      await expect(drain.drainOnce()).resolves.toEqual({ started: 1 });
+      await vi.waitFor(() => expect(dispatches).toEqual(["active", "candidate"]));
+      expect(claimNext).toHaveBeenCalledTimes(2);
+      await expect(queue.enqueue("candidate", { text: "topic" })).resolves.toMatchObject({
+        kind: "completed",
+      });
+
+      releaseActive?.();
+      await drain.waitForIdle();
+      drain.dispose();
+    });
+  });
+});

src/channels/message/ingress-drain-state.ts

@@ -47,8 +47,25 @@ export function activeClaimKey<TPayload, TMetadata>(
 export function resolveLaneKey<TPayload, TMetadata>(
   record: ChannelIngressQueueRecord<TPayload, TMetadata>,
   deriveLaneKey?: (record: ChannelIngressQueueRecord<TPayload, TMetadata>) => string | undefined,
+  reconcileStoredLaneKey?: (
+    record: ChannelIngressQueueRecord<TPayload, TMetadata>,
+    storedLaneKey: string,
+    derivedLaneKey: string,
+  ) => boolean,
 ): string {
-  return deriveLaneKey?.(record) ?? record.laneKey ?? record.id;
+  const derivedLaneKey = deriveLaneKey?.(record);
+  const storedLaneKey = record.laneKey;
+  if (
+    !reconcileStoredLaneKey ||
+    storedLaneKey === undefined ||
+    derivedLaneKey === undefined ||
+    storedLaneKey === derivedLaneKey
+  ) {
+    return derivedLaneKey ?? storedLaneKey ?? record.id;
+  }
+  return reconcileStoredLaneKey(record, storedLaneKey, derivedLaneKey)
+    ? derivedLaneKey
+    : storedLaneKey;
 }
 
 export function sortedKeys(keys: Iterable<string>): string[] {

src/channels/message/ingress-drain.ts

@@ -720,12 +720,16 @@ export function createChannelIngressDrain<
             !state.superseded
           );
         })
-        .map((claim) => resolveLaneKey(claim, options.deriveLaneKey)),
+        .map((claim) =>
+          resolveLaneKey(claim, options.deriveLaneKey, options.reconcileStoredLaneKey),
+        ),
     );
     const retryDelayedLaneKeys = new Set<string>();
     for (const event of pending) {
       if (resolveIngressRetryDelayMs(event, options.retryPolicy, now()) > 0) {
-        retryDelayedLaneKeys.add(resolveLaneKey(event, options.deriveLaneKey));
+        retryDelayedLaneKeys.add(
+          resolveLaneKey(event, options.deriveLaneKey, options.reconcileStoredLaneKey),
+        );
       }
     }
 
@@ -742,13 +746,13 @@ export function createChannelIngressDrain<
       if (shouldStop()) {
         break;
       }
-      const laneKey = resolveLaneKey(event, options.deriveLaneKey);
+      const laneKey = resolveLaneKey(event, options.deriveLaneKey, options.reconcileStoredLaneKey);
       if (await supersedeActiveIfNeeded(event, laneKey)) {
         blockedLaneKeys.delete(laneKey);
       }
     }
 
-    const candidateIds = pending.map((event) => event.id);
+    const candidateIds = new Set(pending.map((event) => event.id));
     let started = 0;
     while (started < startLimit) {
       if (shouldStop()) {
@@ -768,11 +772,18 @@ export function createChannelIngressDrain<
       if (!claimed) {
         break;
       }
+      // One snapshot row gets one attempt per pass. A released claim remains
+      // pending for the next pump instead of spinning through SQLite here.
+      candidateIds.delete(claimed.id);
       if (shouldStop()) {
         await queue.release(claimed, { recordAttempt: false });
         break;
       }
-      const laneKey = resolveLaneKey(claimed, options.deriveLaneKey);
+      const laneKey = resolveLaneKey(
+        claimed,
+        options.deriveLaneKey,
+        options.reconcileStoredLaneKey,
+      );
       const existing = laneOwnerByKey.get(laneKey);
       if (existing && existing.phase !== "settled") {
         if (await supersedeActiveIfNeeded(claimed, laneKey)) {

ui/src/pages/chat/components/chat-composer-dom.ts

@@ -129,6 +129,20 @@ export function focusComposerFromChrome(event: MouseEvent, connected: boolean) {
     ?.focus({ preventScroll: true });
 }
 
+export function preserveComposerFocusOnPrimaryAction(
+  event: PointerEvent,
+  textarea: HTMLTextAreaElement | null,
+): void {
+  const composerShell = textarea?.closest<HTMLElement>(".agent-chat__composer-shell");
+  if (
+    document.activeElement === textarea &&
+    composerShell &&
+    Number.parseFloat(getComputedStyle(composerShell).marginBottom) === 0
+  ) {
+    event.preventDefault();
+  }
+}
+
 export function restoreHistoryCaret(target: HTMLTextAreaElement, direction: "up" | "down") {
   requestAnimationFrame(() => {
     if (document.activeElement !== target) {

ui/src/pages/chat/components/chat-composer.ts

@@ -13,6 +13,7 @@ import {
   adjustTextareaHeight,
   disconnectTextareaOverflowObserver,
   observeTextareaOverflow,
+  preserveComposerFocusOnPrimaryAction,
   restoreHistoryCaret,
   scheduleTextareaHeightAdjustment,
 } from "./chat-composer-dom.ts";
@@ -638,18 +639,6 @@ export function renderChatComposer(props: ChatComposerProps) {
       target.readOnly = true;
     }
   };
-  const handlePrimaryActionPointerDown = (event: PointerEvent) => {
-    const composerShell = state.composerTextarea?.closest<HTMLElement>(
-      ".agent-chat__composer-shell",
-    );
-    if (
-      document.activeElement === state.composerTextarea &&
-      composerShell &&
-      Number.parseFloat(getComputedStyle(composerShell).marginBottom) === 0
-    ) {
-      event.preventDefault();
-    }
-  };
   const runControlsProps: ChatRunControlsProps = {
     canAbort: showAbortableUi,
     canSend: canSubmitDraft(actionDraft),
@@ -678,7 +667,8 @@ export function renderChatComposer(props: ChatComposerProps) {
     microphonePicker,
     dictation,
     onDictationPointerDown: handleDictationPointerDown,
-    onPrimaryActionPointerDown: handlePrimaryActionPointerDown,
+    onPrimaryActionPointerDown: (event) =>
+      preserveComposerFocusOnPrimaryAction(event, state.composerTextarea),
   };
   const cameraFacingMode = props.realtimeTalkVideoStream
     ?.getVideoTracks?.()[0]