refactor(feishu): adopt core claimable dedupe for inbound events (#104384)

docs/refactor/database-first.md

@@ -1129,9 +1129,10 @@ sessionId})`; create, branch, continue, list, and fork flows live in their
   now use shared SQLite plugin state. The old `imessage/catchup/*.json`,
   `imessage/reply-cache.jsonl`, and `imessage/sent-echoes.jsonl` files are
   doctor inputs only.
-- Feishu message dedupe rows now use shared SQLite plugin state instead of
-  `feishu/dedup/*.json` files. Its legacy JSON import plan lives in the Feishu
-  plugin setup/doctor migration surface, not in core migration code.
+- Feishu message dedupe rows now ride the core claimable dedupe
+  (`feishu.dedup.*` namespaces in shared SQLite plugin state) instead of
+  `feishu/dedup/*.json` files or the retired hand-rolled `dedup.*` store, with
+  no legacy import because replay-protection cache rebuilds after upgrade.
 - Microsoft Teams conversations, polls, pending upload buffers, and feedback
   learnings now use shared SQLite plugin state/blob tables. The pending upload
   path uses `plugin_blob_entries` so media buffers are stored as SQLite BLOBs

extensions/feishu/legacy-state-migrations-api.ts

@@ -1,2 +0,0 @@
-// Feishu API module exposes the plugin public contract.
-export { detectFeishuLegacyStateMigrations } from "./src/dedup-migrations.js";

extensions/feishu/package.json

@@ -29,9 +29,6 @@
       "./index.ts"
     ],
     "setupEntry": "./setup-entry.ts",
-    "setupFeatures": {
-      "legacyStateMigrations": true
-    },
     "channel": {
       "id": "feishu",
       "label": "Feishu",

extensions/feishu/setup-entry.test.ts

@@ -15,9 +15,9 @@ describe("feishu setup entry", () => {
     const { default: setupEntry } = await import("./setup-entry.js");
 
     expect(setupEntry.kind).toBe("bundled-channel-setup-entry");
-    expect(setupEntry.features).toEqual({ legacyStateMigrations: true });
+    expect(setupEntry.features).toBeUndefined();
     expect(typeof setupEntry.loadSetupPlugin).toBe("function");
-    expect(setupEntry.loadLegacyStateMigrationDetector?.()).toBeTypeOf("function");
+    expect(setupEntry.loadLegacyStateMigrationDetector).toBeUndefined();
     expect(typeof setupEntry.setChannelRuntime).toBe("function");
   });
 

extensions/feishu/setup-entry.ts

@@ -3,17 +3,10 @@ import { defineBundledChannelSetupEntry } from "openclaw/plugin-sdk/channel-entr
 
 export default defineBundledChannelSetupEntry({
   importMetaUrl: import.meta.url,
-  features: {
-    legacyStateMigrations: true,
-  },
   plugin: {
     specifier: "./setup-api.js",
     exportName: "feishuPlugin",
   },
-  legacyStateMigrations: {
-    specifier: "./legacy-state-migrations-api.js",
-    exportName: "detectFeishuLegacyStateMigrations",
-  },
   secrets: {
     specifier: "./secret-contract-api.js",
     exportName: "channelSecrets",

extensions/feishu/src/bot.ts

@@ -51,7 +51,7 @@ import { type FeishuPermissionError, resolveFeishuSenderName } from "./bot-sende
 import { getChatInfo } from "./chat.js";
 import { createFeishuClient } from "./client.js";
 import { resolveConfiguredFeishuGroupSessionScope } from "./conversation-id.js";
-import { finalizeFeishuMessageProcessing, tryRecordMessagePersistent } from "./dedup.js";
+import { finalizeFeishuMessageProcessing, recordProcessedFeishuMessage } from "./dedup.js";
 import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js";
 import { maybeCreateDynamicAgent } from "./dynamic-agent.js";
 import { extractMentionTargets, isMentionForwardRequest } from "./mention.js";
@@ -1534,7 +1534,7 @@ export async function handleFeishuMessage(params: {
       // Uses a shared "broadcast" namespace (not per-account) so the first handler
       // to reach this point claims the message; subsequent accounts skip.
       if (
-        !(await tryRecordMessagePersistent(messageDedupeKey ?? ctx.messageId, "broadcast", log))
+        !(await recordProcessedFeishuMessage(messageDedupeKey ?? ctx.messageId, "broadcast", log))
       ) {
         log(
           `feishu[${account.accountId}]: broadcast already claimed by another account for message ${ctx.messageId}; skipping`,

extensions/feishu/src/dedup-migrations.test.ts

@@ -1,90 +0,0 @@
-// Feishu tests cover dedup migrations plugin behavior.
-import fs from "node:fs/promises";
-import os from "node:os";
-import path from "node:path";
-import { afterEach, describe, expect, it, vi } from "vitest";
-import { detectFeishuLegacyStateMigrations } from "./dedup-migrations.js";
-
-const tempDirs: string[] = [];
-
-async function makeStateDir(): Promise<string> {
-  const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-dedup-migration-"));
-  tempDirs.push(dir);
-  return dir;
-}
-
-afterEach(async () => {
-  vi.useRealTimers();
-  await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })));
-});
-
-describe("Feishu dedupe migration", () => {
-  it("plans recent legacy dedupe rows with remaining TTL", async () => {
-    vi.useFakeTimers();
-    vi.setSystemTime(2_000);
-    const stateDir = await makeStateDir();
-    const sourcePath = path.join(stateDir, "feishu", "dedup", "account-a.json");
-    await fs.mkdir(path.dirname(sourcePath), { recursive: true });
-    await fs.writeFile(
-      sourcePath,
-      JSON.stringify({
-        fresh: 1_000,
-        expired: 2_000 - 24 * 60 * 60 * 1000,
-        malformed: "nope",
-      }),
-    );
-
-    const plans = await Promise.resolve(
-      detectFeishuLegacyStateMigrations({
-        cfg: {},
-        env: {},
-        oauthDir: path.join(stateDir, "credentials"),
-        stateDir,
-      }),
-    );
-
-    if (!plans) {
-      throw new Error("expected migration plans");
-    }
-    expect(plans).toHaveLength(1);
-    const plan = plans[0];
-    expect(plan?.kind).toBe("plugin-state-import");
-    if (plan?.kind !== "plugin-state-import") {
-      throw new Error("expected plugin-state import plan");
-    }
-    expect(plan.pluginId).toBe("feishu");
-    expect(plan.namespace).toBe("dedup.account-a");
-    const entries = await plan.readEntries();
-    expect(entries).toHaveLength(1);
-    expect(entries[0]?.key).toMatch(/^[0-9a-f]{32}$/u);
-    expect(entries[0]?.value).toEqual({
-      namespace: "account-a",
-      messageId: "fresh",
-      seenAt: 1_000,
-    });
-    expect(entries[0]?.ttlMs).toBe(24 * 60 * 60 * 1000 - 1_000);
-  });
-
-  it("skips expired-only legacy dedupe files", async () => {
-    vi.useFakeTimers();
-    vi.setSystemTime(2_000);
-    const stateDir = await makeStateDir();
-    const sourcePath = path.join(stateDir, "feishu", "dedup", "account-a.json");
-    await fs.mkdir(path.dirname(sourcePath), { recursive: true });
-    await fs.writeFile(
-      sourcePath,
-      JSON.stringify({
-        expired: 2_000 - 24 * 60 * 60 * 1000,
-      }),
-    );
-
-    expect(
-      detectFeishuLegacyStateMigrations({
-        cfg: {},
-        env: {},
-        oauthDir: path.join(stateDir, "credentials"),
-        stateDir,
-      }),
-    ).toStrictEqual([]);
-  });
-});

extensions/feishu/src/dedup-migrations.ts

@@ -1,103 +0,0 @@
-// Feishu plugin module implements dedup migrations behavior.
-import { createHash } from "node:crypto";
-import fs from "node:fs";
-import path from "node:path";
-import type { BundledChannelLegacyStateMigrationDetector } from "openclaw/plugin-sdk/channel-entry-contract";
-
-const DEDUP_TTL_MS = 24 * 60 * 60 * 1000;
-const STORE_MAX_ENTRIES = 10_000;
-
-type LegacyDedupeData = Record<string, number>;
-
-function safeNamespaceFromFileName(fileName: string): string | null {
-  if (!fileName.endsWith(".json")) {
-    return null;
-  }
-  const namespace = fileName.slice(0, -".json".length).trim();
-  return namespace ? namespace : null;
-}
-
-function readLegacyDedupeData(filePath: string): LegacyDedupeData {
-  try {
-    const parsed = JSON.parse(fs.readFileSync(filePath, "utf8")) as unknown;
-    if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
-      return {};
-    }
-    const out: LegacyDedupeData = {};
-    for (const [messageId, seenAt] of Object.entries(parsed as Record<string, unknown>)) {
-      if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) {
-        out[messageId] = seenAt;
-      }
-    }
-    return out;
-  } catch {
-    return {};
-  }
-}
-
-function dedupeStoreKey(namespace: string, messageId: string): string {
-  return createHash("sha256")
-    .update(`${namespace}\0${messageId}`, "utf8")
-    .digest("hex")
-    .slice(0, 32);
-}
-
-function remainingTtlMs(seenAt: number, now: number): number {
-  return Math.max(1, DEDUP_TTL_MS - (now - seenAt));
-}
-
-function buildMigrationEntries(namespace: string, sourcePath: string, now: number) {
-  return Object.entries(readLegacyDedupeData(sourcePath)).flatMap(([messageId, seenAt]) => {
-    if (now - seenAt >= DEDUP_TTL_MS) {
-      return [];
-    }
-    return [
-      {
-        key: dedupeStoreKey(namespace, messageId),
-        value: { namespace, messageId, seenAt },
-        ttlMs: remainingTtlMs(seenAt, now),
-      },
-    ];
-  });
-}
-
-export const detectFeishuLegacyStateMigrations: BundledChannelLegacyStateMigrationDetector = ({
-  stateDir,
-}) => {
-  const dedupDir = path.join(stateDir, "feishu", "dedup");
-  let entries: fs.Dirent[];
-  try {
-    entries = fs.readdirSync(dedupDir, { withFileTypes: true });
-  } catch {
-    return [];
-  }
-  const now = Date.now();
-  return entries.flatMap((entry) => {
-    if (!entry.isFile()) {
-      return [];
-    }
-    const namespace = safeNamespaceFromFileName(entry.name);
-    if (!namespace) {
-      return [];
-    }
-    const sourcePath = path.join(dedupDir, entry.name);
-    const migrationEntries = buildMigrationEntries(namespace, sourcePath, now);
-    if (migrationEntries.length === 0) {
-      return [];
-    }
-    return [
-      {
-        kind: "plugin-state-import" as const,
-        label: `Feishu ${namespace} dedupe`,
-        sourcePath,
-        targetPath: `plugin state:dedup.${namespace}`,
-        pluginId: "feishu",
-        namespace: `dedup.${namespace}`,
-        maxEntries: STORE_MAX_ENTRIES,
-        scopeKey: "",
-        cleanupSource: "rename" as const,
-        readEntries: () => buildMigrationEntries(namespace, sourcePath, now),
-      },
-    ];
-  });
-};

extensions/feishu/src/dedup.test.ts

@@ -1,39 +1,30 @@
 // Feishu tests cover dedup plugin behavior.
-import fs from "node:fs/promises";
+import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
-import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
-import {
-  createPluginStateSyncKeyedStoreForTests,
-  resetPluginStateStoreForTests,
-} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import type { PluginRuntime } from "../runtime-api.js";
 import {
+  claimUnprocessedFeishuMessage,
+  finalizeFeishuMessageProcessing,
   hasProcessedFeishuMessage,
+  recordProcessedFeishuMessage,
+  releaseFeishuMessageProcessing,
   testingHooks,
-  tryRecordMessagePersistent,
   warmupDedupFromPluginState,
 } from "./dedup.js";
-import { setFeishuRuntime } from "./runtime.js";
 
 let tempDir: string | undefined;
 let previousStateDir: string | undefined;
 
-beforeEach(async () => {
+beforeEach(() => {
   previousStateDir = process.env.OPENCLAW_STATE_DIR;
-  tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-dedup-"));
+  tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-feishu-dedup-"));
   process.env.OPENCLAW_STATE_DIR = tempDir;
-  setFeishuRuntime({
-    state: {
-      openSyncKeyedStore: (options: OpenKeyedStoreOptions) =>
-        createPluginStateSyncKeyedStoreForTests("feishu", options),
-    },
-  } as unknown as PluginRuntime);
   testingHooks.resetFeishuDedupForTests();
 });
 
-afterEach(async () => {
+afterEach(() => {
   vi.useRealTimers();
   testingHooks.resetFeishuDedupForTests();
   resetPluginStateStoreForTests();
@@ -43,53 +34,132 @@ afterEach(async () => {
     process.env.OPENCLAW_STATE_DIR = previousStateDir;
   }
   if (tempDir) {
-    await fs.rm(tempDir, { recursive: true, force: true });
+    fs.rmSync(tempDir, { recursive: true, force: true });
   }
   tempDir = undefined;
 });
 
-describe("Feishu persistent dedupe", () => {
-  it("records message ids in plugin state", async () => {
-    await expect(tryRecordMessagePersistent("msg-1", "account-a")).resolves.toBe(true);
-    await expect(tryRecordMessagePersistent("msg-1", "account-a")).resolves.toBe(false);
+// Simulates a process restart: a fresh guard has empty memory and no in-flight
+// claims, so any duplicate verdict must come from the persisted SQLite rows.
+function restartFeishuDedup(): void {
+  testingHooks.resetFeishuDedupForTests();
+}
+
+describe("Feishu claimable dedupe", () => {
+  it("drops a duplicate message within the TTL after commit", async () => {
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-1", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+    await expect(recordProcessedFeishuMessage("msg-1", "account-a")).resolves.toBe(true);
+
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-1", namespace: "account-a" }),
+    ).resolves.toBe("duplicate");
     await expect(hasProcessedFeishuMessage("msg-1", "account-a")).resolves.toBe(true);
     await expect(hasProcessedFeishuMessage("msg-1", "account-b")).resolves.toBe(false);
   });
 
+  it("reports an in-flight claim and lets a released claim retry", async () => {
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }),
+    ).resolves.toBe("inflight");
+
+    releaseFeishuMessageProcessing("msg-2", "account-a");
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+  });
+
+  it("does not persist released claims across a restart", async () => {
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-3", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+    releaseFeishuMessageProcessing("msg-3", "account-a");
+
+    restartFeishuDedup();
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-3", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+  });
+
+  it("prevents replay after a restart once a message is committed", async () => {
+    await expect(
+      finalizeFeishuMessageProcessing({ messageId: "msg-4", namespace: "account-a" }),
+    ).resolves.toBe(true);
+
+    restartFeishuDedup();
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-4", namespace: "account-a" }),
+    ).resolves.toBe("duplicate");
+    await expect(
+      finalizeFeishuMessageProcessing({ messageId: "msg-4", namespace: "account-a" }),
+    ).resolves.toBe(false);
+  });
+
+  it("commits a held claim without reclaiming it", async () => {
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-5", namespace: "account-a" }),
+    ).resolves.toBe("claimed");
+    await expect(
+      finalizeFeishuMessageProcessing({
+        messageId: "msg-5",
+        namespace: "account-a",
+        claimHeld: true,
+      }),
+    ).resolves.toBe(true);
+    await expect(
+      finalizeFeishuMessageProcessing({
+        messageId: "msg-5",
+        namespace: "account-a",
+        claimHeld: true,
+      }),
+    ).resolves.toBe(false);
+  });
+
+  it("dedupes cross-account broadcast claims through the shared namespace", async () => {
+    // Multi-account groups deliver the same event once per bot account; the
+    // shared "broadcast" namespace lets the first account claim dispatch.
+    await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(true);
+    await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false);
+
+    restartFeishuDedup();
+    await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false);
+  });
+
   it("warms memory from persisted plugin state", async () => {
-    await expect(tryRecordMessagePersistent("msg-2", "account-a")).resolves.toBe(true);
-    testingHooks.resetFeishuDedupMemoryForTests();
+    await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(true);
+    restartFeishuDedup();
 
     await expect(warmupDedupFromPluginState("account-a")).resolves.toBe(1);
-    await expect(tryRecordMessagePersistent("msg-2", "account-a")).resolves.toBe(false);
+    await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(false);
   });
 
-  it("ignores expired persisted entries", async () => {
+  it("ignores committed messages after the TTL expires", async () => {
     vi.useFakeTimers();
     vi.setSystemTime(1_000);
-    await expect(tryRecordMessagePersistent("msg-3", "account-a")).resolves.toBe(true);
-    testingHooks.resetFeishuDedupMemoryForTests();
+    await expect(recordProcessedFeishuMessage("msg-8", "account-a")).resolves.toBe(true);
+    restartFeishuDedup();
 
     vi.setSystemTime(1_000 + 24 * 60 * 60 * 1000 + 1);
-    await expect(hasProcessedFeishuMessage("msg-3", "account-a")).resolves.toBe(false);
+    await expect(hasProcessedFeishuMessage("msg-8", "account-a")).resolves.toBe(false);
   });
 
-  it("ignores legacy JSON dedupe files at runtime", async () => {
-    vi.useFakeTimers();
-    vi.setSystemTime(2_000);
-    const legacyPath = path.join(tempDir as string, "feishu", "dedup", "account-a.json");
-    await fs.mkdir(path.dirname(legacyPath), { recursive: true });
-    await fs.writeFile(
-      legacyPath,
-      JSON.stringify({
-        "msg-legacy": 1_000,
-        "msg-expired": 2_000 - 24 * 60 * 60 * 1000 - 1,
-      }),
-      "utf8",
-    );
+  it("keeps deduping in memory and logs when plugin-state persistence fails", async () => {
+    // A regular file where the state dir should be makes every SQLite open fail.
+    const blockedPath = path.join(tempDir as string, "not-a-dir");
+    fs.writeFileSync(blockedPath, "x", "utf8");
+    process.env.OPENCLAW_STATE_DIR = path.join(blockedPath, "nested");
+    const log = vi.fn();
 
-    await expect(hasProcessedFeishuMessage("msg-legacy", "account-a")).resolves.toBe(false);
-    await expect(tryRecordMessagePersistent("msg-legacy", "account-a")).resolves.toBe(true);
-    await expect(hasProcessedFeishuMessage("msg-expired", "account-a")).resolves.toBe(false);
+    await expect(recordProcessedFeishuMessage("msg-9", "account-a", log)).resolves.toBe(true);
+    await expect(
+      claimUnprocessedFeishuMessage({ messageId: "msg-9", namespace: "account-a", log }),
+    ).resolves.toBe("duplicate");
+    expect(log).toHaveBeenCalledWith(
+      expect.stringContaining("feishu-dedup: persistent state error"),
+    );
   });
 });

extensions/feishu/src/dedup.ts

@@ -1,304 +1,157 @@
-// Feishu plugin module implements dedup behavior.
-import { createHash } from "node:crypto";
-import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
-import {
-  releaseFeishuMessageProcessing,
-  tryBeginFeishuMessageProcessing,
-} from "./processing-claims.js";
-import { getFeishuRuntime } from "./runtime.js";
+// Feishu inbound replay protection rides the core claimable dedupe: Feishu
+// redelivers events after reconnects/restarts and multi-account groups receive
+// the same event once per bot, so handlers claim a dedupe key before
+// processing, commit once handling is dispatched, and release on retryable
+// failure so the event can be redelivered.
+import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe";
 
+// Persisted namespaces resolve to `feishu.dedup.<namespace hash>` in the shared
+// plugin-state SQLite store. Rows from the retired hand-rolled `dedup.*` store
+// are dropped without import: replay protection is cache and rebuilds after
+// upgrade, leaving only a brief unclean-shutdown redelivery gap.
+const DEDUPE_NAMESPACE_PREFIX = "feishu.dedup";
 // Persistent TTL: 24 hours — survives restarts & WebSocket reconnects.
 const DEDUP_TTL_MS = 24 * 60 * 60 * 1000;
 const MEMORY_MAX_SIZE = 1_000;
 const STORE_MAX_ENTRIES = 10_000;
-type FeishuDedupStoreEntry = {
-  namespace: string;
-  messageId: string;
-  seenAt: number;
-};
 
-const memory = new Map<string, number>();
-const cachedDedupStores = new Map<string, PluginStateSyncKeyedStore<FeishuDedupStoreEntry>>();
+type FeishuDedupeLog = (...args: unknown[]) => void;
 
-function normalizeMessageId(messageId: string | undefined | null): string | null {
-  const trimmed = messageId?.trim();
-  return trimmed ? trimmed : null;
-}
+type FeishuMessageClaim = "claimed" | "duplicate" | "inflight";
 
-function normalizeNamespace(namespace?: string): string {
-  return namespace?.trim() || "global";
-}
-
-function pluginStateNamespace(namespace: string): string {
-  return `dedup.${namespace.replace(/[^a-zA-Z0-9_-]/g, "_")}`;
-}
-
-function openDedupStore(namespace: string): PluginStateSyncKeyedStore<FeishuDedupStoreEntry> {
-  const stateNamespace = pluginStateNamespace(namespace);
-  const cached = cachedDedupStores.get(stateNamespace);
-  if (cached) {
-    return cached;
-  }
-  const store = getFeishuRuntime().state.openSyncKeyedStore<FeishuDedupStoreEntry>({
-    namespace: stateNamespace,
-    maxEntries: STORE_MAX_ENTRIES,
-    defaultTtlMs: DEDUP_TTL_MS,
+function createFeishuDedupeGuard() {
+  return createClaimableDedupe({
+    pluginId: "feishu",
+    namespacePrefix: DEDUPE_NAMESPACE_PREFIX,
+    ttlMs: DEDUP_TTL_MS,
+    memoryMaxSize: MEMORY_MAX_SIZE,
+    stateMaxEntries: STORE_MAX_ENTRIES,
   });
-  cachedDedupStores.set(stateNamespace, store);
-  return store;
 }
 
-function dedupeStoreKey(namespace: string, messageId: string): string {
-  return createHash("sha256")
-    .update(`${namespace}\0${messageId}`, "utf8")
-    .digest("hex")
-    .slice(0, 32);
+let guard = createFeishuDedupeGuard();
+
+function dedupeKey(messageId: string | undefined | null): string {
+  return messageId?.trim() ?? "";
 }
 
-function memoryKey(namespace: string, messageId: string): string {
-  return `${namespace}\0${messageId}`;
+function dedupeOptions(namespace: string | undefined, log: FeishuDedupeLog | undefined) {
+  return {
+    ...(namespace ? { namespace } : {}),
+    // Persistence is best effort: a broken state DB must never block inbound
+    // handling, so disk errors surface to the caller's log while the memory
+    // layer keeps deduping.
+    ...(log
+      ? {
+          onDiskError: (error: unknown) =>
+            log(`feishu-dedup: persistent state error: ${String(error)}`),
+        }
+      : {}),
+  };
 }
 
-function isRecent(seenAt: number | undefined, now = Date.now()): boolean {
-  return typeof seenAt === "number" && Number.isFinite(seenAt) && now - seenAt < DEDUP_TTL_MS;
-}
-
-function pruneMemory(now = Date.now()): void {
-  for (const [key, seenAt] of memory) {
-    if (!isRecent(seenAt, now)) {
-      memory.delete(key);
-    }
-  }
-  if (memory.size <= MEMORY_MAX_SIZE) {
-    return;
-  }
-  const toRemove = Array.from(memory.entries())
-    .toSorted(([, left], [, right]) => left - right)
-    .slice(0, memory.size - MEMORY_MAX_SIZE);
-  for (const [key] of toRemove) {
-    memory.delete(key);
-  }
-}
-
-function remember(namespace: string, messageId: string, seenAt = Date.now()): void {
-  memory.set(memoryKey(namespace, messageId), seenAt);
-  pruneMemory(seenAt);
-}
-
-function hasMemory(namespace: string, messageId: string, now = Date.now()): boolean {
-  const key = memoryKey(namespace, messageId);
-  const seenAt = memory.get(key);
-  if (isRecent(seenAt, now)) {
-    return true;
-  }
-  memory.delete(key);
-  return false;
-}
-
-export { releaseFeishuMessageProcessing, tryBeginFeishuMessageProcessing };
-
+/**
+ * Claims a dedupe key for exclusive handling. Duplicate (already committed)
+ * and in-flight keys are reported; blank keys fail open as claimed so an
+ * unidentifiable event is never suppressed.
+ */
 export async function claimUnprocessedFeishuMessage(params: {
   messageId: string | undefined | null;
   namespace?: string;
-  log?: (...args: unknown[]) => void;
-}): Promise<"claimed" | "duplicate" | "inflight" | "invalid"> {
-  const { messageId, namespace = "global", log } = params;
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
-    return "invalid";
+  log?: FeishuDedupeLog;
+}): Promise<FeishuMessageClaim> {
+  const key = dedupeKey(params.messageId);
+  if (!key) {
+    return "claimed";
   }
-  if (await hasProcessedFeishuMessage(normalizedMessageId, namespace, log)) {
-    return "duplicate";
-  }
-  if (!tryBeginFeishuMessageProcessing(normalizedMessageId, namespace)) {
-    return "inflight";
-  }
-  return "claimed";
+  return (await guard.claim(key, dedupeOptions(params.namespace, params.log))).kind;
 }
 
