fix(queue): dedupe queued redelivery message IDs across drain restarts

rylena · rylena · commit b86b4d99c590 · 2026-03-03T15:21:11.000Z
diff --git a/src/auto-reply/reply/queue.ts b/src/auto-reply/reply/queue.ts
@@ -2,7 +2,11 @@ export { extractQueueDirective } from "./queue/directive.js";
 export { clearSessionQueues } from "./queue/cleanup.js";
 export type { ClearSessionQueueResult } from "./queue/cleanup.js";
 export { scheduleFollowupDrain } from "./queue/drain.js";
-export { enqueueFollowupRun, getFollowupQueueDepth } from "./queue/enqueue.js";
+export {
+  enqueueFollowupRun,
+  getFollowupQueueDepth,
+  resetRecentQueuedMessageIdDedupe,
+} from "./queue/enqueue.js";
 export { resolveQueueSettings } from "./queue/settings.js";
 export { clearFollowupQueue } from "./queue/state.js";
 export type {
diff --git a/src/auto-reply/reply/queue/enqueue.ts b/src/auto-reply/reply/queue/enqueue.ts
@@ -1,8 +1,28 @@
+import { createDedupeCache } from "../../../infra/dedupe.js";
 import { applyQueueDropPolicy, shouldSkipQueueItem } from "../../../utils/queue-helpers.js";
 import { kickFollowupDrainIfIdle } from "./drain.js";
 import { getExistingFollowupQueue, getFollowupQueue } from "./state.js";
 import type { FollowupRun, QueueDedupeMode, QueueSettings } from "./types.js";
 
+const RECENT_QUEUE_MESSAGE_IDS = createDedupeCache({
+  ttlMs: 5 * 60 * 1000,
+  maxSize: 10_000,
+});
+
+function buildRecentMessageIdKey(run: FollowupRun, queueKey: string): string | undefined {
+  const messageId = run.messageId?.trim();
+  if (!messageId) {
+    return undefined;
+  }
+  const route = [
+    run.originatingChannel ?? "",
+    run.originatingTo ?? "",
+    run.originatingAccountId ?? "",
+    run.originatingThreadId == null ? "" : String(run.originatingThreadId),
+  ].join("|");
+  return `${queueKey}|${route}|${messageId}`;
+}
+
 function isRunAlreadyQueued(
   run: FollowupRun,
   items: FollowupRun[],
@@ -31,6 +51,11 @@ export function enqueueFollowupRun(
   dedupeMode: QueueDedupeMode = "message-id",
 ): boolean {
   const queue = getFollowupQueue(key, settings);
+  const recentMessageIdKey = dedupeMode !== "none" ? buildRecentMessageIdKey(run, key) : undefined;
+  if (recentMessageIdKey && RECENT_QUEUE_MESSAGE_IDS.peek(recentMessageIdKey)) {
+    return false;
+  }
+
   const dedupe =
     dedupeMode === "none"
       ? undefined
@@ -54,6 +79,9 @@ export function enqueueFollowupRun(
   }
 
   queue.items.push(run);
+  if (recentMessageIdKey) {
+    RECENT_QUEUE_MESSAGE_IDS.check(recentMessageIdKey);
+  }
   // If drain finished and deleted the queue before this item arrived, a new queue
   // object was created (draining: false) but nobody scheduled a drain for it.
   // Use the cached callback to restart the drain now.
@@ -70,3 +98,7 @@ export function getFollowupQueueDepth(key: string): number {
   }
   return queue.items.length;
 }
+
+export function resetRecentQueuedMessageIdDedupe(): void {
+  RECENT_QUEUE_MESSAGE_IDS.clear();
+}
diff --git a/src/auto-reply/reply/reply-flow.test.ts b/src/auto-reply/reply/reply-flow.test.ts
@@ -1,4 +1,4 @@
-import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { expectInboundContextContract } from "../../../test/helpers/inbound-contract.js";
 import type { OpenClawConfig } from "../../config/config.js";
 import { defaultRuntime } from "../../runtime.js";
@@ -8,7 +8,11 @@ import { finalizeInboundContext } from "./inbound-context.js";
 import { normalizeInboundTextNewlines } from "./inbound-text.js";
 import { parseLineDirectives, hasLineDirectives } from "./line-directives.js";
 import type { FollowupRun, QueueSettings } from "./queue.js";
-import { enqueueFollowupRun, scheduleFollowupDrain } from "./queue.js";
+import {
+  enqueueFollowupRun,
+  resetRecentQueuedMessageIdDedupe,
+  scheduleFollowupDrain,
+} from "./queue.js";
 import { createReplyDispatcher } from "./reply-dispatcher.js";
 import { createReplyToModeFilter, resolveReplyToMode } from "./reply-threading.js";
 
@@ -627,6 +631,10 @@ function createRun(params: {
 }
 
 describe("followup queue deduplication", () => {
+  beforeEach(() => {
+    resetRecentQueuedMessageIdDedupe();
+  });
+
   it("deduplicates messages with same Discord message_id", async () => {
     const key = `test-dedup-message-id-${Date.now()}`;
     const calls: FollowupRun[] = [];
@@ -690,6 +698,51 @@ describe("followup queue deduplication", () => {
     expect(calls[0]?.prompt).toContain("[Queued messages while agent was busy]");
   });
 
+  it("deduplicates same message_id after queue drain restarts", async () => {
+    const key = `test-dedup-after-drain-${Date.now()}`;
+    const calls: FollowupRun[] = [];
+    const done = createDeferred<void>();
+    const runFollowup = async (run: FollowupRun) => {
+      calls.push(run);
+      done.resolve();
+    };
+    const settings: QueueSettings = {
+      mode: "collect",
+      debounceMs: 0,
+      cap: 50,
+      dropPolicy: "summarize",
+    };
+
+    const first = enqueueFollowupRun(
+      key,
+      createRun({
+        prompt: "first",
+        messageId: "same-id",
+        originatingChannel: "signal",
+        originatingTo: "+10000000000",
+      }),
+      settings,
+    );
+    expect(first).toBe(true);
+
+    scheduleFollowupDrain(key, runFollowup);
+    await done.promise;
+
+    const redelivery = enqueueFollowupRun(
+      key,
+      createRun({
+        prompt: "first-redelivery",
+        messageId: "same-id",
+        originatingChannel: "signal",
+        originatingTo: "+10000000000",
+      }),
+      settings,
+    );
+
+    expect(redelivery).toBe(false);
+    expect(calls).toHaveLength(1);
+  });
+
   it("deduplicates exact prompt when routing matches and no message id", async () => {
     const key = `test-dedup-whatsapp-${Date.now()}`;
     const settings: QueueSettings = {
