fix(telegram): keep timed-out lanes guarded

vortexopenclaw · vortexopenclaw · commit b14c78f55884 · 2026-07-01T13:38:40.000-07:00
diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts
@@ -2803,7 +2803,7 @@ describe("TelegramPollingSession", () => {
     }
   });
 
-  it("releases a timed-out lane when the old handler ignores reply abort", async () => {
+  it("keeps a timed-out lane guarded when the old handler ignores reply abort", async () => {
     vi.useFakeTimers({ shouldAdvanceTime: true });
     const abort = new AbortController();
     const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-spool-"));
@@ -2813,31 +2813,23 @@ describe("TelegramPollingSession", () => {
     const firstTurnDone = new Promise<void>((resolve) => {
       releaseFirstTurn = resolve;
     });
-    const firstBot = {
-      api: {
-        deleteWebhook: vi.fn(async () => true),
-        config: { use: vi.fn() },
-      },
-      init: vi.fn(async () => undefined),
-      handleUpdate: vi.fn(async (update: { update_id?: number }) => {
-        events.push(`first:${update.update_id}`);
-        await firstTurnDone;
-      }),
-      stop: vi.fn(async () => undefined),
-    };
-    const secondBot = {
+    const bot = {
       api: {
         deleteWebhook: vi.fn(async () => true),
         config: { use: vi.fn() },
       },
       init: vi.fn(async () => undefined),
       handleUpdate: vi.fn(async (update: { update_id?: number }) => {
-        events.push(`second:${update.update_id}`);
-        abort.abort();
+        events.push(`handled:${update.update_id}`);
+        if (update.update_id === 42) {
+          await firstTurnDone;
+        } else {
+          abort.abort();
+        }
       }),
       stop: vi.fn(async () => undefined),
     };
-    createTelegramBotMock.mockReturnValueOnce(firstBot).mockReturnValueOnce(secondBot);
+    createTelegramBotMock.mockReturnValue(bot);
     await writeSpooledTestUpdates(tempDir, [
       topicUpdate(42, 10, "wedged topic 10 turn"),
       topicUpdate(43, 10, "later topic 10 turn"),
@@ -2859,23 +2851,25 @@ describe("TelegramPollingSession", () => {
 
     try {
       const runPromise = session.runUntilAbort();
-      await vi.waitFor(() => expect(events).toEqual(["first:42"]));
+      await vi.waitFor(() => expect(events).toEqual(["handled:42"]));
 
       await vi.advanceTimersByTimeAsync(250);
       await vi.waitFor(() => expectLogIncludes(log, "did not stop within 100ms"));
       await vi.advanceTimersByTimeAsync(500);
-      await vi.waitFor(() => expect(worker.createWorker).toHaveBeenCalledTimes(2));
-      await vi.waitFor(() => expect(events).toEqual(["first:42", "second:43"]));
 
       expect(await failedUpdateIds(tempDir)).toEqual([42]);
-      expect(await pendingUpdateIds(tempDir, "all")).toEqual([]);
+      expect(events).toEqual(["handled:42"]);
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
+      expect(worker.createWorker).toHaveBeenCalledTimes(1);
 
+      abort.abort();
       releaseFirstTurn?.();
       await vi.advanceTimersByTimeAsync(20_000);
       await runPromise;
 
-      expect(firstBot.stop).toHaveBeenCalledTimes(1);
-      expect(secondBot.stop).toHaveBeenCalledTimes(1);
+      expect(events).toEqual(["handled:42"]);
+      expect(await pendingUpdateIds(tempDir, "all")).toEqual([43]);
+      expect(bot.stop).toHaveBeenCalledTimes(1);
     } finally {
       releaseFirstTurn?.();
       abort.abort();
diff --git a/extensions/telegram/src/polling-session.ts b/extensions/telegram/src/polling-session.ts
@@ -875,7 +875,7 @@ export class TelegramPollingSession {
     }
     const age = formatDurationPrecise(timedOutHandler.ageMs);
     activeHandler.timedOutAt = Date.now();
-    const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed, aborting active reply work, and restarting isolated ingress so later updates can drain.`;
+    const message = `Telegram isolated polling spool handler timed out behind update ${handler.updateId} on lane ${handler.laneKey} after ${age}; marking the update failed, aborting active reply work, and waiting for the handler to stop before releasing the lane.`;
     activeHandler.timeoutMessage = message;
     try {
       const failed = await failTelegramSpooledUpdateClaim({
@@ -916,13 +916,14 @@ export class TelegramPollingSession {
       !handlerStopped &&
       activeSpooledUpdateHandlersByLane.get(handler.handlerKey) === activeHandler
     ) {
+      // Reply fences can cancel OpenClaw-owned send work, but they cannot kill
+      // arbitrary handler code. Keep the lane guarded until the handler settles
+      // so same-lane updates never run concurrently.
       this.opts.log(
-        `[telegram][diag] timed out spooled update ${handler.updateId} did not stop within ${formatDurationPrecise(this.#spooledUpdateHandlerAbortGraceMs)} after reply abort; releasing lane ${handler.laneKey} and restarting isolated ingress so later updates can drain.`,
+        `[telegram][diag] timed out spooled update ${handler.updateId} did not stop within ${formatDurationPrecise(this.#spooledUpdateHandlerAbortGraceMs)} after reply abort; keeping lane ${handler.laneKey} guarded until the handler stops.`,
       );
-      activeSpooledUpdateHandlersByLane.delete(handler.handlerKey);
-      this.#spooledUpdateHandlerKeys.delete(handler.handlerKey);
       this.#status.notePollingError(message);
-      return { handlerKey: handler.handlerKey, restart: true };
+      return { handlerKey: handler.handlerKey, restart: false };
     }
     if (activeSpooledUpdateHandlersByLane.get(handler.handlerKey) === activeHandler) {
       activeSpooledUpdateHandlersByLane.delete(handler.handlerKey);
