fix: store device-pair notify state in sqlite

Move Device Pair notify subscribers and delivery dedupe state into SQLite-backed plugin state. Add doctor migration for legacy notify subscribers; request-id delivery state is cache-only and rebuilt.

extensions/device-pair/doctor-contract-api.test.ts

@@ -0,0 +1,120 @@
+// Device Pair tests cover doctor migration of legacy notify state.
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import {
+  createPluginStateKeyedStoreForTests,
+  resetPluginStateStoreForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
+import type {
+  OpenKeyedStoreOptions,
+  PluginDoctorStateMigrationContext,
+} from "openclaw/plugin-sdk/runtime-doctor";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { stateMigrations } from "./doctor-contract-api.js";
+import {
+  DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+  notifySubscriberStoreKey,
+  type NotifySubscription,
+} from "./notify-state.js";
+
+function createDoctorContext(env: NodeJS.ProcessEnv): PluginDoctorStateMigrationContext {
+  return {
+    openPluginStateKeyedStore<T>(options: OpenKeyedStoreOptions) {
+      return createPluginStateKeyedStoreForTests<T>("device-pair", {
+        ...options,
+        env: options.env ?? env,
+      });
+    },
+  };
+}
+
+describe("device-pair doctor notify migration", () => {
+  let stateDir = "";
+  let env: NodeJS.ProcessEnv;
+
+  beforeEach(async () => {
+    resetPluginStateStoreForTests();
+    stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-device-pair-doctor-"));
+    env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
+  });
+
+  afterEach(async () => {
+    await fs.rm(stateDir, { recursive: true, force: true });
+  });
+
+  function migrationParams() {
+    return {
+      config: {},
+      env,
+      stateDir,
+      oauthDir: path.join(stateDir, "oauth"),
+      context: createDoctorContext(env),
+    };
+  }
+
+  it("imports legacy notify subscribers into plugin state", async () => {
+    const sourcePath = path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE);
+    const subscriber: NotifySubscription = {
+      to: "chat-123",
+      accountId: "telegram-default",
+      messageThreadId: 271,
+      mode: "persistent",
+      addedAtMs: 1,
+    };
+    await fs.writeFile(
+      sourcePath,
+      JSON.stringify({
+        subscribers: [subscriber],
+        notifiedRequestIds: { stale: Date.now() },
+      }),
+      "utf8",
+    );
+
+    const migration = stateMigrations[0];
+    await expect(migration.detectLegacyState(migrationParams())).resolves.toMatchObject({
+      preview: [expect.stringContaining("Device Pair notify subscribers")],
+    });
+
+    const result = await migration.migrateLegacyState(migrationParams());
+
+    expect(result.warnings).toEqual([]);
+    expect(result.changes).toEqual([
+      "Migrated Device Pair notify subscribers -> plugin state (1 imported, 0 already present)",
+      expect.stringContaining("Archived Device Pair notify-state legacy source"),
+    ]);
+    await expect(fs.access(sourcePath)).rejects.toThrow();
+    await expect(fs.access(`${sourcePath}.migrated`)).resolves.toBeUndefined();
+    await expect(
+      createDoctorContext(env)
+        .openPluginStateKeyedStore<NotifySubscription>({
+          namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+          maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+        })
+        .lookup(notifySubscriberStoreKey(subscriber)),
+    ).resolves.toEqual(subscriber);
+  });
+
+  it("ignores legacy notify files that only contain cache state", async () => {
+    const sourcePath = path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE);
+    await fs.writeFile(
+      sourcePath,
+      JSON.stringify({
+        subscribers: [],
+        notifiedRequestIds: { cached: Date.now() },
+      }),
+      "utf8",
+    );
+
+    const migration = stateMigrations[0];
+
+    await expect(migration.detectLegacyState(migrationParams())).resolves.toBeNull();
+    await expect(migration.migrateLegacyState(migrationParams())).resolves.toEqual({
+      changes: [],
+      warnings: [],
+    });
+    await expect(fs.access(sourcePath)).resolves.toBeUndefined();
+  });
+});

extensions/device-pair/doctor-contract-api.ts