+/** Drops an uncommitted claim so a failed handler can retry the message. */
+export function releaseFeishuMessageProcessing(
+  messageId: string | undefined | null,
+  namespace = "global",
+): void {
+  const key = dedupeKey(messageId);
+  if (key) {
+    guard.release(key, { namespace });
+  }
+}
+
+/**
+ * Claims (unless the caller already holds the claim) and commits a message.
+ * False means another handler owns it, it was already handled, or the key is
+ * blank; handlers must skip dispatch then.
+ */
 export async function finalizeFeishuMessageProcessing(params: {
   messageId: string | undefined | null;
   namespace?: string;
-  log?: (...args: unknown[]) => void;
+  log?: FeishuDedupeLog;
   claimHeld?: boolean;
 }): Promise<boolean> {
-  const { messageId, namespace = "global", log, claimHeld = false } = params;
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
+  const key = dedupeKey(params.messageId);
+  if (!key) {
     return false;
   }
-  if (!claimHeld && !tryBeginFeishuMessageProcessing(normalizedMessageId, namespace)) {
+  const options = dedupeOptions(params.namespace, params.log);
+  if (!params.claimHeld && (await guard.claim(key, options)).kind !== "claimed") {
     return false;
   }
-  if (!(await tryRecordMessagePersistent(normalizedMessageId, namespace, log))) {
-    releaseFeishuMessageProcessing(normalizedMessageId, namespace);
-    return false;
-  }
-  return true;
+  return await guard.commit(key, options);
 }
 
+/** Records a handled message so restart/replay cannot dispatch it again; false when already recorded. */
 export async function recordProcessedFeishuMessage(
   messageId: string | undefined | null,
   namespace = "global",
-  log?: (...args: unknown[]) => void,
+  log?: FeishuDedupeLog,
 ): Promise<boolean> {
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
+  const key = dedupeKey(messageId);
+  if (!key) {
     return false;
   }
-  return await tryRecordMessagePersistent(normalizedMessageId, namespace, log);
+  return await guard.commit(key, dedupeOptions(namespace, log));
 }
 
+/** Forgets a recorded message so a retryable synthetic event can be handled on redelivery. */
 export async function forgetProcessedFeishuMessage(
   messageId: string | undefined | null,
   namespace = "global",
-  log?: (...args: unknown[]) => void,
+  log?: FeishuDedupeLog,
 ): Promise<boolean> {
-  const normalizedNamespace = normalizeNamespace(namespace);
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
-    return false;
-  }
-  memory.delete(memoryKey(normalizedNamespace, normalizedMessageId));
-  const key = dedupeStoreKey(normalizedNamespace, normalizedMessageId);
-  try {
-    return openDedupStore(normalizedNamespace).delete(key);
-  } catch (error) {
-    log?.(`feishu-dedup: persistent delete failed: ${String(error)}`);
+  const key = dedupeKey(messageId);
+  if (!key) {
     return false;
   }
+  return await guard.forget(key, dedupeOptions(namespace, log));
 }
 
+/** Checks recency without claiming or recording. */
 export async function hasProcessedFeishuMessage(
   messageId: string | undefined | null,
   namespace = "global",
-  log?: (...args: unknown[]) => void,
+  log?: FeishuDedupeLog,
 ): Promise<boolean> {
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
+  const key = dedupeKey(messageId);
+  if (!key) {
     return false;
   }
-  return hasRecordedMessagePersistent(normalizedMessageId, namespace, log);
-}
-
-export async function tryRecordMessagePersistent(
-  messageId: string,
-  namespace = "global",
-  log?: (...args: unknown[]) => void,
-): Promise<boolean> {
-  const normalizedNamespace = normalizeNamespace(namespace);
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
-    return true;
-  }
-  const now = Date.now();
-  if (hasMemory(normalizedNamespace, normalizedMessageId, now)) {
-    return false;
-  }
-  const key = dedupeStoreKey(normalizedNamespace, normalizedMessageId);
-  try {
-    const store = openDedupStore(normalizedNamespace);
-    const existing = store.lookup(key);
-    const existingSeenAt = existing?.seenAt;
-    if (isRecent(existingSeenAt, now)) {
-      remember(normalizedNamespace, normalizedMessageId, existingSeenAt);
-      return false;
-    }
-    const recorded = store.registerIfAbsent(
-      key,
-      {
-        namespace: normalizedNamespace,
-        messageId: normalizedMessageId,
-        seenAt: now,
-      },
-      { ttlMs: DEDUP_TTL_MS },
-    );
-    if (!recorded) {
-      const current = store.lookup(key);
-      const currentSeenAt = current?.seenAt;
-      if (isRecent(currentSeenAt, now)) {
-        remember(normalizedNamespace, normalizedMessageId, currentSeenAt);
-        return false;
-      }
-      store.register(
-        key,
-        {
-          namespace: normalizedNamespace,
-          messageId: normalizedMessageId,
-          seenAt: now,
-        },
-        { ttlMs: DEDUP_TTL_MS },
-      );
-    }
-    remember(normalizedNamespace, normalizedMessageId, now);
-    return true;
-  } catch (error) {
-    log?.(`feishu-dedup: persistent state error, falling back to memory: ${String(error)}`);
-    remember(normalizedNamespace, normalizedMessageId, now);
-    return true;
-  }
-}
-
-async function hasRecordedMessagePersistent(
-  messageId: string,
-  namespace = "global",
-  log?: (...args: unknown[]) => void,
-): Promise<boolean> {
-  const normalizedNamespace = normalizeNamespace(namespace);
-  const normalizedMessageId = normalizeMessageId(messageId);
-  if (!normalizedMessageId) {
-    return false;
-  }
-  const now = Date.now();
-  if (hasMemory(normalizedNamespace, normalizedMessageId, now)) {
-    return true;
-  }
-  try {
-    const store = openDedupStore(normalizedNamespace);
-    const existing = store.lookup(dedupeStoreKey(normalizedNamespace, normalizedMessageId));
-    const existingSeenAt = existing?.seenAt;
-    if (!isRecent(existingSeenAt, now)) {
-      return false;
-    }
-    remember(normalizedNamespace, normalizedMessageId, existingSeenAt);
-    return true;
-  } catch (error) {
-    log?.(`feishu-dedup: persistent peek failed: ${String(error)}`);
-    return hasMemory(normalizedNamespace, normalizedMessageId, now);
-  }
+  return await guard.hasRecent(key, dedupeOptions(namespace, log));
 }
 
