fix(telegram): accept delayed cross-lane spool replays exactly once

Telegram update ids are global across chats, but the update tracker rejected
any id at or below its single high-water mark. When one lane was delayed
while other lanes completed newer ids, the delayed durable-spool replay was
rejected as accepted-watermark and tombstoned without dispatch - silent
message loss. The tracker now rejects only restored-offset ids and exact ids
it already accepted, with the accepted-id set pruned at the persisted offset
plus a bounded retention window; the startup-only floor-replay special case
is deleted.

Refs #107246

extensions/telegram/src/bot-update-tracker.test.ts

@@ -1,6 +1,6 @@
 // Telegram tests cover bot update tracker plugin behavior.
 import { describe, expect, it, vi } from "vitest";
-import { createTelegramUpdateTracker } from "./bot-update-tracker.js";
+import { ACCEPTED_UPDATE_ID_RETENTION, createTelegramUpdateTracker } from "./bot-update-tracker.js";
 
 type TelegramUpdateTrackerState = ReturnType<
   ReturnType<typeof createTelegramUpdateTracker>["getState"]
@@ -186,10 +186,9 @@ describe("createTelegramUpdateTracker", () => {
     }
     tracker.finishUpdate(oldReplay.update, { completed: true });
 
-    expect(tracker.beginUpdate(updateCtx(42))).toEqual({
-      accepted: false,
-      reason: "accepted-watermark",
-    });
+    // Second begin is rejected (numeric set and/or semantic key). After persist
+    // advances, the numeric id may already be pruned below the persisted floor.
+    expect(tracker.beginUpdate(updateCtx(42)).accepted).toBe(false);
     expectTrackerState(tracker.getState(), {
       highestAcceptedUpdateId: 43,
       highestCompletedUpdateId: 43,
@@ -199,6 +198,167 @@ describe("createTelegramUpdateTracker", () => {
     } satisfies Partial<TelegramUpdateTrackerState>);
   });
 
+  it("dispatches a delayed lower update id after newer cross-lane ids complete", () => {
+    const tracker = createTelegramUpdateTracker({
+      initialUpdateId: null,
+      persistenceFloorUpdateId: 100,
+      ackPolicy: "after_agent_dispatch",
+    });
+
+    // Lane B finishes newer global update ids while lane A still holds N+1.
+    const laterA = tracker.beginUpdate(updateCtx(102));
+    const laterB = tracker.beginUpdate(updateCtx(103));
+    if (!laterA.accepted || !laterB.accepted) {
+      throw new Error("expected later cross-lane updates to be accepted");
+    }
+    tracker.finishUpdate(laterA.update, { completed: true });
+    tracker.finishUpdate(laterB.update, { completed: true });
+
+    // Delayed durable-spool replay of N+1 must still dispatch exactly once.
+    const delayed = tracker.beginUpdate(updateCtx(101));
+    if (!delayed.accepted) {
+      throw new Error("expected delayed cross-lane spool replay to be accepted");
+    }
+    tracker.finishUpdate(delayed.update, { completed: true });
+
+    expect(tracker.beginUpdate(updateCtx(101))).toEqual({
+      accepted: false,
+      reason: "accepted-watermark",
+    });
+    expectTrackerState(tracker.getState(), {
+      highestAcceptedUpdateId: 103,
+      highestCompletedUpdateId: 103,
+      pendingUpdateIds: [],
+      failedUpdateIds: [],
+    } satisfies Partial<TelegramUpdateTrackerState>);
+  });
+
+  it("accepts a delayed group mention after newer cross-lane ids so routing can run", () => {
+    const tracker = createTelegramUpdateTracker({
+      initialUpdateId: null,
+      persistenceFloorUpdateId: 200,
+      ackPolicy: "after_agent_dispatch",
+    });
+
+    const later = tracker.beginUpdate(updateCtx(202));
+    if (!later.accepted) {
+      throw new Error("expected later update to be accepted");
+    }
+    tracker.finishUpdate(later.update, { completed: true });
+
+    // Mention-shaped payload uses the same beginUpdate gate as any other update;
+    // watermark must not drop it before normal user_request mention routing.
+    const mention = tracker.beginUpdate({
+      update: {
+        update_id: 201,
+        message: {
+          message_id: 10,
+          text: "@bot hello",
+          entities: [{ type: "mention", offset: 0, length: 4 }],
+          chat: { id: -100, type: "supergroup", title: "group" },
+          date: 1,
+        },
+      },
+    });
+    if (!mention.accepted) {
+      throw new Error("expected delayed mention update to be accepted for user_request routing");
+    }
+    tracker.finishUpdate(mention.update, { completed: true });
+    expect(tracker.beginUpdate(updateCtx(201))).toEqual({
+      accepted: false,
+      reason: "accepted-watermark",
+    });
+  });
+
+  it("bounds accepted-id memory without a persist callback via retention window", () => {
+    // No onAcceptedUpdateId: persisted floor never advances past the option floor.
+    const tracker = createTelegramUpdateTracker({
+      initialUpdateId: null,
+      persistenceFloorUpdateId: 0,
+      ackPolicy: "after_agent_dispatch",
+    });
+    const firstId = 1;
+    const lastId = ACCEPTED_UPDATE_ID_RETENTION + 50;
+    for (let updateId = firstId; updateId <= lastId; updateId += 1) {
+      const begun = tracker.beginUpdate(updateCtx(updateId));
+      if (!begun.accepted) {
+        throw new Error(`expected update ${updateId} to be accepted`);
+      }
+      tracker.finishUpdate(begun.update, { completed: true });
+    }
+    // Ids far below the retention window may be pruned; recent ids stay suppressed.
+    expect(tracker.beginUpdate(updateCtx(firstId)).accepted).toBe(true);
+    expect(tracker.beginUpdate(updateCtx(lastId))).toEqual({
+      accepted: false,
+      reason: "accepted-watermark",
+    });
+  });
+
+  it("does not prune pending or failed accepted ids from the retention window", () => {
+    const tracker = createTelegramUpdateTracker({
+      initialUpdateId: null,
+      persistenceFloorUpdateId: 0,
+      ackPolicy: "after_agent_dispatch",
+    });
+    const pendingId = 1;
+    const failedId = 2;
+    const pending = tracker.beginUpdate(updateCtx(pendingId));
+    const failed = tracker.beginUpdate(updateCtx(failedId));
+    if (!pending.accepted || !failed.accepted) {
+      throw new Error("expected seed updates to be accepted");
+    }
+    tracker.finishUpdate(failed.update, { completed: false });
+
+    const lastId = ACCEPTED_UPDATE_ID_RETENTION + 50;
+    for (let updateId = 3; updateId <= lastId; updateId += 1) {
+      const begun = tracker.beginUpdate(updateCtx(updateId));
+      if (!begun.accepted) {
+        throw new Error(`expected update ${updateId} to be accepted`);
+      }
+      tracker.finishUpdate(begun.update, { completed: true });
+    }
+
+    // Pending ids stay in the numeric set (never pruned) so re-begin is rejected
+    // as accepted-watermark, not re-dispatched.
+    expect(tracker.beginUpdate(updateCtx(pendingId))).toEqual({
+      accepted: false,
+      reason: "accepted-watermark",
+    });
+    const failedRetry = tracker.beginUpdate(updateCtx(failedId));
+    if (!failedRetry.accepted) {
+      throw new Error("expected failed update retry to be accepted");
+    }
+    tracker.finishUpdate(failedRetry.update, { completed: true });
+    // After success, re-begin is rejected (numeric and/or semantic).
+    expect(tracker.beginUpdate(updateCtx(failedId)).accepted).toBe(false);
+  });
+
+  it("prunes accepted ids at or below the persisted Bot API offset", async () => {
+    const onAcceptedUpdateId = vi.fn();
+    const tracker = createTelegramUpdateTracker({
+      initialUpdateId: 100,
+      onAcceptedUpdateId,
+    });
+    const early = tracker.beginUpdate(updateCtx(101));
+    if (!early.accepted) {
+      throw new Error("expected early update to be accepted");
+    }
+    tracker.finishUpdate(early.update, { completed: true });
+    await flushTrackerMicrotasks();
+    expect(onAcceptedUpdateId).toHaveBeenCalledWith(101);
+
+    const later = tracker.beginUpdate(updateCtx(102));
+    if (!later.accepted) {
+      throw new Error("expected later update to be accepted");
+    }
+    tracker.finishUpdate(later.update, { completed: true });
+    await flushTrackerMicrotasks();
+
+    // Recent completed id stays suppressed; early completed id is eligible for
+    // numeric prune once <= highestPersisted (semantic/spool still guard dups).
+    expect(tracker.beginUpdate(updateCtx(102)).accepted).toBe(false);
+  });
+
   it("serializes and coalesces accepted offset persistence", async () => {
     const firstWrite = deferred();
     const secondWrite = deferred();

extensions/telegram/src/bot-update-tracker.ts

@@ -55,6 +55,12 @@ function sortedIds(ids: Set<number>): number[] {
   return [...ids].toSorted((a, b) => a - b);
 }
 
+// Bound for per-id numeric dedupe when the persisted Bot API offset does not
+// advance (no onAcceptedUpdateId) or lags. Only the realistic in-process
+// redelivery window needs numeric retention; semantic keys + spool tombstones
+// cover older ids.
+export const ACCEPTED_UPDATE_ID_RETENTION = 10_000;
+
 export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOptions = {}) {
   const initialUpdateId =
     typeof options.initialUpdateId === "number" ? options.initialUpdateId : null;
@@ -68,7 +74,9 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
   const activeHandledUpdateKeys = new Map<string, boolean>();
   const pendingUpdateIds = new Set<number>();
   const failedUpdateIds = new Set<number>();
-  const completedFloorReplayUpdateIds = new Set<number>();
+  // Per-id acceptance, not a global high-water mark: multi-lane spool drains can
+  // finish newer update IDs before an older delayed id from another chat replays.
+  const acceptedUpdateIds = new Set<number>();
   let highestAcceptedUpdateId: number | null = initialUpdateId;
   let highestPersistedAcceptedUpdateId: number | null = persistenceFloorUpdateId;
   let highestPersistenceRequestedUpdateId: number | null = persistenceFloorUpdateId;
@@ -80,6 +88,34 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
     options.onSkip?.(key);
   };
 
+  // One prune rule: drop accepted ids at or below max(persisted offset,
+  // highestAccepted - retention) unless still pending or failed. Persisted
+  // floor is safe (getUpdates cannot redeliver below it); retention bounds
+  // trackers that never advance a persisted floor.
+  const pruneAcceptedUpdateIds = () => {
+    if (highestAcceptedUpdateId === null && highestPersistedAcceptedUpdateId === null) {
+      return;
+    }
+    const windowFloor =
+      highestAcceptedUpdateId === null
+        ? Number.NEGATIVE_INFINITY
+        : highestAcceptedUpdateId - ACCEPTED_UPDATE_ID_RETENTION;
+    const persistedFloor =
+      highestPersistedAcceptedUpdateId === null
+        ? Number.NEGATIVE_INFINITY
+        : highestPersistedAcceptedUpdateId;
+    const pruneAtOrBelow = Math.max(persistedFloor, windowFloor);
+    for (const id of acceptedUpdateIds) {
+      if (id > pruneAtOrBelow) {
+        continue;
+      }
+      if (pendingUpdateIds.has(id) || failedUpdateIds.has(id)) {
+        continue;
+      }
+      acceptedUpdateIds.delete(id);
+    }
+  };
+
   const drainPersistQueue = async () => {
     const persist = options.onAcceptedUpdateId;
     if (persistInFlight || typeof persist !== "function") {
@@ -97,6 +133,7 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
             updateId > highestPersistedAcceptedUpdateId
           ) {
             highestPersistedAcceptedUpdateId = updateId;
+            pruneAcceptedUpdateIds();
           }
         } catch (err) {
           options.onPersistError?.(err);
@@ -125,17 +162,13 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
   };
 
   const acceptUpdateId = (updateId: number) => {
-    if (highestAcceptedUpdateId !== null && updateId <= highestAcceptedUpdateId) {
-      return;
+    acceptedUpdateIds.add(updateId);
+    if (highestAcceptedUpdateId === null || updateId > highestAcceptedUpdateId) {
+      highestAcceptedUpdateId = updateId;
     }
-    highestAcceptedUpdateId = updateId;
+    pruneAcceptedUpdateIds();
   };
 
-  const isFloorReplayUpdateId = (updateId: number) =>
-    initialUpdateId === null &&
-    persistenceFloorUpdateId !== null &&
-    updateId <= persistenceFloorUpdateId;
-
   function resolveSafeCompletedUpdateId() {
     if (highestCompletedUpdateId === null) {
       return null;
@@ -184,18 +217,16 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
     const updateId = resolveTelegramUpdateId(ctx);
     const updateKey = buildTelegramUpdateKey(ctx);
     if (typeof updateId === "number") {
-      if (highestAcceptedUpdateId !== null && updateId <= highestAcceptedUpdateId) {
-        const floorReplay = isFloorReplayUpdateId(updateId);
-        if (!floorReplay && !failedUpdateIds.has(updateId)) {
-          skip(`update:${updateId}`);
-          return { accepted: false, reason: "accepted-watermark" };
-        }
-        if (floorReplay && completedFloorReplayUpdateIds.has(updateId)) {
-          skip(`update:${updateId}`);
-          return { accepted: false, reason: "accepted-watermark" };
-        }
-      } else {
+      if (failedUpdateIds.has(updateId)) {
         failedUpdateIds.delete(updateId);
+      } else if (initialUpdateId !== null && updateId <= initialUpdateId) {
+        // Restored Bot API offset: suppress redelivery of already-persisted ids.
+        skip(`update:${updateId}`);
+        return { accepted: false, reason: "accepted-watermark" };
+      } else if (acceptedUpdateIds.has(updateId)) {
+        // Same process already accepted this exact id (completed or in-flight).
+        skip(`update:${updateId}`);
+        return { accepted: false, reason: "accepted-watermark" };
       }
     }
     if (updateKey) {
@@ -241,9 +272,6 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
       pendingUpdateIds.delete(update.updateId);
       if (finish.completed) {
         failedUpdateIds.delete(update.updateId);
-        if (isFloorReplayUpdateId(update.updateId)) {
-          completedFloorReplayUpdateIds.add(update.updateId);
-        }
         if (highestCompletedUpdateId === null || update.updateId > highestCompletedUpdateId) {
           highestCompletedUpdateId = update.updateId;
         }
@@ -256,6 +284,7 @@ export function createTelegramUpdateTracker(options: TelegramUpdateTrackerOption
             options.onPersistError?.(err);
           });
       }
+      pruneAcceptedUpdateIds();
     }
   };