@@ -0,0 +1,106 @@
+// Device Pair doctor contract migrates shipped plugin-owned state.
+import fs from "node:fs/promises";
+import path from "node:path";
+import type { PluginDoctorStateMigration } from "openclaw/plugin-sdk/runtime-doctor";
+import {
+  DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+  normalizeLegacyNotifyState,
+  notifySubscriberStoreKey,
+  type LegacyNotifyStateFile,
+  type NotifySubscription,
+} from "./notify-state.js";
+
+function resolveLegacyNotifyStatePath(stateDir: string): string {
+  return path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE);
+}
+
+async function fileExists(filePath: string): Promise<boolean> {
+  try {
+    const stat = await fs.stat(filePath);
+    return stat.isFile();
+  } catch {
+    return false;
+  }
+}
+
+async function readLegacyNotifyState(filePath: string): Promise<LegacyNotifyStateFile | null> {
+  try {
+    return normalizeLegacyNotifyState(JSON.parse(await fs.readFile(filePath, "utf8")) as unknown);
+  } catch {
+    return null;
+  }
+}
+
+async function archiveLegacySource(params: {
+  filePath: string;
+  changes: string[];
+  warnings: string[];
+}): Promise<void> {
+  const archivedPath = `${params.filePath}.migrated`;
+  if (await fileExists(archivedPath)) {
+    params.warnings.push(
+      `Left migrated Device Pair notify-state source in place because ${archivedPath} already exists`,
+    );
+    return;
+  }
+  try {
+    await fs.rename(params.filePath, archivedPath);
+    params.changes.push(`Archived Device Pair notify-state legacy source -> ${archivedPath}`);
+  } catch (err) {
+    params.warnings.push(`Failed archiving Device Pair notify-state legacy source: ${String(err)}`);
+  }
+}
+
+export const stateMigrations: PluginDoctorStateMigration[] = [
+  {
+    id: "device-pair-notify-json-to-plugin-state",
+    label: "Device Pair notify subscribers",
+    async detectLegacyState(params) {
+      const filePath = resolveLegacyNotifyStatePath(params.stateDir);
+      const state = await readLegacyNotifyState(filePath);
+      if (!state || state.subscribers.length === 0) {
+        return null;
+      }
+      return {
+        preview: [
+          `- Device Pair notify subscribers: ${filePath} -> plugin state (${DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE}, ${state.subscribers.length} subscriber(s))`,
+        ],
+      };
+    },
+    async migrateLegacyState(params) {
+      const changes: string[] = [];
+      const warnings: string[] = [];
+      const filePath = resolveLegacyNotifyStatePath(params.stateDir);
+      const state = await readLegacyNotifyState(filePath);
+      if (!state || state.subscribers.length === 0) {
+        return { changes, warnings };
+      }
+
+      const store = params.context.openPluginStateKeyedStore<NotifySubscription>({
+        namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+        maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+      });
+      let imported = 0;
+      let alreadyPresent = 0;
+      for (const subscriber of state.subscribers) {
+        const inserted = await store.registerIfAbsent(
+          notifySubscriberStoreKey(subscriber),
+          subscriber,
+        );
+        if (inserted) {
+          imported++;
+        } else {
+          alreadyPresent++;
+        }
+      }
+
+      changes.push(
+        `Migrated Device Pair notify subscribers -> plugin state (${imported} imported, ${alreadyPresent} already present)`,
+      );
+      await archiveLegacySource({ filePath, changes, warnings });
+      return { changes, warnings };
+    },
+  },
+];

extensions/device-pair/notify-state.ts