+/** Loads recent persisted entries into memory at account start; returns the loaded count. */
 export async function warmupDedupFromPluginState(
   namespace: string,
-  log?: (...args: unknown[]) => void,
+  log?: FeishuDedupeLog,
 ): Promise<number> {
-  const normalizedNamespace = normalizeNamespace(namespace);
-  try {
-    let loaded = 0;
-    const now = Date.now();
-    for (const entry of openDedupStore(normalizedNamespace).entries()) {
-      if (entry.value.namespace !== normalizedNamespace || !isRecent(entry.value.seenAt, now)) {
-        continue;
-      }
-      remember(normalizedNamespace, entry.value.messageId, entry.value.seenAt);
-      loaded++;
-    }
-    return loaded;
-  } catch (error) {
-    log?.(`feishu-dedup: warmup persistent state error: ${String(error)}`);
-    return 0;
-  }
+  return await guard.warmup(namespace, (error) =>
+    log?.(`feishu-dedup: warmup persistent state error: ${String(error)}`),
+  );
 }
 
 export const testingHooks = {
+  /** Drops in-flight claims and process memory; persisted rows follow the test's state dir. */
   resetFeishuDedupForTests() {
-    memory.clear();
-    for (const store of cachedDedupStores.values()) {
-      store.clear();
-    }
-    cachedDedupStores.clear();
-  },
-  resetFeishuDedupMemoryForTests() {
-    memory.clear();
+    guard = createFeishuDedupeGuard();
   },
 };

