refactor(plugin-sdk): persist dedupe state in sqlite

2026-08-07 18:42:25 +00:00 · 2026-06-07 02:41:45 -07:00
parent a4e78aec4b
commit bab18d567b
7 changed files with 482 additions and 190 deletions
--- a/extensions/nextcloud-talk/src/doctor.test.ts
+++ b/extensions/nextcloud-talk/src/doctor.test.ts
@@ -1,5 +1,10 @@
 // Nextcloud Talk tests cover doctor plugin behavior.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import { createNextcloudTalkReplayGuard } from "./replay-guard.js";

 const hoisted = vi.hoisted(() => ({
  probeNextcloudTalkBotResponseFeature: vi.fn(),
@@ -24,6 +29,7 @@ function getNextcloudTalkCompatibilityNormalizer(): NonNullable<
 describe("nextcloud-talk doctor", () => {
  beforeEach(() => {
    hoisted.probeNextcloudTalkBotResponseFeature.mockReset();
+    resetPluginStateStoreForTests();
  });

  it("normalizes legacy private-network aliases", () => {
@@ -85,4 +91,48 @@ describe("nextcloud-talk doctor", () => {
      '- channels.nextcloud-talk.default: Nextcloud Talk bot "OpenClaw" (1) is missing the response feature (features=9); outbound replies will fail.',
    ]);
  });
+
+  it("migrates legacy replay dedupe JSON into SQLite during doctor repair", async () => {
+    const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-nextcloud-doctor-"));
+    const legacyDir = path.join(stateDir, "nextcloud-talk", "replay-dedupe");
+    const legacyPath = path.join(legacyDir, "account-a.json");
+    await fs.mkdir(legacyDir, { recursive: true });
+    await fs.writeFile(
+      legacyPath,
+      JSON.stringify({
+        "room-1:msg-1": Date.now(),
+      }),
+    );
+
+    const mutation = await nextcloudTalkDoctor.repairConfig?.({
+      cfg: {
+        channels: {
+          "nextcloud-talk": {
+            accounts: {
+              "account-a": {
+                baseUrl: "https://cloud.example.com",
+                botSecret: "secret",
+              },
+            },
+          },
+        },
+      } as never,
+      doctorFixCommand: "openclaw doctor --fix",
+      env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
+    });
+
+    expect(mutation?.changes.join("\n")).toContain(
+      'Migrated Nextcloud Talk replay dedupe cache for account "account-a" to SQLite',
+    );
+    await expect(fs.access(legacyPath)).rejects.toThrow();
+
+    const guard = createNextcloudTalkReplayGuard({ stateDir });
+    await expect(
+      guard.shouldProcessMessage({
+        accountId: "account-a",
+        roomToken: "room-1",
+        messageId: "msg-1",
+      }),
+    ).resolves.toBe(false);
+  });
 });
--- a/extensions/nextcloud-talk/src/doctor.ts
+++ b/extensions/nextcloud-talk/src/doctor.ts
@@ -1,13 +1,42 @@
 // Nextcloud Talk plugin module implements doctor behavior.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
 import type { ChannelDoctorAdapter } from "openclaw/plugin-sdk/channel-contract";
+import { migratePersistentDedupeLegacyJsonFile } from "openclaw/plugin-sdk/persistent-dedupe";
+import { resolveStateDir } from "openclaw/plugin-sdk/state-paths";
 import { listNextcloudTalkAccountIds, resolveNextcloudTalkAccount } from "./accounts.js";
 import { probeNextcloudTalkBotResponseFeature } from "./bot-preflight.js";
 import {
  legacyConfigRules as NEXTCLOUD_TALK_LEGACY_CONFIG_RULES,
  normalizeCompatibilityConfig as normalizeNextcloudTalkCompatibilityConfig,
 } from "./doctor-contract.js";
+import {
+  NEXTCLOUD_TALK_PLUGIN_ID,
+  NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX,
+} from "./replay-guard.js";
 import type { CoreConfig } from "./types.js";

+const REPLAY_DEDUPE_TTL_MS = 24 * 60 * 60 * 1000;
+const REPLAY_DEDUPE_MAX_ENTRIES = 10_000;
+
+function sanitizeLegacyReplaySegment(value: string): string {
+  const trimmed = value.trim();
+  if (!trimmed) {
+    return "default";
+  }
+  return trimmed.replace(/[^a-zA-Z0-9_-]/g, "_");
+}
+
+async function fileExists(filePath: string): Promise<boolean> {
+  try {
+    await fs.access(filePath);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
 async function collectNextcloudTalkBotResponseWarnings(params: {
  cfg: CoreConfig;
 }): Promise<string[]> {
@@ -33,9 +62,59 @@ async function collectNextcloudTalkBotResponseWarnings(params: {
  return warnings;
 }

+async function repairNextcloudTalkReplayDedupeState(params: {
+  cfg: CoreConfig;
+  env?: NodeJS.ProcessEnv;
+}): Promise<{ changes: string[]; warnings: string[] }> {
+  const changes: string[] = [];
+  const warnings: string[] = [];
+  const env = params.env ?? process.env;
+  const stateDir = resolveStateDir(env, os.homedir);
+  const replayDir = path.join(stateDir, "nextcloud-talk", "replay-dedupe");
+
+  for (const accountId of listNextcloudTalkAccountIds(params.cfg)) {
+    const legacyPath = path.join(replayDir, `${sanitizeLegacyReplaySegment(accountId)}.json`);
+    if (!(await fileExists(legacyPath))) {
+      continue;
+    }
+    try {
+      const result = await migratePersistentDedupeLegacyJsonFile({
+        filePath: legacyPath,
+        namespace: accountId,
+        ttlMs: REPLAY_DEDUPE_TTL_MS,
+        memoryMaxSize: 0,
+        pluginId: NEXTCLOUD_TALK_PLUGIN_ID,
+        namespacePrefix: NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX,
+        stateMaxEntries: REPLAY_DEDUPE_MAX_ENTRIES,
+        env,
+      });
+      changes.push(
+        `Migrated Nextcloud Talk replay dedupe cache for account "${accountId}" to SQLite (${result.imported} imported, ${result.skippedExpired} expired, ${result.skippedExisting} already current).`,
+      );
+    } catch (error) {
+      warnings.push(
+        `Skipped Nextcloud Talk replay dedupe cache for account "${accountId}": ${String(error)}`,
+      );
+    }
+  }
+
+  return { changes, warnings };
+}
+
 export const nextcloudTalkDoctor: ChannelDoctorAdapter = {
  legacyConfigRules: NEXTCLOUD_TALK_LEGACY_CONFIG_RULES,
  normalizeCompatibilityConfig: normalizeNextcloudTalkCompatibilityConfig,
  collectPreviewWarnings: async ({ cfg }) =>
    await collectNextcloudTalkBotResponseWarnings({ cfg: cfg as CoreConfig }),
+  repairConfig: async ({ cfg, env }) => {
+    const repair = await repairNextcloudTalkReplayDedupeState({
+      cfg: cfg as CoreConfig,
+      ...(env ? { env } : {}),
+    });
+    return {
+      config: cfg,
+      changes: repair.changes,
+      warnings: repair.warnings,
+    };
+  },
 };
--- a/extensions/nextcloud-talk/src/replay-guard.ts
+++ b/extensions/nextcloud-talk/src/replay-guard.ts
@@ -1,18 +1,11 @@
 // Nextcloud Talk plugin module implements replay guard behavior.
-import path from "node:path";
 import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe";

+export const NEXTCLOUD_TALK_PLUGIN_ID = "nextcloud-talk";
+export const NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX = "replay-dedupe";
 const DEFAULT_REPLAY_TTL_MS = 24 * 60 * 60 * 1000;
 const DEFAULT_MEMORY_MAX_SIZE = 1_000;
-const DEFAULT_FILE_MAX_ENTRIES = 10_000;
-
-function sanitizeSegment(value: string): string {
-  const trimmed = value.trim();
-  if (!trimmed) {
-    return "default";
-  }
-  return trimmed.replace(/[^a-zA-Z0-9_-]/g, "_");
-}
+const DEFAULT_STATE_MAX_ENTRIES = 10_000;

 function buildReplayKey(params: { roomToken: string; messageId: string }): string | null {
  const roomToken = params.roomToken.trim();
@@ -27,6 +20,8 @@ type NextcloudTalkReplayGuardOptions = {
  stateDir?: string;
  ttlMs?: number;
  memoryMaxSize?: number;
+  stateMaxEntries?: number;
+  /** @deprecated Use stateMaxEntries. */
  fileMaxEntries?: number;
  onDiskError?: (error: unknown) => void;
 };
@@ -67,14 +62,14 @@ export function createNextcloudTalkReplayGuard(
    stateDir
      ? {
          ...baseOptions,
-          fileMaxEntries: options.fileMaxEntries ?? DEFAULT_FILE_MAX_ENTRIES,
-          resolveFilePath: (namespace) =>
-            path.join(
-              stateDir,
-              "nextcloud-talk",
-              "replay-dedupe",
-              `${sanitizeSegment(namespace)}.json`,
-            ),
+          pluginId: NEXTCLOUD_TALK_PLUGIN_ID,
+          namespacePrefix: NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX,
+          stateMaxEntries:
+            options.stateMaxEntries ?? options.fileMaxEntries ?? DEFAULT_STATE_MAX_ENTRIES,
+          env: {
+            ...process.env,
+            OPENCLAW_STATE_DIR: stateDir,
+          },
          onDiskError: options.onDiskError,
        }
      : baseOptions,
--- a/src/channels/plugins/types.adapters.ts
+++ b/src/channels/plugins/types.adapters.ts
@@ -526,6 +526,7 @@ export type ChannelDoctorAdapter = {
  repairConfig?: (params: {
    cfg: OpenClawConfig;
    doctorFixCommand: string;
+    env?: NodeJS.ProcessEnv;
  }) => ChannelDoctorConfigMutation | Promise<ChannelDoctorConfigMutation>;
  runConfigSequence?: (params: {
    cfg: OpenClawConfig;
--- a/src/commands/doctor/shared/channel-doctor.ts
+++ b/src/commands/doctor/shared/channel-doctor.ts
@@ -407,6 +407,7 @@ export async function collectChannelDoctorRepairMutations(params: {
    const mutation = await entry.doctor.repairConfig?.({
      cfg: nextCfg,
      doctorFixCommand: params.doctorFixCommand,
+      ...(params.env ? { env: params.env } : {}),
    });
    if (!mutation || mutation.changes.length === 0) {
      if (mutation?.warnings?.length) {
--- a/src/plugin-sdk/memory-host-events.test.ts
+++ b/src/plugin-sdk/memory-host-events.test.ts
@@ -3,7 +3,8 @@
 */
 import fs from "node:fs/promises";
 import path from "node:path";
-import { describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it } from "vitest";
+import { resetPluginStateStoreForTests } from "../plugin-state/plugin-state-store.js";
 import {
  appendMemoryHostEvent,
  readMemoryHostEvents,
@@ -18,11 +19,17 @@ function createDedupe(root: string, overrides?: { ttlMs?: number }) {
  return createPersistentDedupe({
    ttlMs: overrides?.ttlMs ?? 24 * 60 * 60 * 1000,
    memoryMaxSize: 100,
-    fileMaxEntries: 1000,
-    resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
+    pluginId: "test-persistent-dedupe",
+    namespacePrefix: "test-dedupe",
+    stateMaxEntries: 1000,
+    env: { ...process.env, OPENCLAW_STATE_DIR: root },
  });
 }

+afterEach(() => {
+  resetPluginStateStoreForTests();
+});
+
 describe("memory host event journal helpers", () => {
  it("appends and reads typed workspace events", async () => {
    const workspaceDir = await createTempDir("memory-host-events-");
@@ -94,8 +101,10 @@ describe("createPersistentDedupe", () => {
    const dedupe = createPersistentDedupe({
      ttlMs: Number.NaN,
      memoryMaxSize: Number.NaN,
-      fileMaxEntries: Number.NaN,
-      resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
+      pluginId: "test-persistent-dedupe",
+      namespacePrefix: "test-bounds",
+      stateMaxEntries: Number.NaN,
+      env: { ...process.env, OPENCLAW_STATE_DIR: root },
    });

    expect(await dedupe.checkAndRecord("m1", { namespace: "a", now: 100 })).toBe(true);
@@ -104,33 +113,32 @@ describe("createPersistentDedupe", () => {
    expect(dedupe.memorySize()).toBe(0);
  });

-  it("falls back to memory-only behavior on disk errors", async () => {
+  it("uses legacy JSON paths only as SQLite namespace identifiers", async () => {
+    const root = await createTempDir("openclaw-legacy-dedupe-");
+    const legacyPath = path.join(root, "legacy.json");
    const dedupe = createPersistentDedupe({
      ttlMs: 10_000,
      memoryMaxSize: 100,
      fileMaxEntries: 1000,
-      resolveFilePath: () => path.join("/dev/null", "dedupe.json"),
+      resolveFilePath: () => legacyPath,
+      env: { ...process.env, OPENCLAW_STATE_DIR: root },
    });

-    expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(true);
-    expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(false);
+    expect(await dedupe.checkAndRecord("sqlite-only", { namespace: "x" })).toBe(true);
+    expect(await dedupe.checkAndRecord("sqlite-only", { namespace: "x" })).toBe(false);
+    await expect(fs.access(legacyPath)).rejects.toThrow();
  });

-  it("warms empty namespaces and skips expired disk entries", async () => {
+  it("warms empty namespaces and ignores retired JSON cache files", async () => {
    const root = await createTempDir("openclaw-dedupe-");
    const emptyReader = createDedupe(root, { ttlMs: 10_000 });
    expect(await emptyReader.warmup("nonexistent")).toBe(0);

-    const oldNow = Date.now() - 2000;
-    await fs.writeFile(
-      path.join(root, "acct.json"),
-      JSON.stringify({ "old-msg": oldNow, "new-msg": Date.now() }),
-    );
+    await fs.writeFile(path.join(root, "acct.json"), JSON.stringify({ "retired-msg": Date.now() }));

    const reader = createDedupe(root, { ttlMs: 1000 });
-    expect(await reader.warmup("acct")).toBe(1);
-    expect(await reader.checkAndRecord("old-msg", { namespace: "acct" })).toBe(true);
-    expect(await reader.checkAndRecord("new-msg", { namespace: "acct" })).toBe(false);
+    expect(await reader.warmup("acct")).toBe(0);
+    expect(await reader.checkAndRecord("retired-msg", { namespace: "acct" })).toBe(true);
  });
 });

@@ -189,8 +197,10 @@ describe("createClaimableDedupe", () => {
    const writer = createClaimableDedupe({
      ttlMs: 10_000,
      memoryMaxSize: 100,
-      fileMaxEntries: 1000,
-      resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
+      pluginId: "test-claimable-dedupe",
+      namespacePrefix: "test-claimable-dedupe",
+      stateMaxEntries: 1000,
+      env: { ...process.env, OPENCLAW_STATE_DIR: root },
    });

    await expect(writer.claim("m1", { namespace: "acct" })).resolves.toEqual({ kind: "claimed" });
@@ -199,8 +209,10 @@ describe("createClaimableDedupe", () => {
    const reader = createClaimableDedupe({
      ttlMs: 10_000,
      memoryMaxSize: 100,
-      fileMaxEntries: 1000,
-      resolveFilePath: (namespace) => path.join(root, `${namespace}.json`),
+      pluginId: "test-claimable-dedupe",
+      namespacePrefix: "test-claimable-dedupe",
+      stateMaxEntries: 1000,
+      env: { ...process.env, OPENCLAW_STATE_DIR: root },
    });

    expect(await reader.hasRecent("m1", { namespace: "acct" })).toBe(true);
--- a/src/plugin-sdk/persistent-dedupe.ts
+++ b/src/plugin-sdk/persistent-dedupe.ts
@@ -1,26 +1,81 @@
 // Persistent dedupe helpers give plugins bounded replay protection across process restarts.
+import { createHash } from "node:crypto";
+import fs from "node:fs/promises";
 import { createDedupeCache } from "../infra/dedupe.js";
 import { resolveNonNegativeIntegerOption } from "../infra/numeric-options.js";
+import {
+  createCorePluginStateSyncKeyedStore,
+  createPluginStateSyncKeyedStore,
+} from "../plugin-state/plugin-state-store.js";
+import type { PluginStateSyncKeyedStore } from "../plugin-state/plugin-state-store.types.js";
 import type { FileLockOptions } from "./file-lock.js";
-import { withFileLock } from "./file-lock.js";
-import { readJsonFileWithFallback, writeJsonFileAtomically } from "./json-store.js";

-type PersistentDedupeData = Record<string, number>;
+const LEGACY_PATH_OWNER_ID = "core:persistent-dedupe";
+const DEFAULT_NAMESPACE_PREFIX = "persistent-dedupe";

-/** Configuration for a disk-backed dedupe namespace cache. */
-export type PersistentDedupeOptions = {
+type PersistentDedupeEntry = {
+  key: string;
+  seenAt: number;
+};
+
+type PersistentDedupeBaseOptions = {
  /** Milliseconds a recorded key remains recent; `0` keeps keys until cache pruning. */
  ttlMs: number;
-  /** Maximum process-local cache entries used before consulting disk. */
+  /** Maximum process-local cache entries used before consulting SQLite. */
  memoryMaxSize: number;
-  /** Maximum persisted entries retained per namespace file. */
-  fileMaxEntries: number;
-  /** Maps a namespace to the JSON file that stores its persisted dedupe timestamps. */
-  resolveFilePath: (namespace: string) => string;
-  lockOptions?: Partial<FileLockOptions>;
  onDiskError?: (error: unknown) => void;
 };

+/** Configuration for a SQLite plugin-state dedupe namespace cache. */
+export type PersistentDedupePluginStateOptions = PersistentDedupeBaseOptions & {
+  /** Plugin id that owns the persisted dedupe namespace. */
+  pluginId: string;
+  /** Prefix for persisted plugin-state namespaces; defaults to `persistent-dedupe`. */
+  namespacePrefix?: string;
+  /** Maximum persisted entries retained per namespace. */
+  stateMaxEntries: number;
+  /** Test/runtime env used to resolve the shared OpenClaw state database. */
+  env?: NodeJS.ProcessEnv;
+  resolveFilePath?: undefined;
+  fileMaxEntries?: undefined;
+  lockOptions?: undefined;
+};
+
+/** Legacy path-shaped configuration. Paths now name SQLite namespaces, not JSON files. */
+export type PersistentDedupeLegacyPathOptions = PersistentDedupeBaseOptions & {
+  pluginId?: undefined;
+  stateMaxEntries?: undefined;
+  namespacePrefix?: undefined;
+  /** Maximum persisted entries retained per legacy namespace. */
+  fileMaxEntries: number;
+  /** Maps a namespace to the retired JSON path; used only to derive a stable SQLite namespace. */
+  resolveFilePath: (namespace: string) => string;
+  /** Test/runtime env used to resolve the shared OpenClaw state database. */
+  env?: NodeJS.ProcessEnv;
+  /** @deprecated File locks are ignored because persistence is SQLite-backed. */
+  lockOptions?: Partial<FileLockOptions>;
+};
+
+/** Configuration for a persisted dedupe namespace cache. */
+export type PersistentDedupeOptions =
+  | PersistentDedupePluginStateOptions
+  | PersistentDedupeLegacyPathOptions;
+
+export type PersistentDedupeLegacyJsonMigrationResult = {
+  imported: number;
+  skippedExpired: number;
+  skippedInvalid: number;
+  skippedExisting: number;
+  removed: boolean;
+};
+
+export type PersistentDedupeLegacyJsonMigrationOptions = PersistentDedupePluginStateOptions & {
+  filePath: string;
+  namespace: string;
+  now?: number;
+  removeFile?: boolean;
+};
+
 /** Per-call options used when checking or recording a dedupe key. */
 export type PersistentDedupeCheckOptions = {
  /** Logical bucket for the key; omitted/blank values use `global`. */
@@ -53,17 +108,15 @@ export type ClaimableDedupeClaimResult =

 /** Options for a claimable dedupe guard, either persistent or memory-only. */
 export type ClaimableDedupeOptions =
+  | PersistentDedupePluginStateOptions
+  | PersistentDedupeLegacyPathOptions
  | {
      ttlMs: number;
      memoryMaxSize: number;
-      resolveFilePath: (namespace: string) => string;
-      fileMaxEntries: number;
-      lockOptions?: Partial<FileLockOptions>;
-      onDiskError?: (error: unknown) => void;
-    }
-  | {
-      ttlMs: number;
-      memoryMaxSize: number;
+      pluginId?: undefined;
+      stateMaxEntries?: undefined;
+      namespacePrefix?: undefined;
+      env?: undefined;
      resolveFilePath?: undefined;
      fileMaxEntries?: undefined;
      lockOptions?: undefined;
@@ -97,70 +150,6 @@ export type ClaimableDedupe = {
  memorySize: () => number;
 };

-const DEFAULT_LOCK_OPTIONS: FileLockOptions = {
-  retries: {
-    retries: 6,
-    factor: 1.35,
-    minTimeout: 8,
-    maxTimeout: 180,
-    randomize: true,
-  },
-  stale: 60_000,
-};
-
-function mergeLockOptions(overrides?: Partial<FileLockOptions>): FileLockOptions {
-  return {
-    stale: overrides?.stale ?? DEFAULT_LOCK_OPTIONS.stale,
-    retries: {
-      retries: overrides?.retries?.retries ?? DEFAULT_LOCK_OPTIONS.retries.retries,
-      factor: overrides?.retries?.factor ?? DEFAULT_LOCK_OPTIONS.retries.factor,
-      minTimeout: overrides?.retries?.minTimeout ?? DEFAULT_LOCK_OPTIONS.retries.minTimeout,
-      maxTimeout: overrides?.retries?.maxTimeout ?? DEFAULT_LOCK_OPTIONS.retries.maxTimeout,
-      randomize: overrides?.retries?.randomize ?? DEFAULT_LOCK_OPTIONS.retries.randomize,
-    },
-  };
-}
-
-function sanitizeData(value: unknown): PersistentDedupeData {
-  if (!value || typeof value !== "object") {
-    return {};
-  }
-  const out: PersistentDedupeData = {};
-  for (const [key, ts] of Object.entries(value as Record<string, unknown>)) {
-    if (typeof ts === "number" && Number.isFinite(ts) && ts > 0) {
-      out[key] = ts;
-    }
-  }
-  return out;
-}
-
-function pruneData(
-  data: PersistentDedupeData,
-  now: number,
-  ttlMs: number,
-  maxEntries: number,
-): void {
-  if (ttlMs > 0) {
-    for (const [key, ts] of Object.entries(data)) {
-      if (now - ts >= ttlMs) {
-        delete data[key];
-      }
-    }
-  }
-
-  const keys = Object.keys(data);
-  if (keys.length <= maxEntries) {
-    return;
-  }
-
-  keys
-    .toSorted((a, b) => data[a] - data[b])
-    .slice(0, keys.length - maxEntries)
-    .forEach((key) => {
-      delete data[key];
-    });
-}
-
 function resolveNamespace(namespace?: string): string {
  return namespace?.trim() || "global";
 }
@@ -173,43 +162,195 @@ function isRecentTimestamp(seenAt: number | undefined, ttlMs: number, now: numbe
  return seenAt != null && (ttlMs <= 0 || now - seenAt < ttlMs);
 }

-/** Create a dedupe helper that combines in-memory fast checks with a lock-protected disk store. */
+function resolveEntrySeenAt(entry: PersistentDedupeEntry | undefined): number | undefined {
+  return typeof entry?.seenAt === "number" && Number.isFinite(entry.seenAt)
+    ? entry.seenAt
+    : undefined;
+}
+
+function shortHash(value: string): string {
+  return createHash("sha256").update(value).digest("hex").slice(0, 32);
+}
+
+function resolveEntryKey(key: string): string {
+  return `k.${shortHash(key)}`;
+}
+
+function resolveRemainingTtlMs(
+  seenAt: number,
+  ttlMs: number,
+  now: number,
+): { ttlMs: number } | undefined | null {
+  if (ttlMs <= 0) {
+    return undefined;
+  }
+  const remaining = ttlMs - (now - seenAt);
+  return remaining > 0 ? { ttlMs: Math.max(1, Math.floor(remaining)) } : null;
+}
+
+function normalizeNamespacePrefix(value: string | undefined): string {
+  const normalized = (value ?? DEFAULT_NAMESPACE_PREFIX)
+    .trim()
+    .toLowerCase()
+    .replace(/[^a-z0-9._-]+/g, "-")
+    .replace(/^[._-]+|[._-]+$/g, "")
+    .slice(0, 48);
+  return normalized || DEFAULT_NAMESPACE_PREFIX;
+}
+
+function resolveStateNamespace(prefix: string, namespace: string): string {
+  return `${prefix}.${shortHash(namespace)}`;
+}
+
+function hasPluginStateOptions(
+  options: ClaimableDedupeOptions | PersistentDedupeOptions,
+): options is PersistentDedupePluginStateOptions {
+  return typeof options.pluginId === "string";
+}
+
+function hasLegacyPathOptions(
+  options: ClaimableDedupeOptions | PersistentDedupeOptions,
+): options is PersistentDedupeLegacyPathOptions {
+  return typeof options.resolveFilePath === "function";
+}
+
+function resolveStateMaxEntries(options: PersistentDedupeOptions): number {
+  const maxEntries = hasPluginStateOptions(options)
+    ? options.stateMaxEntries
+    : options.fileMaxEntries;
+  return Math.max(1, resolveNonNegativeIntegerOption(maxEntries, 1));
+}
+
+function resolvePersistentStoreCacheKey(pluginId: string, namespace: string): string {
+  return `${pluginId}\0${namespace}`;
+}
+
+function createPersistentStoreResolver(
+  options: PersistentDedupeOptions,
+): (namespace: string) => PluginStateSyncKeyedStore<PersistentDedupeEntry> {
+  const maxEntries = resolveStateMaxEntries(options);
+  const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0);
+  const defaultTtlMs = ttlMs > 0 ? ttlMs : undefined;
+  const stores = new Map<string, PluginStateSyncKeyedStore<PersistentDedupeEntry>>();
+
+  if (hasPluginStateOptions(options)) {
+    const pluginId = options.pluginId;
+    const prefix = normalizeNamespacePrefix(options.namespacePrefix);
+    return (namespace) => {
+      const stateNamespace = resolveStateNamespace(prefix, namespace);
+      const cacheKey = resolvePersistentStoreCacheKey(pluginId, stateNamespace);
+      const existing = stores.get(cacheKey);
+      if (existing) {
+        return existing;
+      }
+      const store = createPluginStateSyncKeyedStore<PersistentDedupeEntry>(pluginId, {
+        namespace: stateNamespace,
+        maxEntries,
+        ...(defaultTtlMs != null ? { defaultTtlMs } : {}),
+        ...(options.env ? { env: options.env } : {}),
+      });
+      stores.set(cacheKey, store);
+      return store;
+    };
+  }
+
+  const prefix = normalizeNamespacePrefix("legacy-path");
+  return (namespace) => {
+    const legacyPath = options.resolveFilePath(namespace);
+    const stateNamespace = resolveStateNamespace(prefix, legacyPath);
+    const cacheKey = resolvePersistentStoreCacheKey(LEGACY_PATH_OWNER_ID, stateNamespace);
+    const existing = stores.get(cacheKey);
+    if (existing) {
+      return existing;
+    }
+    const store = createCorePluginStateSyncKeyedStore<PersistentDedupeEntry>({
+      ownerId: LEGACY_PATH_OWNER_ID,
+      namespace: stateNamespace,
+      maxEntries,
+      ...(defaultTtlMs != null ? { defaultTtlMs } : {}),
+      ...(options.env ? { env: options.env } : {}),
+    });
+    stores.set(cacheKey, store);
+    return store;
+  };
+}
+
+function parseLegacyDedupeData(raw: string): {
+  data: Record<string, number>;
+  invalidCount: number;
+} {
+  const parsed = JSON.parse(raw) as unknown;
+  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
+    return { data: {}, invalidCount: 0 };
+  }
+  const data: Record<string, number> = {};
+  let invalidCount = 0;
+  for (const [key, seenAt] of Object.entries(parsed)) {
+    if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) {
+      data[key] = seenAt;
+    } else {
+      invalidCount++;
+    }
+  }
+  return { data, invalidCount };
+}
+
+/** Import one retired JSON dedupe cache file into plugin-state SQLite during doctor repair. */
+export async function migratePersistentDedupeLegacyJsonFile(
+  options: PersistentDedupeLegacyJsonMigrationOptions,
+): Promise<PersistentDedupeLegacyJsonMigrationResult> {
+  const raw = await fs.readFile(options.filePath, "utf8");
+  const { data, invalidCount } = parseLegacyDedupeData(raw);
+  const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0);
+  const now = options.now ?? Date.now();
+  const store = createPersistentStoreResolver(options)(resolveNamespace(options.namespace));
+  const result: PersistentDedupeLegacyJsonMigrationResult = {
+    imported: 0,
+    skippedExpired: 0,
+    skippedInvalid: 0,
+    skippedExisting: 0,
+    removed: false,
+  };
+
+  for (const [key, seenAt] of Object.entries(data)) {
+    const ttlOption = resolveRemainingTtlMs(seenAt, ttlMs, now);
+    if (ttlOption === null) {
+      result.skippedExpired++;
+      continue;
+    }
+    const changed = store.update?.(
+      resolveEntryKey(key),
+      (current) => {
+        const currentSeenAt = resolveEntrySeenAt(current);
+        if (currentSeenAt != null && currentSeenAt >= seenAt) {
+          return undefined;
+        }
+        return { key, seenAt };
+      },
+      ttlOption,
+    );
+    if (changed) {
+      result.imported++;
+    } else {
+      result.skippedExisting++;
+    }
+  }
+
+  result.skippedInvalid = invalidCount;
+  if (options.removeFile !== false) {
+    await fs.rm(options.filePath, { force: true });
+    result.removed = true;
+  }
+  return result;
+}
+
+/** Create a dedupe helper that combines in-memory fast checks with SQLite-backed state. */
 export function createPersistentDedupe(options: PersistentDedupeOptions): PersistentDedupe {
  const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0);
  const memoryMaxSize = resolveNonNegativeIntegerOption(options.memoryMaxSize, 0);
-  const fileMaxEntries = Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1));
-  const lockOptions = mergeLockOptions(options.lockOptions);
+  const getStore = createPersistentStoreResolver(options);
  const memory = createDedupeCache({ ttlMs, maxSize: memoryMaxSize });
  const inflight = new Map<string, Promise<boolean>>();
-  // In-process write queue per file path. `withFileLock` is re-entrant
-  // within the same process (a second caller for the same path gets
-  // immediate access instead of waiting), so two concurrent
-  // checkAndRecordInner calls for different keys but the same file can
-  // race: both read the same stale data, and the last writer's
-  // writeJsonFileAtomically silently overwrites the first writer's
-  // additions. This queue serializes all read-modify-write cycles
-  // targeting the same file within this process, preventing the lost
-  // update while still allowing cross-process file-lock contention to
-  // be handled by the file lock itself.
-  const fileWriteQueues = new Map<string, Promise<unknown>>();
-
-  function enqueueFileWrite<T>(filePath: string, fn: () => Promise<T>): Promise<T> {
-    const prev = fileWriteQueues.get(filePath) ?? Promise.resolve();
-    const next = prev.then(fn, fn);
-    fileWriteQueues.set(filePath, next);
-    // Cleanup: remove the queue entry once this link settles, but only if
-    // no newer work was chained after us. The `.catch(() => {})` prevents
-    // an unhandled rejection when `next` rejects — callers still observe
-    // the rejection through the returned `next` promise directly.
-    next
-      .finally(() => {
-        if (fileWriteQueues.get(filePath) === next) {
-          fileWriteQueues.delete(filePath);
-        }
-      })
-      .catch(() => {});
-    return next;
-  }

  async function checkAndRecordInner(
    key: string,
@@ -222,24 +363,28 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis
      return false;
    }

-    const path = options.resolveFilePath(namespace);
    try {
-      const duplicate = await enqueueFileWrite(path, () =>
-        withFileLock(path, lockOptions, async () => {
-          const { value } = await readJsonFileWithFallback<PersistentDedupeData>(path, {});
-          const data = sanitizeData(value);
-          const seenAt = data[key];
-          const isRecent = seenAt != null && (ttlMs <= 0 || now - seenAt < ttlMs);
-          if (isRecent) {
-            return true;
+      const entryKey = resolveEntryKey(key);
+      const store = getStore(namespace);
+      let duplicateSeenAt: number | undefined;
+      store.update?.(
+        entryKey,
+        (entry) => {
+          const seenAt = resolveEntrySeenAt(entry);
+          if (isRecentTimestamp(seenAt, ttlMs, now)) {
+            duplicateSeenAt = seenAt;
+            return undefined;
          }
-          data[key] = now;
-          pruneData(data, now, ttlMs, fileMaxEntries);
-          await writeJsonFileAtomically(path, data);
-          return false;
-        }),
+          return { key, seenAt: now };
+        },
+        ttlMs > 0 ? { ttlMs } : undefined,
      );
-      return !duplicate;
+      if (duplicateSeenAt != null) {
+        memory.check(scopedKey, duplicateSeenAt);
+        return false;
+      }
+      memory.check(scopedKey, now);
+      return true;
    } catch (error) {
      onDiskError?.(error);
      memory.check(scopedKey, now);
@@ -258,11 +403,8 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis
      return true;
    }

-    const path = options.resolveFilePath(namespace);
    try {
-      const { value } = await readJsonFileWithFallback<PersistentDedupeData>(path, {});
-      const data = sanitizeData(value);
-      const seenAt = data[key];
+      const seenAt = resolveEntrySeenAt(getStore(namespace).lookup(resolveEntryKey(key)));
      if (!isRecentTimestamp(seenAt, ttlMs, now)) {
        return false;
      }
@@ -275,17 +417,18 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis
  }

  async function warmup(namespace = "global", onError?: (error: unknown) => void): Promise<number> {
-    const filePath = options.resolveFilePath(namespace);
    const now = Date.now();
    try {
-      const { value } = await readJsonFileWithFallback<PersistentDedupeData>(filePath, {});
-      const data = sanitizeData(value);
      let loaded = 0;
-      for (const [key, ts] of Object.entries(data)) {
+      for (const entry of getStore(resolveNamespace(namespace)).entries()) {
+        const ts = resolveEntrySeenAt(entry.value);
+        if (ts == null) {
+          continue;
+        }
        if (ttlMs > 0 && now - ts >= ttlMs) {
          continue;
        }
-        const scopedKey = `${namespace}:${key}`;
+        const scopedKey = `${resolveNamespace(namespace)}:${entry.value.key}`;
        memory.check(scopedKey, ts);
        loaded++;
      }
@@ -354,17 +497,28 @@ export function createClaimableDedupe(options: ClaimableDedupeOptions): Claimabl
  const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0);
  const memoryMaxSize = resolveNonNegativeIntegerOption(options.memoryMaxSize, 0);
  const memory = createDedupeCache({ ttlMs, maxSize: memoryMaxSize });
-  const persistent =
-    options.resolveFilePath != null
-      ? createPersistentDedupe({
-          ttlMs,
-          memoryMaxSize,
-          fileMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1)),
-          resolveFilePath: options.resolveFilePath,
-          lockOptions: options.lockOptions,
-          onDiskError: options.onDiskError,
-        })
-      : null;
+  let persistent: PersistentDedupe | null = null;
+  if (hasPluginStateOptions(options)) {
+    persistent = createPersistentDedupe({
+      ttlMs,
+      memoryMaxSize,
+      pluginId: options.pluginId,
+      stateMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.stateMaxEntries, 1)),
+      ...(options.namespacePrefix ? { namespacePrefix: options.namespacePrefix } : {}),
+      ...(options.env ? { env: options.env } : {}),
+      ...(options.onDiskError ? { onDiskError: options.onDiskError } : {}),
+    });
+  } else if (hasLegacyPathOptions(options)) {
+    persistent = createPersistentDedupe({
+      ttlMs,
+      memoryMaxSize,
+      fileMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1)),
+      resolveFilePath: options.resolveFilePath,
+      ...(options.env ? { env: options.env } : {}),
+      ...(options.lockOptions ? { lockOptions: options.lockOptions } : {}),
+      ...(options.onDiskError ? { onDiskError: options.onDiskError } : {}),
+    });
+  }

  const inflight = new Map<
    string,