@@ -0,0 +1,131 @@
+// Device Pair notify state helpers keep runtime and doctor migration in sync.
+import { createHash } from "node:crypto";
+import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
+
+export const DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE = "device-pair-notify.json";
+export const DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE = "notify-subscribers";
+export const DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE = "notify-seen-requests";
+export const DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES = 1024;
+export const DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES = 4096;
+export const DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS = 24 * 60 * 60 * 1000;
+
+export type NotifySubscription = {
+  to: string;
+  accountId?: string;
+  messageThreadId?: string | number;
+  mode: "persistent" | "once";
+  addedAtMs: number;
+};
+
+export type NotifySeenRequest = {
+  requestId: string;
+  notifiedAtMs: number;
+};
+
+export type LegacyNotifyStateFile = {
+  subscribers: NotifySubscription[];
+  notifiedRequestIds: Record<string, number>;
+};
+
+export function normalizeLegacyNotifyState(raw: unknown): LegacyNotifyStateFile {
+  const root = typeof raw === "object" && raw !== null ? (raw as Record<string, unknown>) : {};
+  const subscribersRaw = Array.isArray(root.subscribers) ? root.subscribers : [];
+  const notifiedRaw =
+    typeof root.notifiedRequestIds === "object" && root.notifiedRequestIds !== null
+      ? (root.notifiedRequestIds as Record<string, unknown>)
+      : {};
+
+  const subscribers: NotifySubscription[] = [];
+  for (const item of subscribersRaw) {
+    if (typeof item !== "object" || item === null) {
+      continue;
+    }
+    const record = item as Record<string, unknown>;
+    const to = normalizeOptionalString(record.to) ?? "";
+    if (!to) {
+      continue;
+    }
+    const accountId = normalizeOptionalString(record.accountId) ?? undefined;
+    const messageThreadId =
+      typeof record.messageThreadId === "string"
+        ? normalizeOptionalString(record.messageThreadId) || undefined
+        : typeof record.messageThreadId === "number" && Number.isFinite(record.messageThreadId)
+          ? Math.trunc(record.messageThreadId)
+          : undefined;
+    const mode = record.mode === "once" ? "once" : "persistent";
+    const addedAtMs =
+      typeof record.addedAtMs === "number" && Number.isFinite(record.addedAtMs)
+        ? Math.trunc(record.addedAtMs)
+        : Date.now();
+    subscribers.push({
+      to,
+      accountId,
+      messageThreadId,
+      mode,
+      addedAtMs,
+    });
+  }
+
+  const notifiedRequestIds: Record<string, number> = {};
+  for (const [requestId, ts] of Object.entries(notifiedRaw)) {
+    const normalizedRequestId = normalizeOptionalString(requestId);
+    if (!normalizedRequestId) {
+      continue;
+    }
+    if (typeof ts !== "number" || !Number.isFinite(ts) || ts <= 0) {
+      continue;
+    }
+    notifiedRequestIds[normalizedRequestId] = Math.trunc(ts);
+  }
+
+  return { subscribers, notifiedRequestIds };
+}
+
+export function normalizeNotifyThreadKey(messageThreadId?: string | number): string {
+  if (typeof messageThreadId === "number" && Number.isFinite(messageThreadId)) {
+    return String(Math.trunc(messageThreadId));
+  }
+  if (typeof messageThreadId !== "string") {
+    return "";
+  }
+  const normalized = normalizeOptionalString(messageThreadId);
+  if (!normalized) {
+    return "";
+  }
+  if (!/^-?\d+$/u.test(normalized)) {
+    return normalized;
+  }
+  try {
+    return BigInt(normalized).toString();
+  } catch {
+    return normalized;
+  }
+}
+
+export function notifySubscriberKey(subscriber: {
+  to: string;
+  accountId?: string;
+  messageThreadId?: string | number;
+}): string {
+  return JSON.stringify([
+    subscriber.to,
+    subscriber.accountId ?? "",
+    normalizeNotifyThreadKey(subscriber.messageThreadId),
+  ]);
+}
+
+function hashStoreKey(value: string): string {
+  return createHash("sha256").update(value).digest("hex");
+}
+
+export function notifySubscriberStoreKey(subscriber: {
+  to: string;
+  accountId?: string;
+  messageThreadId?: string | number;
+}): string {
+  return hashStoreKey(notifySubscriberKey(subscriber));
+}
+
+export function notifyRequestStoreKey(requestId: string): string {
+  return hashStoreKey(requestId);
+}

extensions/device-pair/notify.test.ts

@@ -2,8 +2,19 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
+import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
+import {
+  createPluginStateKeyedStoreForTests,
+  resetPluginStateStoreForTests,
+} from "openclaw/plugin-sdk/plugin-state-test-runtime";
 import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api";
 import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+  notifySubscriberStoreKey,