extensions/feishu/src/lifecycle.test-support.ts

@@ -1,7 +1,6 @@
 // Feishu plugin module implements lifecycle support behavior.
 import { vi, type Mock } from "vitest";
 import { testingHooks as dedupTestingHooks } from "./dedup.js";
-import { testingHooks as processingClaimTestingHooks } from "./processing-claims.js";
 
 type BoundConversation = {
   bindingId: string;
@@ -92,7 +91,6 @@ export function getFeishuLifecycleTestMocks(): FeishuLifecycleTestMocks {
 
 export function resetFeishuLifecycleTestMocks(): void {
   dedupTestingHooks.resetFeishuDedupForTests();
-  processingClaimTestingHooks.resetFeishuMessageProcessingClaimsForTests();
   for (const mock of Object.values(feishuLifecycleTestMocks)) {
     mock.mockReset();
   }

extensions/feishu/src/monitor.message-handler.ts

@@ -1,13 +1,10 @@
 // Feishu plugin module implements monitor.message handler behavior.
 import { isRecord, readStringValue as readString } from "openclaw/plugin-sdk/string-coerce-runtime";
 import type { ClawdbotConfig, HistoryEntry, PluginRuntime, RuntimeEnv } from "../runtime-api.js";
+import { claimUnprocessedFeishuMessage, releaseFeishuMessageProcessing } from "./dedup.js";
 import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js";
 import type { FeishuMessageEvent } from "./event-types.js";
 import { isMentionForwardRequest } from "./mention.js";
-import {
-  releaseFeishuMessageProcessing,
-  tryBeginFeishuMessageProcessing,
-} from "./processing-claims.js";
 import { createSequentialQueue } from "./sequential-queue.js";
 import type { FeishuChatType } from "./types.js";
 
@@ -347,8 +344,13 @@ export function createFeishuMessageReceiveHandler({
       return;
     }
     const messageDedupeKey = resolveFeishuMessageDedupeKey(event);
-    if (!tryBeginFeishuMessageProcessing(messageDedupeKey, accountId)) {
-      log(`feishu[${accountId}]: dropping duplicate event for message ${messageId}`);
+    const claim = await claimUnprocessedFeishuMessage({
+      messageId: messageDedupeKey,
+      namespace: accountId,
+      log,
+    });
+    if (claim !== "claimed") {
+      log(`feishu[${accountId}]: dropping ${claim} event for message ${messageId}`);
       return;
     }
     const processMessage = async () => {

extensions/feishu/src/monitor.reaction.test.ts

@@ -234,7 +234,7 @@ function expectParsedFirstDispatchedEvent(botOpenId = "ou_bot") {
 }
 
 function setDedupPassThroughMocks(): void {
-  vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true);
+  vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed");
   vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true);
   vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false);
 }
@@ -595,7 +595,7 @@ describe("Feishu inbound debounce regressions", () => {
   });
 
   it("passes prefetched botName through to handleFeishuMessage", async () => {
-    vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true);
+    vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed");
     vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true);
     vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false);
     const onMessage = await setupDebounceMonitor({ botName: "OpenClaw Bot" });
@@ -678,7 +678,7 @@ describe("Feishu inbound debounce regressions", () => {
   });
 
   it("excludes previously processed retries from combined debounce text", async () => {
-    vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true);
+    vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed");
     vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true);
     setStaleRetryMocks();
     const onMessage = await setupDebounceMonitor();
@@ -704,7 +704,7 @@ describe("Feishu inbound debounce regressions", () => {
   });
 
   it("uses latest fresh message id when debounce batch ends with stale retry", async () => {
-    vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true);
+    vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed");
     const recordSpy = vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true);
     setStaleRetryMocks("om_old_latest_fresh");
     const onMessage = await setupDebounceMonitor();

extensions/feishu/src/processing-claims.ts

@@ -1,66 +0,0 @@
-// Feishu plugin module implements processing claims behavior.
-const EVENT_DEDUP_TTL_MS = 5 * 60 * 1000;
-const EVENT_MEMORY_MAX_SIZE = 2_000;
-
-const processingClaims = new Map<string, number>();
-
-function resolveEventDedupeKey(
-  namespace: string,
-  messageId: string | undefined | null,
-): string | null {
-  const trimmed = messageId?.trim();
-  return trimmed ? `${namespace}:${trimmed}` : null;
-}
-
-function pruneProcessingClaims(now: number): void {
-  const cutoff = now - EVENT_DEDUP_TTL_MS;
-  for (const [key, seenAt] of processingClaims) {
-    if (seenAt < cutoff) {
-      processingClaims.delete(key);
-    }
-  }
-  while (processingClaims.size > EVENT_MEMORY_MAX_SIZE) {
-    const oldestKey = processingClaims.keys().next().value;
-    if (!oldestKey) {
-      return;
-    }
-    processingClaims.delete(oldestKey);
-  }
-}
-
-export function tryBeginFeishuMessageProcessing(
-  messageId: string | undefined | null,
-  namespace = "global",
-): boolean {
-  const key = resolveEventDedupeKey(namespace, messageId);
-  if (!key) {
-    return true;
-  }
-  const now = Date.now();
-  pruneProcessingClaims(now);
-  if (processingClaims.has(key)) {
-    processingClaims.delete(key);
-    processingClaims.set(key, now);
-    pruneProcessingClaims(now);
-    return false;
-  }
-  processingClaims.set(key, now);
-  pruneProcessingClaims(now);
-  return true;
-}
-
-export function releaseFeishuMessageProcessing(
-  messageId: string | undefined | null,
-  namespace = "global",
-): void {
-  const key = resolveEventDedupeKey(namespace, messageId);
-  if (key) {
-    processingClaims.delete(key);
-  }
-}
-
-export const testingHooks = {
-  resetFeishuMessageProcessingClaimsForTests() {
-    processingClaims.clear();
-  },
-};

scripts/check-no-random-messaging-tmp.mjs

@@ -2,7 +2,6 @@
 
 // Blocks host-random tmpdir usage in messaging/channel runtime sources.
 import ts from "typescript";
-import { bundledPluginFile } from "./lib/bundled-plugin-paths.mjs";
 import { runCallsiteGuard } from "./lib/callsite-guard.mjs";
 import {
   collectCallExpressionLines,
@@ -21,8 +20,6 @@ export const messagingTmpdirGuardSourceRoots = [
   "src/media-understanding",
   "extensions",
 ];
-const allowedRelativePaths = new Set([bundledPluginFile("feishu", "src/dedup.ts")]);
-
 function collectOsTmpdirImports(sourceFile) {
   const osModuleSpecifiers = new Set(["node:os", "os"]);
   const osNamespaceOrDefault = new Set();
@@ -85,7 +82,6 @@ export async function main() {
     importMetaUrl: import.meta.url,
     sourceRoots: messagingTmpdirGuardSourceRoots,
     findCallLines: findMessagingTmpdirCallLines,
-    skipRelativePath: (relativePath) => allowedRelativePaths.has(relativePath),
     header: "Found os.tmpdir()/tmpdir() usage in messaging/channel runtime sources:",
     footer:
       "Use resolvePreferredOpenClawTmpDir() or plugin-sdk temp helpers instead of host tmp defaults.",

src/cli/program/config-guard.test.ts

@@ -363,7 +363,6 @@ describe("ensureConfigReady", () => {
   it.each([
     ["Discord model picker preferences", "discord/model-picker-preferences.json"],
     ["Discord thread bindings", "discord/thread-bindings.json"],
-    ["Feishu dedupe sidecar", "feishu/dedup/default.json"],
     ["Telegram bot info cache", "telegram/bot-info-default.json"],
     ["Telegram update offset", "telegram/update-offset-default.json"],
     ["Telegram sticker cache", "telegram/sticker-cache.json"],

src/cli/program/config-guard.ts

@@ -88,9 +88,6 @@ function hasBundledChannelLegacyStateMigrationInputs(stateDir: string, oauthDir:
   ) {
     return true;
   }
-  if (dirHasFile(path.join(stateDir, "feishu", "dedup"), (name) => name.endsWith(".json"))) {
-    return true;
-  }
   if (hasLegacyIMessageStateFiles(stateDir)) {
     return true;
   }