fix: dedupe inbound messages across providers

src/auto-reply/reply/dispatch-from-config.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it, vi } from "vitest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
 
 import type { ClawdbotConfig } from "../../config/config.js";
 import type { MsgContext } from "../templating.js";
@@ -34,6 +34,7 @@ vi.mock("./abort.js", () => ({
 }));
 
 const { dispatchReplyFromConfig } = await import("./dispatch-from-config.js");
+const { resetInboundDedupe } = await import("./inbound-dedupe.js");
 
 function createDispatcher(): ReplyDispatcher {
   return {
@@ -46,6 +47,9 @@ function createDispatcher(): ReplyDispatcher {
 }
 
 describe("dispatchReplyFromConfig", () => {
+  beforeEach(() => {
+    resetInboundDedupe();
+  });
   it("does not route when Provider matches OriginatingChannel (even if Surface is missing)", async () => {
     mocks.tryFastAbortFromMessage.mockResolvedValue({
       handled: false,
@@ -125,4 +129,34 @@ describe("dispatchReplyFromConfig", () => {
       text: "⚙️ Agent was aborted.",
     });
   });
+
+  it("deduplicates inbound messages by MessageSid and origin", async () => {
+    mocks.tryFastAbortFromMessage.mockResolvedValue({
+      handled: false,
+      aborted: false,
+    });
+    const cfg = {} as ClawdbotConfig;
+    const ctx: MsgContext = {
+      Provider: "whatsapp",
+      OriginatingChannel: "whatsapp",
+      OriginatingTo: "whatsapp:+15555550123",
+      MessageSid: "msg-1",
+    };
+    const replyResolver = vi.fn(async () => ({ text: "hi" }) as ReplyPayload);
+
+    await dispatchReplyFromConfig({
+      ctx,
+      cfg,
+      dispatcher: createDispatcher(),
+      replyResolver,
+    });
+    await dispatchReplyFromConfig({
+      ctx,
+      cfg,
+      dispatcher: createDispatcher(),
+      replyResolver,
+    });
+
+    expect(replyResolver).toHaveBeenCalledTimes(1);
+  });
 });

src/auto-reply/reply/dispatch-from-config.ts

@@ -4,6 +4,7 @@ import { getReplyFromConfig } from "../reply.js";
 import type { MsgContext } from "../templating.js";
 import type { GetReplyOptions, ReplyPayload } from "../types.js";
 import { tryFastAbortFromMessage } from "./abort.js";
+import { shouldSkipDuplicateInbound } from "./inbound-dedupe.js";
 import type { ReplyDispatcher, ReplyDispatchKind } from "./reply-dispatcher.js";
 import { isRoutableChannel, routeReply } from "./route-reply.js";
 
@@ -21,6 +22,10 @@ export async function dispatchReplyFromConfig(params: {
 }): Promise<DispatchFromConfigResult> {
   const { ctx, cfg, dispatcher } = params;
 
+  if (shouldSkipDuplicateInbound(ctx)) {
+    return { queuedFinal: false, counts: dispatcher.getQueuedCounts() };
+  }
+
   // Check if we should route replies to originating channel instead of dispatcher.
   // Only route when the originating channel is DIFFERENT from the current surface.
   // This handles cross-provider routing (e.g., message from Telegram being processed

src/auto-reply/reply/inbound-dedupe.test.ts

@@ -0,0 +1,81 @@
+import { describe, expect, it } from "vitest";
+
+import type { MsgContext } from "../templating.js";
+import {
+  buildInboundDedupeKey,
+  resetInboundDedupe,
+  shouldSkipDuplicateInbound,
+} from "./inbound-dedupe.js";
+
+describe("inbound dedupe", () => {
+  it("builds a stable key when MessageSid is present", () => {
+    const ctx: MsgContext = {
+      Provider: "telegram",
+      OriginatingChannel: "telegram",
+      OriginatingTo: "telegram:123",
+      MessageSid: "42",
+    };
+    expect(buildInboundDedupeKey(ctx)).toBe("telegram|telegram:123|42");
+  });
+
+  it("skips duplicates with the same key", () => {
+    resetInboundDedupe();
+    const ctx: MsgContext = {
+      Provider: "whatsapp",
+      OriginatingChannel: "whatsapp",
+      OriginatingTo: "whatsapp:+1555",
+      MessageSid: "msg-1",
+    };
+    expect(shouldSkipDuplicateInbound(ctx, { now: 100 })).toBe(false);
+    expect(shouldSkipDuplicateInbound(ctx, { now: 200 })).toBe(true);
+  });
+
+  it("does not dedupe when the peer changes", () => {
+    resetInboundDedupe();
+    const base: MsgContext = {
+      Provider: "whatsapp",
+      OriginatingChannel: "whatsapp",
+      MessageSid: "msg-1",
+    };
+    expect(
+      shouldSkipDuplicateInbound(
+        { ...base, OriginatingTo: "whatsapp:+1000" },
+        { now: 100 },
+      ),
+    ).toBe(false);
+    expect(
+      shouldSkipDuplicateInbound(
+        { ...base, OriginatingTo: "whatsapp:+2000" },
+        { now: 200 },
+      ),
+    ).toBe(false);
+  });
+
+  it("does not dedupe across session keys", () => {
+    resetInboundDedupe();
+    const base: MsgContext = {
+      Provider: "whatsapp",
+      OriginatingChannel: "whatsapp",
+      OriginatingTo: "whatsapp:+1555",
+      MessageSid: "msg-1",
+    };
+    expect(
+      shouldSkipDuplicateInbound(
+        { ...base, SessionKey: "agent:alpha:main" },
+        { now: 100 },
+      ),
+    ).toBe(false);
+    expect(
+      shouldSkipDuplicateInbound(
+        { ...base, SessionKey: "agent:bravo:main" },
+        { now: 200 },
+      ),
+    ).toBe(false);
+    expect(
+      shouldSkipDuplicateInbound(
+        { ...base, SessionKey: "agent:alpha:main" },
+        { now: 300 },
+      ),
+    ).toBe(true);
+  });
+});

src/auto-reply/reply/inbound-dedupe.ts

@@ -0,0 +1,52 @@
+import { logVerbose, shouldLogVerbose } from "../../globals.js";
+import { createDedupeCache, type DedupeCache } from "../../infra/dedupe.js";
+import type { MsgContext } from "../templating.js";
+
+const DEFAULT_INBOUND_DEDUPE_TTL_MS = 20 * 60_000;
+const DEFAULT_INBOUND_DEDUPE_MAX = 5000;
+
+const inboundDedupeCache = createDedupeCache({
+  ttlMs: DEFAULT_INBOUND_DEDUPE_TTL_MS,
+  maxSize: DEFAULT_INBOUND_DEDUPE_MAX,
+});
+
+const normalizeProvider = (value?: string | null) =>
+  value?.trim().toLowerCase() || "";
+
+const resolveInboundPeerId = (ctx: MsgContext) =>
+  ctx.OriginatingTo ?? ctx.To ?? ctx.From ?? ctx.SessionKey;
+
+export function buildInboundDedupeKey(ctx: MsgContext): string | null {
+  const provider = normalizeProvider(
+    ctx.OriginatingChannel ?? ctx.Provider ?? ctx.Surface,
+  );
+  const messageId = ctx.MessageSid?.trim();
+  if (!provider || !messageId) return null;
+  const peerId = resolveInboundPeerId(ctx);
+  if (!peerId) return null;
+  const sessionKey = ctx.SessionKey?.trim() ?? "";
+  const accountId = ctx.AccountId?.trim() ?? "";
+  const threadId =
+    typeof ctx.MessageThreadId === "number" ? String(ctx.MessageThreadId) : "";
+  return [provider, accountId, sessionKey, peerId, threadId, messageId]
+    .filter(Boolean)
+    .join("|");
+}
+
+export function shouldSkipDuplicateInbound(
+  ctx: MsgContext,
+  opts?: { cache?: DedupeCache; now?: number },
+): boolean {
+  const key = buildInboundDedupeKey(ctx);
+  if (!key) return false;
+  const cache = opts?.cache ?? inboundDedupeCache;
+  const skipped = cache.check(key, opts?.now);
+  if (skipped && shouldLogVerbose()) {
+    logVerbose(`inbound dedupe: skipped ${key}`);
+  }
+  return skipped;
+}
+
+export function resetInboundDedupe(): void {
+  inboundDedupeCache.clear();
+}