+  type NotifySubscription,
+} from "./notify-state.js";
 
 const listDevicePairingMock = vi.hoisted(() => vi.fn(async () => ({ pending: [] })));
 
@@ -20,46 +31,55 @@ afterAll(() => {
 
 describe("device-pair notify persistence", () => {
   let stateDir: string;
+  let env: NodeJS.ProcessEnv;
 
   beforeEach(async () => {
+    resetPluginStateStoreForTests();
     vi.clearAllMocks();
     listDevicePairingMock.mockResolvedValue({ pending: [] });
     stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "device-pair-notify-"));
+    env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
   });
 
   afterEach(async () => {
     await fs.rm(stateDir, { recursive: true, force: true });
   });
 
-  it("matches persisted telegram thread ids across number and string roundtrips", async () => {
-    await fs.writeFile(
-      path.join(stateDir, "device-pair-notify.json"),
-      JSON.stringify(
-        {
-          subscribers: [
-            {
-              to: "chat-123",
-              accountId: "telegram-default",
-              messageThreadId: 271,
-              mode: "persistent",
-              addedAtMs: 1,
-            },
-          ],
-          notifiedRequestIds: {},
-        },
-        null,
-        2,
-      ),
-      "utf8",
-    );
+  function openStore<T>(options: OpenKeyedStoreOptions) {
+    return createPluginStateKeyedStoreForTests<T>("device-pair", {
+      ...options,
+      env: options.env ?? env,
+    });
+  }
 
-    const api = createTestPluginApi({
+  function createApi() {
+    return createTestPluginApi({
       runtime: {
         state: {
           resolveStateDir: () => stateDir,
+          openKeyedStore: openStore,
         },
       } as never,
     });
+  }
+
+  function openSubscriberStore() {
+    return openStore<NotifySubscription>({
+      namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+      maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+    });
+  }
+
+  it("matches persisted telegram thread ids across number and string roundtrips", async () => {
+    const subscriber: NotifySubscription = {
+      to: "chat-123",
+      accountId: "telegram-default",
+      messageThreadId: 271,
+      mode: "persistent",
+      addedAtMs: 1,
+    };
+    await openSubscriberStore().register(notifySubscriberStoreKey(subscriber), subscriber);
+    const api = createApi();
 
     const status = await handleNotifyCommand({
       api,
@@ -86,46 +106,26 @@ describe("device-pair notify persistence", () => {
       action: "off",
     });
 
-    const persisted = JSON.parse(
-      await fs.readFile(path.join(stateDir, "device-pair-notify.json"), "utf8"),
-    ) as { subscribers: unknown[] };
-    expect(persisted.subscribers).toStrictEqual([]);
+    await expect(openSubscriberStore().entries()).resolves.toStrictEqual([]);
   });
 
   it("does not remove a different persisted subscriber when notify fields contain pipes", async () => {
-    await fs.writeFile(
-      path.join(stateDir, "device-pair-notify.json"),
-      JSON.stringify(
-        {
-          subscribers: [
-            {
-              to: "chat|123",
-              accountId: "acct",
-              mode: "persistent",
-              addedAtMs: 1,
-            },
-            {
-              to: "chat",
-              accountId: "123|acct",
-              mode: "persistent",
-              addedAtMs: 2,
-            },
-          ],
-          notifiedRequestIds: {},
-        },
-        null,
-        2,
-      ),
-      "utf8",
-    );
-
-    const api = createTestPluginApi({
-      runtime: {
-        state: {
-          resolveStateDir: () => stateDir,
-        },
-      } as never,
-    });
+    const firstSubscriber: NotifySubscription = {
+      to: "chat|123",
+      accountId: "acct",
+      mode: "persistent",
+      addedAtMs: 1,
+    };
+    const secondSubscriber: NotifySubscription = {
+      to: "chat",
+      accountId: "123|acct",
+      mode: "persistent",
+      addedAtMs: 2,
+    };
+    const store = openSubscriberStore();
+    await store.register(notifySubscriberStoreKey(firstSubscriber), firstSubscriber);
+    await store.register(notifySubscriberStoreKey(secondSubscriber), secondSubscriber);
+    const api = createApi();
 
     await handleNotifyCommand({
       api,
@@ -148,19 +148,11 @@ describe("device-pair notify persistence", () => {
     });
     expect(status.text).toContain("Pair request notifications: disabled for this chat.");
 
-    const persisted = JSON.parse(
-      await fs.readFile(path.join(stateDir, "device-pair-notify.json"), "utf8"),
-    ) as unknown;
-    expect(persisted).toStrictEqual({
-      subscribers: [
-        {
-          to: "chat|123",
-          accountId: "acct",
-          mode: "persistent",
-          addedAtMs: 1,
-        },
-      ],
-      notifiedRequestIds: {},
-    });
+    await expect(openSubscriberStore().entries()).resolves.toMatchObject([
+      {
+        key: notifySubscriberStoreKey(firstSubscriber),
+        value: firstSubscriber,
+      },
+    ]);
   });
 });

extensions/device-pair/notify.ts

@@ -1,24 +1,24 @@
 // Device Pair plugin module implements notify behavior.
-import { promises as fs } from "node:fs";
-import path from "node:path";
 import type { OpenClawPluginService } from "openclaw/plugin-sdk/core";
 import { listDevicePairing } from "openclaw/plugin-sdk/device-bootstrap";
 import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
 import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry";
-import { replaceFileAtomic } from "openclaw/plugin-sdk/security-runtime";
+import type { PluginStateKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
 import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime";
+import {
+  DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS,
+  DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES,
+  DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+  DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+  notifyRequestStoreKey,
+  notifySubscriberKey,
+  notifySubscriberStoreKey,
+  type NotifySeenRequest,
+  type NotifySubscription,
+} from "./notify-state.js";
 
-const NOTIFY_STATE_FILE = "device-pair-notify.json";
 const NOTIFY_POLL_INTERVAL_MS = 10_000;
-const NOTIFY_MAX_SEEN_AGE_MS = 24 * 60 * 60 * 1000;
-
-type NotifySubscription = {
-  to: string;
-  accountId?: string;
-  messageThreadId?: string | number;
-  mode: "persistent" | "once";
-  addedAtMs: number;
-};
 
 type NotifyStateFile = {
   subscribers: NotifySubscription[];
@@ -79,112 +79,79 @@ export function formatPendingRequests(pending: PendingPairingRequest[]): string
   return lines.join("\n");
 }
 
-function resolveNotifyStatePath(stateDir: string): string {
-  return path.join(stateDir, NOTIFY_STATE_FILE);
+function openNotifySubscriberStore(
+  api: OpenClawPluginApi,
+): PluginStateKeyedStore<NotifySubscription> {
+  return api.runtime.state.openKeyedStore<NotifySubscription>({
+    namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE,
+    maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES,
+  });
 }
 
-function normalizeNotifyState(raw: unknown): NotifyStateFile {
-  const root = typeof raw === "object" && raw !== null ? (raw as Record<string, unknown>) : {};
-  const subscribersRaw = Array.isArray(root.subscribers) ? root.subscribers : [];
-  const notifiedRaw =
-    typeof root.notifiedRequestIds === "object" && root.notifiedRequestIds !== null
-      ? (root.notifiedRequestIds as Record<string, unknown>)
-      : {};
+function openNotifySeenRequestStore(
+  api: OpenClawPluginApi,
+): PluginStateKeyedStore<NotifySeenRequest> {
+  return api.runtime.state.openKeyedStore<NotifySeenRequest>({
+    namespace: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE,
+    maxEntries: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES,
+    defaultTtlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS,
+  });
+}
 
-  const subscribers: NotifySubscription[] = [];
-  for (const item of subscribersRaw) {
-    if (typeof item !== "object" || item === null) {
-      continue;
-    }
-    const record = item as Record<string, unknown>;
-    const to = normalizeOptionalString(record.to) ?? "";
-    if (!to) {
-      continue;
-    }
-    const accountId = normalizeOptionalString(record.accountId) ?? undefined;
-    const messageThreadId =
-      typeof record.messageThreadId === "string"
-        ? normalizeOptionalString(record.messageThreadId) || undefined
-        : typeof record.messageThreadId === "number" && Number.isFinite(record.messageThreadId)
-          ? Math.trunc(record.messageThreadId)
-          : undefined;
-    const mode = record.mode === "once" ? "once" : "persistent";
-    const addedAtMs =
-      typeof record.addedAtMs === "number" && Number.isFinite(record.addedAtMs)
-        ? Math.trunc(record.addedAtMs)
-        : Date.now();
-    subscribers.push({
-      to,
-      accountId,
-      messageThreadId,
-      mode,
-      addedAtMs,
-    });
-  }
+async function readNotifyState(api: OpenClawPluginApi): Promise<NotifyStateFile> {
+  const subscriberStore = openNotifySubscriberStore(api);
+  const seenRequestStore = openNotifySeenRequestStore(api);
+  const [subscriberEntries, seenRequestEntries] = await Promise.all([
+    subscriberStore.entries(),
+    seenRequestStore.entries(),
+  ]);
 
+  const subscribers = subscriberEntries
+    .map((entry) => entry.value)
+    .sort((a, b) => a.addedAtMs - b.addedAtMs);
   const notifiedRequestIds: Record<string, number> = {};
-  for (const [requestId, ts] of Object.entries(notifiedRaw)) {
-    const normalizedRequestId = normalizeOptionalString(requestId);
-    if (!normalizedRequestId) {
+  for (const entry of seenRequestEntries) {
+    const requestId = normalizeOptionalString(entry.value.requestId);
+    const notifiedAtMs = entry.value.notifiedAtMs;
+    if (!requestId || !Number.isFinite(notifiedAtMs) || notifiedAtMs <= 0) {
       continue;
     }
-    if (typeof ts !== "number" || !Number.isFinite(ts) || ts <= 0) {
-      continue;
-    }
-    notifiedRequestIds[normalizedRequestId] = Math.trunc(ts);
+    notifiedRequestIds[requestId] = Math.trunc(notifiedAtMs);
   }
 
   return { subscribers, notifiedRequestIds };
 }
 
-async function readNotifyState(filePath: string): Promise<NotifyStateFile> {
-  try {
-    const content = await fs.readFile(filePath, "utf8");
-    return normalizeNotifyState(JSON.parse(content));
-  } catch {
-    return { subscribers: [], notifiedRequestIds: {} };
+async function writeNotifyState(api: OpenClawPluginApi, state: NotifyStateFile): Promise<void> {
+  const subscriberStore = openNotifySubscriberStore(api);
+  const nextSubscribers = new Map(
+    state.subscribers.map((subscriber) => [notifySubscriberStoreKey(subscriber), subscriber]),
+  );
+  for (const entry of await subscriberStore.entries()) {
+    if (!nextSubscribers.has(entry.key)) {
+      await subscriberStore.delete(entry.key);
+    }
+  }
+  for (const [key, subscriber] of nextSubscribers) {
+    await subscriberStore.register(key, subscriber);
   }
-}
 
-async function writeNotifyState(filePath: string, state: NotifyStateFile): Promise<void> {
-  const content = JSON.stringify(state, null, 2);
-  await replaceFileAtomic({
-    filePath,
-    content: `${content}\n`,
-    tempPrefix: ".device-pair-notify",
-  });
-}
-
-function notifySubscriberKey(subscriber: {
-  to: string;
-  accountId?: string;
-  messageThreadId?: string | number;
-}): string {
-  return JSON.stringify([
-    subscriber.to,
-    subscriber.accountId ?? "",
-    normalizeNotifyThreadKey(subscriber.messageThreadId),
-  ]);
-}
-
-function normalizeNotifyThreadKey(messageThreadId?: string | number): string {
-  if (typeof messageThreadId === "number" && Number.isFinite(messageThreadId)) {
-    return String(Math.trunc(messageThreadId));
+  const seenRequestStore = openNotifySeenRequestStore(api);
+  const nextSeenRequests = new Map(
+    Object.entries(state.notifiedRequestIds).map(([requestId, notifiedAtMs]) => [
+      notifyRequestStoreKey(requestId),
+      { requestId, notifiedAtMs },
+    ]),
+  );
+  for (const entry of await seenRequestStore.entries()) {
+    if (!nextSeenRequests.has(entry.key)) {
+      await seenRequestStore.delete(entry.key);
+    }
   }
-  if (typeof messageThreadId !== "string") {
-    return "";
-  }
-  const normalized = normalizeOptionalString(messageThreadId);
-  if (!normalized) {
-    return "";
-  }
-  if (!/^-?\d+$/u.test(normalized)) {
-    return normalized;
-  }
-  try {
-    return BigInt(normalized).toString();
-  } catch {
-    return normalized;
+  for (const [key, value] of nextSeenRequests) {
+    await seenRequestStore.register(key, value, {
+      ttlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS,
+    });
   }
 }
 
@@ -317,11 +284,8 @@ async function notifySubscriber(params: {
   }
 }
 
-async function notifyPendingPairingRequests(params: {
-  api: OpenClawPluginApi;
-  statePath: string;
-}): Promise<void> {
-  const state = await readNotifyState(params.statePath);
+async function notifyPendingPairingRequests(params: { api: OpenClawPluginApi }): Promise<void> {
+  const state = await readNotifyState(params.api);
   const pairing = await listDevicePairing();
   const pending = pairing.pending as PendingPairingRequest[];
   const now = Date.now();
@@ -329,7 +293,7 @@ async function notifyPendingPairingRequests(params: {
   let changed = false;
 
   for (const [requestId, ts] of Object.entries(state.notifiedRequestIds)) {
-    if (!pendingIds.has(requestId) || now - ts > NOTIFY_MAX_SEEN_AGE_MS) {
+    if (!pendingIds.has(requestId) || now - ts > DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS) {
       delete state.notifiedRequestIds[requestId];
       changed = true;
     }
@@ -376,7 +340,7 @@ async function notifyPendingPairingRequests(params: {
   }
 
   if (changed) {
-    await writeNotifyState(params.statePath, state);
+    await writeNotifyState(params.api, state);
   }
 }
 
@@ -399,9 +363,7 @@ export async function armPairNotifyOnce(params: {
     return false;
   }
 
-  const stateDir = params.api.runtime.state.resolveStateDir();
-  const statePath = resolveNotifyStatePath(stateDir);
-  const state = await readNotifyState(statePath);
+  const state = await readNotifyState(params.api);
   let changed = false;
 
   if (upsertNotifySubscriber(state.subscribers, target, "once")) {
@@ -409,7 +371,7 @@ export async function armPairNotifyOnce(params: {
   }
 
   if (changed) {
-    await writeNotifyState(statePath, state);
+    await writeNotifyState(params.api, state);
   }
   return true;
 }
@@ -435,15 +397,13 @@ export async function handleNotifyCommand(params: {
     return { text: "Could not resolve Telegram target for this chat." };
   }
 
-  const stateDir = params.api.runtime.state.resolveStateDir();
-  const statePath = resolveNotifyStatePath(stateDir);
-  const state = await readNotifyState(statePath);
+  const state = await readNotifyState(params.api);
   const targetKey = notifySubscriberKey(target);
   const current = state.subscribers.find((entry) => notifySubscriberKey(entry) === targetKey);
 
   if (params.action === "on" || params.action === "enable") {
     if (upsertNotifySubscriber(state.subscribers, target, "persistent")) {
-      await writeNotifyState(statePath, state);
+      await writeNotifyState(params.api, state);
     }
     return {
       text:
@@ -458,7 +418,7 @@ export async function handleNotifyCommand(params: {
     );
     if (currentIndex !== -1) {
       state.subscribers.splice(currentIndex, 1);
-      await writeNotifyState(statePath, state);
+      await writeNotifyState(params.api, state);
     }
     return { text: "✅ Pair request notifications disabled for this Telegram chat." };
   }
@@ -499,10 +459,9 @@ export function createPairingNotifierService(api: OpenClawPluginApi): OpenClawPl
 
   return {
     id: "device-pair-notifier",
-    start: async (ctx) => {
-      const statePath = resolveNotifyStatePath(ctx.stateDir);
+    start: async () => {
       const tick = async () => {
-        await notifyPendingPairingRequests({ api, statePath });
+        await notifyPendingPairingRequests({ api });
       };
 
       await tick().catch((err: unknown) => {