diff --git a/docs/channels/feishu.md b/docs/channels/feishu.md index 0b97818ec6ec..4642dc262ad8 100644 --- a/docs/channels/feishu.md +++ b/docs/channels/feishu.md @@ -15,7 +15,7 @@ Feishu/Lark is an all-in-one collaboration platform where teams chat, share docu ## Quick start -Requires OpenClaw 2026.4.25 or above. Run `openclaw --version` to check. Upgrade with `openclaw update`. +Requires OpenClaw 2026.5.29 or above. Run `openclaw --version` to check. Upgrade with `openclaw update`. diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index 6558560a8b75..f09f8cd7e15c 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -237,6 +237,7 @@ and pairing-path families. | `plugin-sdk/session-store-runtime` | Session workflow helpers (`getSessionEntry`, `listSessionEntries`, `patchSessionEntry`, `upsertSessionEntry`), legacy session store path/session-key helpers, updated-at reads, and deprecated whole-store mutation helpers | | `plugin-sdk/cron-store-runtime` | Cron store path/load/save helpers | | `plugin-sdk/state-paths` | State/OAuth dir path helpers | + | `plugin-sdk/plugin-state-runtime` | Plugin sidecar SQLite keyed-state types | | `plugin-sdk/routing` | Route/session-key/account binding helpers such as `resolveAgentRoute`, `buildAgentSessionKey`, and `resolveDefaultAgentBoundAccountId` | | `plugin-sdk/status-helpers` | Shared channel/account status summary helpers, runtime-state defaults, and issue metadata helpers | | `plugin-sdk/target-resolver-runtime` | Shared target resolver helpers | diff --git a/extensions/discord/legacy-state-migrations-api.ts b/extensions/discord/legacy-state-migrations-api.ts new file mode 100644 index 000000000000..570784700eae --- /dev/null +++ b/extensions/discord/legacy-state-migrations-api.ts @@ -0,0 +1 @@ +export { detectDiscordLegacyStateMigrations } from "./src/monitor/model-picker-preferences-migrations.js"; diff --git a/extensions/discord/package.json b/extensions/discord/package.json index 58649017d5e3..a9539f18c759 100644 --- a/extensions/discord/package.json +++ b/extensions/discord/package.json @@ -32,6 +32,9 @@ "./index.ts" ], "setupEntry": "./setup-entry.ts", + "setupFeatures": { + "legacyStateMigrations": true + }, "channel": { "id": "discord", "label": "Discord", diff --git a/extensions/discord/setup-entry.test.ts b/extensions/discord/setup-entry.test.ts new file mode 100644 index 000000000000..c9bcef525afe --- /dev/null +++ b/extensions/discord/setup-entry.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, it } from "vitest"; +import setupEntry from "./setup-entry.js"; + +describe("discord setup entry", () => { + it("exposes legacy state migration detector through setup entry metadata", () => { + expect(setupEntry.kind).toBe("bundled-channel-setup-entry"); + expect(setupEntry.features).toEqual({ legacyStateMigrations: true }); + expect(setupEntry.loadLegacyStateMigrationDetector?.()).toBeTypeOf("function"); + }); +}); diff --git a/extensions/discord/setup-entry.ts b/extensions/discord/setup-entry.ts index aa5c385f21d1..be9480f4d7d1 100644 --- a/extensions/discord/setup-entry.ts +++ b/extensions/discord/setup-entry.ts @@ -2,8 +2,15 @@ import { defineBundledChannelSetupEntry } from "openclaw/plugin-sdk/channel-entr export default defineBundledChannelSetupEntry({ importMetaUrl: import.meta.url, + features: { + legacyStateMigrations: true, + }, plugin: { specifier: "./setup-plugin-api.js", exportName: "discordSetupPlugin", }, + legacyStateMigrations: { + specifier: "./legacy-state-migrations-api.js", + exportName: "detectDiscordLegacyStateMigrations", + }, }); diff --git a/extensions/discord/src/monitor/model-picker-preferences-migrations.test.ts b/extensions/discord/src/monitor/model-picker-preferences-migrations.test.ts new file mode 100644 index 000000000000..83bd0d2d637c --- /dev/null +++ b/extensions/discord/src/monitor/model-picker-preferences-migrations.test.ts @@ -0,0 +1,66 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { detectDiscordLegacyStateMigrations } from "./model-picker-preferences-migrations.js"; + +const tempDirs: string[] = []; + +async function makeStateDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-discord-model-picker-migration-")); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); +}); + +describe("Discord model picker preference migration", () => { + it("plans legacy JSON import into plugin state", async () => { + const stateDir = await makeStateDir(); + const sourcePath = path.join(stateDir, "discord", "model-picker-preferences.json"); + await fs.mkdir(path.dirname(sourcePath), { recursive: true }); + await fs.writeFile( + sourcePath, + JSON.stringify({ + version: 1, + entries: { + "discord:default:dm:user:123": { + recent: ["OpenAI/gpt-5", "bad", "openai/gpt-5"], + updatedAt: "2026-05-29T00:00:00.000Z", + }, + }, + }), + ); + + const plans = await Promise.resolve( + detectDiscordLegacyStateMigrations({ + 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("discord"); + expect(plan.namespace).toBe("model-picker-preferences"); + const entries = await plan.readEntries(); + expect(entries).toHaveLength(1); + expect(entries[0]?.key).toMatch(/^v1:[0-9a-f]{32}:[0-9a-f]{24}$/u); + expect(entries[0]?.value).toEqual({ + scopeKey: "discord:default:dm:user:123", + modelRef: "openai/gpt-5", + updatedAt: "2026-05-29T00:00:00.001Z", + }); + }); +}); diff --git a/extensions/discord/src/monitor/model-picker-preferences-migrations.ts b/extensions/discord/src/monitor/model-picker-preferences-migrations.ts new file mode 100644 index 000000000000..6db44f01f0d6 --- /dev/null +++ b/extensions/discord/src/monitor/model-picker-preferences-migrations.ts @@ -0,0 +1,145 @@ +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"; +import { normalizeProviderId } from "openclaw/plugin-sdk/provider-model-shared"; + +const PREFERENCE_MAX_ENTRIES = 2_000; +const MAX_PLUGIN_STATE_KEY_BYTES = 512; +const textEncoder = new TextEncoder(); + +type LegacyModelPickerPreferencesEntry = { + recent?: unknown; + updatedAt?: unknown; +}; + +type LegacyModelPickerPreferencesStore = { + entries?: unknown; +}; + +function fileExists(filePath: string): boolean { + try { + return fs.statSync(filePath).isFile(); + } catch { + return false; + } +} + +function readLegacyStore(filePath: string): LegacyModelPickerPreferencesStore | null { + try { + const parsed = JSON.parse(fs.readFileSync(filePath, "utf8")) as unknown; + return parsed && typeof parsed === "object" + ? (parsed as LegacyModelPickerPreferencesStore) + : null; + } catch { + return null; + } +} + +function normalizeLegacyPreferenceKey(key: string): string | undefined { + const trimmed = key.trim(); + if (!trimmed || textEncoder.encode(trimmed).length > MAX_PLUGIN_STATE_KEY_BYTES) { + return undefined; + } + return trimmed; +} + +function normalizeModelRef(raw?: string): string | null { + const value = raw?.trim(); + if (!value) { + return null; + } + const slashIndex = value.indexOf("/"); + if (slashIndex <= 0 || slashIndex >= value.length - 1) { + return null; + } + const provider = normalizeProviderId(value.slice(0, slashIndex)); + const model = value.slice(slashIndex + 1).trim(); + return provider && model ? `${provider}/${model}` : null; +} + +function sanitizeRecentModels(models: unknown, limit: number): string[] { + const deduped: string[] = []; + const seen = new Set(); + if (!Array.isArray(models)) { + return deduped; + } + for (const item of models) { + const normalized = normalizeModelRef(typeof item === "string" ? item : undefined); + if (!normalized || seen.has(normalized)) { + continue; + } + seen.add(normalized); + deduped.push(normalized); + if (deduped.length >= limit) { + break; + } + } + return deduped; +} + +function hashSegment(value: string, length: number): string { + return createHash("sha256").update(value, "utf8").digest("hex").slice(0, length); +} + +function buildPreferenceModelKey(scopeKey: string, modelRef: string): string { + return `v1:${hashSegment(scopeKey, 32)}:${hashSegment(modelRef, 24)}`; +} + +function timestampMs(value: unknown): number { + const parsed = typeof value === "string" ? Date.parse(value) : Number.NaN; + return Number.isFinite(parsed) ? parsed : 0; +} + +function legacyUpdatedAtForIndex(updatedAt: unknown, index: number, total: number): string { + return new Date(timestampMs(updatedAt) + Math.max(0, total - index)).toISOString(); +} + +export const detectDiscordLegacyStateMigrations: BundledChannelLegacyStateMigrationDetector = ({ + stateDir, +}) => { + const sourcePath = path.join(stateDir, "discord", "model-picker-preferences.json"); + if (!fileExists(sourcePath)) { + return []; + } + return [ + { + kind: "plugin-state-import", + label: "Discord model picker preferences", + sourcePath, + targetPath: "plugin state:model-picker-preferences", + pluginId: "discord", + namespace: "model-picker-preferences", + maxEntries: PREFERENCE_MAX_ENTRIES, + scopeKey: "", + cleanupSource: "rename", + readEntries: () => { + const store = readLegacyStore(sourcePath); + if (!store || !store.entries || typeof store.entries !== "object") { + return []; + } + const out: Array<{ key: string; value: unknown }> = []; + for (const [rawKey, rawEntry] of Object.entries( + store.entries as Record, + )) { + const scopeKey = normalizeLegacyPreferenceKey(rawKey); + if (!scopeKey || !rawEntry || typeof rawEntry !== "object") { + continue; + } + const recent = sanitizeRecentModels(rawEntry.recent, 10); + for (const [index, modelRef] of recent.entries()) { + out.push({ + key: buildPreferenceModelKey(scopeKey, modelRef), + value: { + scopeKey, + modelRef, + updatedAt: legacyUpdatedAtForIndex(rawEntry.updatedAt, index, recent.length), + }, + }); + } + } + return out; + }, + }, + ]; +}; diff --git a/extensions/discord/src/monitor/model-picker-preferences.test.ts b/extensions/discord/src/monitor/model-picker-preferences.test.ts index 14703cf571b5..716a83d1a787 100644 --- a/extensions/discord/src/monitor/model-picker-preferences.test.ts +++ b/extensions/discord/src/monitor/model-picker-preferences.test.ts @@ -1,8 +1,15 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; import { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it } from "vitest"; +import { setDiscordRuntime, type DiscordRuntime } from "../runtime.js"; +import { + buildDiscordModelPickerPreferenceKey, readDiscordModelPickerRecentModels, recordDiscordModelPickerRecentModel, } from "./model-picker-preferences.js"; @@ -12,10 +19,21 @@ const tempDirs: string[] = []; async function createStateEnv(): Promise { const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-model-picker-")); tempDirs.push(dir); - return { ...process.env, OPENCLAW_STATE_DIR: dir }; + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + setDiscordRuntime({ + state: { + openKeyedStore: (options: OpenKeyedStoreOptions) => + createPluginStateKeyedStoreForTests("discord", { + ...options, + env: options.env ?? env, + }), + }, + } as unknown as DiscordRuntime); + return env; } afterEach(async () => { + resetPluginStateStoreForTests(); await Promise.all( tempDirs.splice(0).map(async (dir) => { await fs.rm(dir, { recursive: true, force: true }); @@ -51,12 +69,40 @@ describe("discord model picker preferences", () => { expect(recent).toEqual(["openai/gpt-4.1"]); }); - it("falls back to an empty store when the file is corrupt", async () => { + it("prunes older stored models beyond the recent limit", async () => { const env = await createStateEnv(); - const stateDir = env.OPENCLAW_STATE_DIR as string; - const filePath = path.join(stateDir, "discord", "model-picker-preferences.json"); - await fs.mkdir(path.dirname(filePath), { recursive: true }); - await fs.writeFile(filePath, "{not-json", "utf-8"); + const scope = { userId: "limited-user" }; + for (const modelRef of [ + "openai/model-a", + "openai/model-b", + "openai/model-c", + "openai/model-d", + ]) { + await recordDiscordModelPickerRecentModel({ env, scope, modelRef, limit: 2 }); + } + + await expect(readDiscordModelPickerRecentModels({ env, scope, limit: 10 })).resolves.toEqual([ + "openai/model-d", + "openai/model-c", + ]); + const store = createPluginStateKeyedStoreForTests("discord", { + namespace: "model-picker-preferences", + maxEntries: 2_000, + env, + }); + expect(await store.entries()).toHaveLength(2); + }); + + it("falls back to empty recents when stored state is malformed", async () => { + const env = await createStateEnv(); + const key = buildDiscordModelPickerPreferenceKey({ userId: "789" }); + expect(key).toBeTruthy(); + const store = createPluginStateKeyedStoreForTests("discord", { + namespace: "model-picker-preferences", + maxEntries: 2_000, + env, + }); + await store.register(key as string, "not-an-entry"); const recent = await readDiscordModelPickerRecentModels({ env, @@ -64,4 +110,101 @@ describe("discord model picker preferences", () => { }); expect(recent).toStrictEqual([]); }); + + it("treats plugin-state failures as optional preference misses", async () => { + const env = await createStateEnv(); + const scope = { userId: "state-failure-user" }; + setDiscordRuntime({ + state: { + openKeyedStore: () => { + throw new Error("state unavailable"); + }, + }, + } as unknown as DiscordRuntime); + + await expect(readDiscordModelPickerRecentModels({ env, scope })).resolves.toEqual([]); + await expect( + recordDiscordModelPickerRecentModel({ env, scope, modelRef: "openai/gpt-4.1" }), + ).resolves.toBeUndefined(); + }); + + it("imports legacy JSON preferences into plugin state", async () => { + const env = await createStateEnv(); + const scope = { accountId: "main", guildId: "guild-1", userId: "user-1" }; + const key = buildDiscordModelPickerPreferenceKey(scope); + expect(key).toBeTruthy(); + const legacyPath = path.join( + env.OPENCLAW_STATE_DIR as string, + "discord", + "model-picker-preferences.json", + ); + await fs.mkdir(path.dirname(legacyPath), { recursive: true }); + await fs.writeFile( + legacyPath, + JSON.stringify({ + version: 1, + entries: { + [key as string]: { + recent: ["openai/gpt-4.1", "bad-model", "openai/gpt-4o"], + updatedAt: "2026-01-01T00:00:00.000Z", + }, + }, + }), + "utf8", + ); + + const recent = await readDiscordModelPickerRecentModels({ env, scope }); + expect(recent).toEqual(["openai/gpt-4.1", "openai/gpt-4o"]); + + await fs.rm(legacyPath, { force: true }); + expect(await readDiscordModelPickerRecentModels({ env, scope })).toEqual([ + "openai/gpt-4.1", + "openai/gpt-4o", + ]); + }); + + it("skips malformed legacy JSON entries during import", async () => { + const env = await createStateEnv(); + const scope = { userId: "valid-legacy-user" }; + const key = buildDiscordModelPickerPreferenceKey(scope); + expect(key).toBeTruthy(); + const legacyPath = path.join( + env.OPENCLAW_STATE_DIR as string, + "discord", + "model-picker-preferences.json", + ); + await fs.mkdir(path.dirname(legacyPath), { recursive: true }); + await fs.writeFile( + legacyPath, + JSON.stringify({ + version: 1, + entries: { + "": { recent: ["openai/bad-empty"], updatedAt: "bad" }, + ["x".repeat(600)]: { recent: ["openai/bad-long"], updatedAt: "bad" }, + [key as string]: { + recent: ["not-a-model", "openai/gpt-4.1"], + updatedAt: "2026-01-01T00:00:00.000Z", + }, + }, + }), + "utf8", + ); + + await expect(readDiscordModelPickerRecentModels({ env, scope })).resolves.toEqual([ + "openai/gpt-4.1", + ]); + }); + + it("preserves concurrent model picker selections for the same scope", async () => { + const env = await createStateEnv(); + const scope = { userId: "concurrent-user" }; + + await Promise.all([ + recordDiscordModelPickerRecentModel({ env, scope, modelRef: "openai/gpt-4.1" }), + recordDiscordModelPickerRecentModel({ env, scope, modelRef: "openai/gpt-4o" }), + ]); + + const recent = await readDiscordModelPickerRecentModels({ env, scope }); + expect(new Set(recent)).toEqual(new Set(["openai/gpt-4o", "openai/gpt-4.1"])); + }); }); diff --git a/extensions/discord/src/monitor/model-picker-preferences.ts b/extensions/discord/src/monitor/model-picker-preferences.ts index 4d68faf22394..2f1b5c802b34 100644 --- a/extensions/discord/src/monitor/model-picker-preferences.ts +++ b/extensions/discord/src/monitor/model-picker-preferences.ts @@ -1,55 +1,43 @@ +import { createHash } from "node:crypto"; import os from "node:os"; import path from "node:path"; import { normalizeAccountId as normalizeSharedAccountId } from "openclaw/plugin-sdk/account-id"; -import { withFileLock } from "openclaw/plugin-sdk/file-lock"; -import { readJsonFileWithFallback, writeJsonFileAtomically } from "openclaw/plugin-sdk/json-store"; +import { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store"; import { normalizeProviderId } from "openclaw/plugin-sdk/provider-model-shared"; import { resolveStateDir } from "openclaw/plugin-sdk/state-paths"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; - -const MODEL_PICKER_PREFERENCES_LOCK_OPTIONS = { - retries: { - retries: 8, - factor: 2, - minTimeout: 50, - maxTimeout: 5_000, - randomize: true, - }, - stale: 15_000, -} as const; +import { getDiscordRuntime } from "../runtime.js"; const DEFAULT_RECENT_LIMIT = 5; +const PREFERENCE_MAX_ENTRIES = 2_000; +const MAX_PLUGIN_STATE_KEY_BYTES = 512; +const textEncoder = new TextEncoder(); +let lastPreferenceTimestampMs = 0; type ModelPickerPreferencesEntry = { + scopeKey: string; + modelRef: string; + updatedAt: string; +}; + +type LegacyModelPickerPreferencesEntry = { recent: string[]; updatedAt: string; }; -type ModelPickerPreferencesStore = { - version: 1; - entries: Record; +type LegacyModelPickerPreferencesStore = { + version?: unknown; + entries?: unknown; }; -function sanitizePreferenceEntries(entries: unknown): Record { - if (!entries || typeof entries !== "object") { - return {}; - } - const normalizedEntries: Record = {}; - for (const [key, value] of Object.entries(entries)) { - if (!value || typeof value !== "object") { - continue; - } - const typedValue = value as { - recent?: unknown; - updatedAt?: unknown; - }; - const recent = Array.isArray(typedValue.recent) - ? typedValue.recent.filter((item: unknown): item is string => typeof item === "string") - : []; - const updatedAt = typeof typedValue.updatedAt === "string" ? typedValue.updatedAt : ""; - normalizedEntries[key] = { recent, updatedAt }; - } - return normalizedEntries; +const legacyPreferenceImports = new Map>(); + +function openPreferenceStore(env?: NodeJS.ProcessEnv) { + return getDiscordRuntime().state.openKeyedStore({ + namespace: "model-picker-preferences", + maxEntries: PREFERENCE_MAX_ENTRIES, + ...(env ? { env } : {}), + }); } export type DiscordModelPickerPreferenceScope = { @@ -58,11 +46,6 @@ export type DiscordModelPickerPreferenceScope = { userId: string; }; -function resolvePreferencesStorePath(env: NodeJS.ProcessEnv = process.env): string { - const stateDir = resolveStateDir(env, os.homedir); - return path.join(stateDir, "discord", "model-picker-preferences.json"); -} - function normalizeId(value?: string): string { return normalizeOptionalString(value) ?? ""; } @@ -116,20 +99,129 @@ function sanitizeRecentModels(models: string[] | undefined, limit: number): stri return deduped; } -async function readPreferencesStore(filePath: string): Promise { - const { value } = await readJsonFileWithFallback(filePath, { - version: 1, - entries: {} as Record, - }); - if (!value || typeof value !== "object" || value.version !== 1) { - return { version: 1, entries: {} }; +function sanitizeLegacyPreferenceEntry( + value: unknown, +): LegacyModelPickerPreferencesEntry | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const typedValue = value as { + recent?: unknown; + updatedAt?: unknown; + }; + const recent = Array.isArray(typedValue.recent) + ? typedValue.recent.filter((item: unknown): item is string => typeof item === "string") + : []; + const updatedAt = typeof typedValue.updatedAt === "string" ? typedValue.updatedAt : ""; + return { recent, updatedAt }; +} + +function sanitizeStoredPreferenceEntry(value: unknown): ModelPickerPreferencesEntry | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const typedValue = value as { + scopeKey?: unknown; + modelRef?: unknown; + updatedAt?: unknown; + }; + if (typeof typedValue.scopeKey !== "string" || typeof typedValue.modelRef !== "string") { + return undefined; + } + const modelRef = normalizeModelRef(typedValue.modelRef); + if (!modelRef) { + return undefined; } return { - version: 1, - entries: sanitizePreferenceEntries(value.entries), + scopeKey: typedValue.scopeKey, + modelRef, + updatedAt: typeof typedValue.updatedAt === "string" ? typedValue.updatedAt : "", }; } +function hashSegment(value: string, length: number): string { + return createHash("sha256").update(value, "utf8").digest("hex").slice(0, length); +} + +function buildPreferenceModelKey(scopeKey: string, modelRef: string): string { + return `v1:${hashSegment(scopeKey, 32)}:${hashSegment(modelRef, 24)}`; +} + +function timestampMs(value: string): number { + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : 0; +} + +function legacyUpdatedAtForIndex(updatedAt: string, index: number, total: number): string { + return new Date(timestampMs(updatedAt) + Math.max(0, total - index)).toISOString(); +} + +function nextPreferenceTimestampIso(): string { + lastPreferenceTimestampMs = Math.max(Date.now(), lastPreferenceTimestampMs + 1); + return new Date(lastPreferenceTimestampMs).toISOString(); +} + +function normalizeLegacyPreferenceKey(key: string): string | undefined { + const trimmed = key.trim(); + if (!trimmed || textEncoder.encode(trimmed).length > MAX_PLUGIN_STATE_KEY_BYTES) { + return undefined; + } + return trimmed; +} + +function resolveLegacyPreferencesPath(env?: NodeJS.ProcessEnv): string { + return path.join( + resolveStateDir(env ?? process.env, os.homedir), + "discord", + "model-picker-preferences.json", + ); +} + +async function importLegacyPreferences(env?: NodeJS.ProcessEnv): Promise { + const legacyPath = resolveLegacyPreferencesPath(env); + const stateDir = path.dirname(path.dirname(legacyPath)); + const existingImport = legacyPreferenceImports.get(stateDir); + if (existingImport) { + await existingImport; + return; + } + + const importPromise = (async () => { + const { value, exists } = + await readJsonFileWithFallback(legacyPath, null); + if (!exists || !value || typeof value.entries !== "object" || value.entries == null) { + return; + } + + const store = openPreferenceStore(env); + for (const [rawKey, rawEntry] of Object.entries(value.entries as Record)) { + const key = normalizeLegacyPreferenceKey(rawKey); + if (!key) { + continue; + } + const entry = sanitizeLegacyPreferenceEntry(rawEntry); + if (!entry || (entry.recent.length === 0 && !entry.updatedAt)) { + continue; + } + const recent = sanitizeRecentModels(entry.recent, 10); + for (const [index, modelRef] of recent.entries()) { + await store.registerIfAbsent(buildPreferenceModelKey(key, modelRef), { + scopeKey: key, + modelRef, + updatedAt: legacyUpdatedAtForIndex(entry.updatedAt, index, recent.length), + }); + } + } + })(); + legacyPreferenceImports.set(stateDir, importPromise); + try { + await importPromise; + } catch (error) { + legacyPreferenceImports.delete(stateDir); + throw error; + } +} + export async function readDiscordModelPickerRecentModels(params: { scope: DiscordModelPickerPreferenceScope; limit?: number; @@ -141,14 +233,24 @@ export async function readDiscordModelPickerRecentModels(params: { return []; } const limit = Math.max(1, Math.min(params.limit ?? DEFAULT_RECENT_LIMIT, 10)); - const filePath = resolvePreferencesStorePath(params.env); - const store = await readPreferencesStore(filePath); - const entry = store.entries[key]; - const recent = sanitizeRecentModels(entry?.recent, limit); - if (!params.allowedModelRefs || params.allowedModelRefs.size === 0) { - return recent; + try { + await importLegacyPreferences(params.env); + const store = openPreferenceStore(params.env); + const recent = (await store.entries()) + .map((entry) => sanitizeStoredPreferenceEntry(entry.value)) + .filter((entry): entry is ModelPickerPreferencesEntry => entry?.scopeKey === key) + .toSorted((left, right) => timestampMs(right.updatedAt) - timestampMs(left.updatedAt)) + .map((entry) => entry.modelRef); + if (!params.allowedModelRefs || params.allowedModelRefs.size === 0) { + return sanitizeRecentModels(recent, limit); + } + return sanitizeRecentModels( + recent.filter((modelRef) => params.allowedModelRefs?.has(modelRef)), + limit, + ); + } catch { + return []; } - return recent.filter((modelRef) => params.allowedModelRefs?.has(modelRef)); } export async function recordDiscordModelPickerRecentModel(params: { @@ -163,22 +265,28 @@ export async function recordDiscordModelPickerRecentModel(params: { return; } - const limit = Math.max(1, Math.min(params.limit ?? DEFAULT_RECENT_LIMIT, 10)); - const filePath = resolvePreferencesStorePath(params.env); - - await withFileLock(filePath, MODEL_PICKER_PREFERENCES_LOCK_OPTIONS, async () => { - const store = await readPreferencesStore(filePath); - const existing = sanitizeRecentModels(store.entries[key]?.recent, limit); - const next = [ - normalizedModelRef, - ...existing.filter((entry) => entry !== normalizedModelRef), - ].slice(0, limit); - - store.entries[key] = { - recent: next, - updatedAt: new Date().toISOString(), - }; - - await writeJsonFileAtomically(filePath, store); - }); + try { + await importLegacyPreferences(params.env); + const store = openPreferenceStore(params.env); + await store.register(buildPreferenceModelKey(key, normalizedModelRef), { + scopeKey: key, + modelRef: normalizedModelRef, + updatedAt: nextPreferenceTimestampIso(), + }); + const limit = Math.max(1, Math.min(params.limit ?? DEFAULT_RECENT_LIMIT, 10)); + const scopedEntries = (await store.entries()) + .map((entry) => ({ key: entry.key, value: sanitizeStoredPreferenceEntry(entry.value) })) + .filter( + (entry): entry is { key: string; value: ModelPickerPreferencesEntry } => + entry.value?.scopeKey === key, + ) + .toSorted( + (left, right) => + timestampMs(right.value.updatedAt) - timestampMs(left.value.updatedAt) || + left.key.localeCompare(right.key), + ); + await Promise.all(scopedEntries.slice(limit).map((entry) => store.delete(entry.key))); + } catch { + return; + } } diff --git a/extensions/feishu/legacy-state-migrations-api.ts b/extensions/feishu/legacy-state-migrations-api.ts new file mode 100644 index 000000000000..0648be423559 --- /dev/null +++ b/extensions/feishu/legacy-state-migrations-api.ts @@ -0,0 +1 @@ +export { detectFeishuLegacyStateMigrations } from "./src/dedup-migrations.js"; diff --git a/extensions/feishu/npm-shrinkwrap.json b/extensions/feishu/npm-shrinkwrap.json index dd558b1c0a4d..d4a8b49fc2bd 100644 --- a/extensions/feishu/npm-shrinkwrap.json +++ b/extensions/feishu/npm-shrinkwrap.json @@ -13,7 +13,7 @@ "zod": "4.4.3" }, "peerDependencies": { - "openclaw": ">=2026.5.28" + "openclaw": ">=2026.5.29" }, "peerDependenciesMeta": { "openclaw": { diff --git a/extensions/feishu/package.json b/extensions/feishu/package.json index 2928e3c874ad..0825fb73ca46 100644 --- a/extensions/feishu/package.json +++ b/extensions/feishu/package.json @@ -17,7 +17,7 @@ "openclaw": "workspace:*" }, "peerDependencies": { - "openclaw": ">=2026.5.28" + "openclaw": ">=2026.5.29" }, "peerDependenciesMeta": { "openclaw": { @@ -29,6 +29,9 @@ "./index.ts" ], "setupEntry": "./setup-entry.ts", + "setupFeatures": { + "legacyStateMigrations": true + }, "channel": { "id": "feishu", "label": "Feishu", @@ -45,10 +48,10 @@ "install": { "npmSpec": "@openclaw/feishu", "defaultChoice": "npm", - "minHostVersion": ">=2026.4.25" + "minHostVersion": ">=2026.5.29" }, "compat": { - "pluginApi": ">=2026.5.28" + "pluginApi": ">=2026.5.29" }, "build": { "openclawVersion": "2026.5.28" diff --git a/extensions/feishu/runtime-api.ts b/extensions/feishu/runtime-api.ts index 357953d6022d..1afa04bab800 100644 --- a/extensions/feishu/runtime-api.ts +++ b/extensions/feishu/runtime-api.ts @@ -44,7 +44,6 @@ export { resolveSessionStoreEntry, } from "openclaw/plugin-sdk/session-store-runtime"; export { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store"; -export { createPersistentDedupe } from "openclaw/plugin-sdk/persistent-dedupe"; export { normalizeAgentId } from "openclaw/plugin-sdk/routing"; export { chunkTextForOutbound } from "openclaw/plugin-sdk/text-chunking"; export { diff --git a/extensions/feishu/setup-entry.test.ts b/extensions/feishu/setup-entry.test.ts index 14607633e79f..83ed1b193d33 100644 --- a/extensions/feishu/setup-entry.test.ts +++ b/extensions/feishu/setup-entry.test.ts @@ -14,6 +14,8 @@ 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(typeof setupEntry.loadSetupPlugin).toBe("function"); + expect(setupEntry.loadLegacyStateMigrationDetector?.()).toBeTypeOf("function"); }); }); diff --git a/extensions/feishu/setup-entry.ts b/extensions/feishu/setup-entry.ts index 41216a676d96..bdca1aa6bcab 100644 --- a/extensions/feishu/setup-entry.ts +++ b/extensions/feishu/setup-entry.ts @@ -2,10 +2,17 @@ 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", diff --git a/extensions/feishu/src/dedup-migrations.test.ts b/extensions/feishu/src/dedup-migrations.test.ts new file mode 100644 index 000000000000..5c1e6819bab9 --- /dev/null +++ b/extensions/feishu/src/dedup-migrations.test.ts @@ -0,0 +1,89 @@ +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 { + 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([]); + }); +}); diff --git a/extensions/feishu/src/dedup-migrations.ts b/extensions/feishu/src/dedup-migrations.ts new file mode 100644 index 000000000000..8acf6e3e6f39 --- /dev/null +++ b/extensions/feishu/src/dedup-migrations.ts @@ -0,0 +1,102 @@ +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; + +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)) { + 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), + }, + ]; + }); +}; diff --git a/extensions/feishu/src/dedup-runtime-api.ts b/extensions/feishu/src/dedup-runtime-api.ts deleted file mode 100644 index e252fbeb4f91..000000000000 --- a/extensions/feishu/src/dedup-runtime-api.ts +++ /dev/null @@ -1 +0,0 @@ -export { createPersistentDedupe } from "openclaw/plugin-sdk/persistent-dedupe"; diff --git a/extensions/feishu/src/dedup.test.ts b/extensions/feishu/src/dedup.test.ts new file mode 100644 index 000000000000..1b3f1c749c82 --- /dev/null +++ b/extensions/feishu/src/dedup.test.ts @@ -0,0 +1,94 @@ +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 { + createPluginStateSyncKeyedStoreForTests, + 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 { + hasProcessedFeishuMessage, + testingHooks, + tryRecordMessagePersistent, + warmupDedupFromDisk, +} from "./dedup.js"; +import { setFeishuRuntime } from "./runtime.js"; + +let tempDir: string | undefined; +let previousStateDir: string | undefined; + +beforeEach(async () => { + previousStateDir = process.env.OPENCLAW_STATE_DIR; + tempDir = await fs.mkdtemp(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 () => { + vi.useRealTimers(); + testingHooks.resetFeishuDedupForTests(); + resetPluginStateStoreForTests(); + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + if (tempDir) { + await fs.rm(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); + await expect(hasProcessedFeishuMessage("msg-1", "account-a")).resolves.toBe(true); + await expect(hasProcessedFeishuMessage("msg-1", "account-b")).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(warmupDedupFromDisk("account-a")).resolves.toBe(1); + await expect(tryRecordMessagePersistent("msg-2", "account-a")).resolves.toBe(false); + }); + + it("ignores expired persisted entries", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + await expect(tryRecordMessagePersistent("msg-3", "account-a")).resolves.toBe(true); + testingHooks.resetFeishuDedupMemoryForTests(); + + vi.setSystemTime(1_000 + 24 * 60 * 60 * 1000 + 1); + await expect(hasProcessedFeishuMessage("msg-3", "account-a")).resolves.toBe(false); + }); + + it("imports legacy JSON dedupe entries before checking plugin state", 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", + ); + + await expect(hasProcessedFeishuMessage("msg-legacy", "account-a")).resolves.toBe(true); + await expect(tryRecordMessagePersistent("msg-legacy", "account-a")).resolves.toBe(false); + await expect(hasProcessedFeishuMessage("msg-expired", "account-a")).resolves.toBe(false); + }); +}); diff --git a/extensions/feishu/src/dedup.ts b/extensions/feishu/src/dedup.ts index f73c0ee75220..3e0c606b1f1b 100644 --- a/extensions/feishu/src/dedup.ts +++ b/extensions/feishu/src/dedup.ts @@ -1,17 +1,40 @@ +import { createHash } from "node:crypto"; import os from "node:os"; import path from "node:path"; -import { createPersistentDedupe } from "./dedup-runtime-api.js"; +import { loadJsonFile } from "openclaw/plugin-sdk/json-store"; +import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime"; import { releaseFeishuMessageProcessing, tryBeginFeishuMessageProcessing, } from "./processing-claims.js"; +import { getFeishuRuntime } from "./runtime.js"; // Persistent TTL: 24 hours — survives restarts & WebSocket reconnects. const DEDUP_TTL_MS = 24 * 60 * 60 * 1000; const MEMORY_MAX_SIZE = 1_000; -const FILE_MAX_ENTRIES = 10_000; +const STORE_MAX_ENTRIES = 10_000; +type FeishuDedupStoreEntry = { + namespace: string; + messageId: string; + seenAt: number; +}; -function resolveStateDirFromEnv(env: NodeJS.ProcessEnv = process.env): string { +const memory = new Map(); +const importedLegacyNamespaces = new Set(); +const cachedDedupStores = new Map>(); + +type LegacyDedupeData = Record; + +function normalizeMessageId(messageId: string | undefined | null): string | null { + const trimmed = messageId?.trim(); + return trimmed ? trimmed : null; +} + +function normalizeNamespace(namespace?: string): string { + return namespace?.trim() || "global"; +} + +function resolveLegacyStateDir(env: NodeJS.ProcessEnv = process.env): string { const stateOverride = env.OPENCLAW_STATE_DIR?.trim(); if (stateOverride) { return stateOverride; @@ -22,21 +45,121 @@ function resolveStateDirFromEnv(env: NodeJS.ProcessEnv = process.env): string { return path.join(os.homedir(), ".openclaw"); } -function resolveNamespaceFilePath(namespace: string): string { +function resolveLegacyNamespaceFilePath(namespace: string): string { const safe = namespace.replace(/[^a-zA-Z0-9_-]/g, "_"); - return path.join(resolveStateDirFromEnv(), "feishu", "dedup", `${safe}.json`); + return path.join(resolveLegacyStateDir(), "feishu", "dedup", `${safe}.json`); } -const persistentDedupe = createPersistentDedupe({ - ttlMs: DEDUP_TTL_MS, - memoryMaxSize: MEMORY_MAX_SIZE, - fileMaxEntries: FILE_MAX_ENTRIES, - resolveFilePath: resolveNamespaceFilePath, -}); +function pluginStateNamespace(namespace: string): string { + return `dedup.${namespace.replace(/[^a-zA-Z0-9_-]/g, "_")}`; +} -function normalizeMessageId(messageId: string | undefined | null): string | null { - const trimmed = messageId?.trim(); - return trimmed ? trimmed : null; +function openDedupStore(namespace: string): PluginStateSyncKeyedStore { + const stateNamespace = pluginStateNamespace(namespace); + const cached = cachedDedupStores.get(stateNamespace); + if (cached) { + return cached; + } + const store = getFeishuRuntime().state.openSyncKeyedStore({ + namespace: stateNamespace, + maxEntries: STORE_MAX_ENTRIES, + defaultTtlMs: DEDUP_TTL_MS, + }); + 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); +} + +function memoryKey(namespace: string, messageId: string): string { + return `${namespace}\0${messageId}`; +} + +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; +} + +function sanitizeLegacyDedupeData(value: unknown): LegacyDedupeData { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return {}; + } + const out: LegacyDedupeData = {}; + for (const [key, seenAt] of Object.entries(value as Record)) { + if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) { + out[key] = seenAt; + } + } + return out; +} + +function importLegacyDedupNamespace( + namespace: string, + now = Date.now(), + log?: (...args: unknown[]) => void, +): void { + if (importedLegacyNamespaces.has(namespace)) { + return; + } + + try { + const data = sanitizeLegacyDedupeData(loadJsonFile(resolveLegacyNamespaceFilePath(namespace))); + const store = openDedupStore(namespace); + for (const [messageId, seenAt] of Object.entries(data)) { + if (!isRecent(seenAt, now)) { + continue; + } + const key = dedupeStoreKey(namespace, messageId); + if (store.lookup(key) != null) { + continue; + } + store.register( + key, + { namespace, messageId, seenAt }, + { ttlMs: Math.max(1, DEDUP_TTL_MS - (now - seenAt)) }, + ); + } + importedLegacyNamespaces.add(namespace); + } catch (error) { + importedLegacyNamespaces.delete(namespace); + log?.(`feishu-dedup: legacy state import failed: ${String(error)}`); + } } export { releaseFeishuMessageProcessing, tryBeginFeishuMessageProcessing }; @@ -110,12 +233,58 @@ export async function tryRecordMessagePersistent( namespace = "global", log?: (...args: unknown[]) => void, ): Promise { - return persistentDedupe.checkAndRecord(messageId, { - namespace, - onDiskError: (error) => { - log?.(`feishu-dedup: disk error, falling back to memory: ${String(error)}`); - }, - }); + const normalizedNamespace = normalizeNamespace(namespace); + const normalizedMessageId = normalizeMessageId(messageId); + if (!normalizedMessageId) { + return true; + } + const now = Date.now(); + importLegacyDedupNamespace(normalizedNamespace, now, log); + 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( @@ -123,19 +292,64 @@ async function hasRecordedMessagePersistent( namespace = "global", log?: (...args: unknown[]) => void, ): Promise { - return persistentDedupe.hasRecent(messageId, { - namespace, - onDiskError: (error) => { - log?.(`feishu-dedup: persistent peek failed: ${String(error)}`); - }, - }); + const normalizedNamespace = normalizeNamespace(namespace); + const normalizedMessageId = normalizeMessageId(messageId); + if (!normalizedMessageId) { + return false; + } + const now = Date.now(); + importLegacyDedupNamespace(normalizedNamespace, now, log); + 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); + } } export async function warmupDedupFromDisk( namespace: string, log?: (...args: unknown[]) => void, ): Promise { - return persistentDedupe.warmup(namespace, (error) => { - log?.(`feishu-dedup: warmup disk error: ${String(error)}`); - }); + const normalizedNamespace = normalizeNamespace(namespace); + try { + let loaded = 0; + const now = Date.now(); + importLegacyDedupNamespace(normalizedNamespace, now, log); + 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; + } } + +export const testingHooks = { + resetFeishuDedupForTests() { + memory.clear(); + importedLegacyNamespaces.clear(); + for (const store of cachedDedupStores.values()) { + store.clear(); + } + cachedDedupStores.clear(); + }, + resetFeishuDedupMemoryForTests() { + memory.clear(); + }, +}; diff --git a/extensions/tsconfig.package-boundary.paths.json b/extensions/tsconfig.package-boundary.paths.json index a4e0d1d25a66..ea41a0d8112e 100644 --- a/extensions/tsconfig.package-boundary.paths.json +++ b/extensions/tsconfig.package-boundary.paths.json @@ -29,6 +29,9 @@ "openclaw/plugin-sdk/plugin-test-contracts": [ "../packages/plugin-sdk/dist/src/plugin-sdk/plugin-test-contracts.d.ts" ], + "openclaw/plugin-sdk/plugin-state-test-runtime": [ + "../packages/plugin-sdk/dist/src/plugin-sdk/plugin-state-test-runtime.d.ts" + ], "openclaw/plugin-sdk/plugin-test-runtime": [ "../packages/plugin-sdk/dist/src/plugin-sdk/plugin-test-runtime.d.ts" ], diff --git a/extensions/xai/tsconfig.json b/extensions/xai/tsconfig.json index 5185d603865e..b4a92551965b 100644 --- a/extensions/xai/tsconfig.json +++ b/extensions/xai/tsconfig.json @@ -30,6 +30,9 @@ "openclaw/plugin-sdk/plugin-test-contracts": [ "../../packages/plugin-sdk/dist/src/plugin-sdk/plugin-test-contracts.d.ts" ], + "openclaw/plugin-sdk/plugin-state-test-runtime": [ + "../../packages/plugin-sdk/dist/src/plugin-sdk/plugin-state-test-runtime.d.ts" + ], "openclaw/plugin-sdk/plugin-test-runtime": [ "../../packages/plugin-sdk/dist/src/plugin-sdk/plugin-test-runtime.d.ts" ], diff --git a/package.json b/package.json index 3df42afa8145..3520bc1a1df8 100644 --- a/package.json +++ b/package.json @@ -47,6 +47,8 @@ "!dist/plugin-sdk/plugin-test-api.d.ts", "!dist/plugin-sdk/plugin-test-contracts.js", "!dist/plugin-sdk/plugin-test-contracts.d.ts", + "!dist/plugin-sdk/plugin-state-test-runtime.js", + "!dist/plugin-sdk/plugin-state-test-runtime.d.ts", "!dist/plugin-sdk/plugin-test-runtime.js", "!dist/plugin-sdk/plugin-test-runtime.d.ts", "!dist/plugin-sdk/provider-http-test-mocks.js", @@ -637,6 +639,10 @@ "types": "./dist/plugin-sdk/migration-runtime.d.ts", "default": "./dist/plugin-sdk/migration-runtime.js" }, + "./plugin-sdk/plugin-state-runtime": { + "types": "./dist/plugin-sdk/plugin-state-runtime.d.ts", + "default": "./dist/plugin-sdk/plugin-state-runtime.js" + }, "./plugin-sdk/markdown-table-runtime": { "types": "./dist/plugin-sdk/markdown-table-runtime.d.ts", "default": "./dist/plugin-sdk/markdown-table-runtime.js" diff --git a/scripts/lib/official-external-channel-catalog.json b/scripts/lib/official-external-channel-catalog.json index ce64a33ba4a0..f66ee85860d6 100644 --- a/scripts/lib/official-external-channel-catalog.json +++ b/scripts/lib/official-external-channel-catalog.json @@ -167,7 +167,7 @@ "install": { "npmSpec": "@openclaw/feishu", "defaultChoice": "npm", - "minHostVersion": ">=2026.4.25" + "minHostVersion": ">=2026.5.29" } } }, diff --git a/scripts/lib/plugin-sdk-entrypoints.json b/scripts/lib/plugin-sdk-entrypoints.json index 05782291993e..ec9e9b67ee66 100644 --- a/scripts/lib/plugin-sdk-entrypoints.json +++ b/scripts/lib/plugin-sdk-entrypoints.json @@ -137,6 +137,8 @@ "logging-core", "migration", "migration-runtime", + "plugin-state-runtime", + "plugin-state-test-runtime", "markdown-table-runtime", "account-helpers", "account-core", diff --git a/scripts/lib/plugin-sdk-private-local-only-subpaths.json b/scripts/lib/plugin-sdk-private-local-only-subpaths.json index 82f65e719aa5..0c3f60f5f3d4 100644 --- a/scripts/lib/plugin-sdk-private-local-only-subpaths.json +++ b/scripts/lib/plugin-sdk-private-local-only-subpaths.json @@ -6,6 +6,7 @@ "codex-native-task-runtime", "plugin-test-api", "plugin-test-contracts", + "plugin-state-test-runtime", "plugin-test-runtime", "provider-http-test-mocks", "provider-test-contracts", diff --git a/src/channels/plugins/types.core.ts b/src/channels/plugins/types.core.ts index 9b6e4799c12f..21a3f1fca0fd 100644 --- a/src/channels/plugins/types.core.ts +++ b/src/channels/plugins/types.core.ts @@ -174,8 +174,8 @@ export type ChannelLegacyStateMigrationPlan = cleanupSource?: "rename"; preview?: string; readEntries: () => - | Array<{ key: string; value: unknown }> - | Promise>; + | Array<{ key: string; value: unknown; ttlMs?: number }> + | Promise>; }; /** User-facing metadata used in docs, pickers, and setup surfaces. */ diff --git a/src/commands/doctor-state-migrations.test.ts b/src/commands/doctor-state-migrations.test.ts index fa586a5d3c2a..094902d18693 100644 --- a/src/commands/doctor-state-migrations.test.ts +++ b/src/commands/doctor-state-migrations.test.ts @@ -4,6 +4,7 @@ import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../config/config.js"; import { + MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, createPluginStateKeyedStore, resetPluginStateStoreForTests, } from "../plugin-state/plugin-state-store.js"; @@ -650,7 +651,7 @@ describe("doctor legacy state migrations", () => { maxEntries: 4, scopeKey: "", cleanupSource: "rename", - readEntries: () => [{ key: "default", value: { body: "global" } }], + readEntries: () => [{ key: "default", value: { body: "global" }, ttlMs: 60_000 }], }, ]; @@ -709,6 +710,8 @@ describe("doctor legacy state migrations", () => { expect(Object.fromEntries(globalValuesByKey)).toEqual({ default: "global", }); + const globalEntries = await globalStore.entries(); + expect(globalEntries[0]?.expiresAt).toBeGreaterThan(Date.now()); }); }); @@ -724,7 +727,7 @@ describe("doctor legacy state migrations", () => { targetPath: "plugin state:test.capped-cache", pluginId: "telegram", namespace: "test.capped-cache", - maxEntries: 6_000, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, scopeKey: "scope", cleanupSource: "rename", readEntries: () => [ @@ -736,7 +739,7 @@ describe("doctor legacy state migrations", () => { await withStateDir(root, async () => { seedPluginStateEntriesForTests( - Array.from({ length: 5_999 }, (_, index) => ({ + Array.from({ length: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN - 1 }, (_, index) => ({ pluginId: "telegram", namespace: "test.sibling-cache", key: `sibling-${index}`, @@ -765,7 +768,7 @@ describe("doctor legacy state migrations", () => { await withStateDir(root, async () => { const store = createPluginStateKeyedStore<{ body: string }>("telegram", { namespace: "test.capped-cache", - maxEntries: 6_000, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, }); const valuesByKey = new Map( (await store.entries()).map(({ key, value }) => [key, value.body]), diff --git a/src/infra/state-migrations.ts b/src/infra/state-migrations.ts index 7e66c9043e8f..30c9b9726425 100644 --- a/src/infra/state-migrations.ts +++ b/src/infra/state-migrations.ts @@ -210,7 +210,11 @@ async function runLegacyMigrationPlans( break; } try { - await store.register(targetKey, entry.value); + await store.register( + targetKey, + entry.value, + entry.ttlMs != null ? { ttlMs: entry.ttlMs } : undefined, + ); const nextExpectedKeys = new Set(expectedKeys); nextExpectedKeys.add(targetKey); const liveKeys = new Set((await store.entries()).map(({ key }) => key)); diff --git a/src/plugin-sdk/plugin-state-runtime.ts b/src/plugin-sdk/plugin-state-runtime.ts new file mode 100644 index 000000000000..70178b8d3b73 --- /dev/null +++ b/src/plugin-sdk/plugin-state-runtime.ts @@ -0,0 +1,6 @@ +export type { + OpenKeyedStoreOptions, + PluginStateEntry, + PluginStateKeyedStore, + PluginStateSyncKeyedStore, +} from "../plugin-state/plugin-state-store.js"; diff --git a/src/plugin-sdk/plugin-state-test-runtime.ts b/src/plugin-sdk/plugin-state-test-runtime.ts new file mode 100644 index 000000000000..7750b4f3d335 --- /dev/null +++ b/src/plugin-sdk/plugin-state-test-runtime.ts @@ -0,0 +1,5 @@ +export { + createPluginStateKeyedStore as createPluginStateKeyedStoreForTests, + createPluginStateSyncKeyedStore as createPluginStateSyncKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "../plugin-state/plugin-state-store.js"; diff --git a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts index 80dcc8614fc3..5b622ce8f88e 100644 --- a/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts +++ b/src/plugin-sdk/test-helpers/plugin-runtime-mock.ts @@ -764,6 +764,9 @@ export function createPluginRuntimeMock(overrides: DeepPartial = openKeyedStore: vi.fn(() => { throw new Error("openKeyedStore mock is not configured"); }) as unknown as PluginRuntime["state"]["openKeyedStore"], + openSyncKeyedStore: vi.fn(() => { + throw new Error("openSyncKeyedStore mock is not configured"); + }) as unknown as PluginRuntime["state"]["openSyncKeyedStore"], }, tasks: { runs: { diff --git a/src/plugin-state/plugin-state-store.runtime.test.ts b/src/plugin-state/plugin-state-store.runtime.test.ts index 758c13a658ab..79960743cfbe 100644 --- a/src/plugin-state/plugin-state-store.runtime.test.ts +++ b/src/plugin-state/plugin-state-store.runtime.test.ts @@ -57,6 +57,9 @@ function createTestPluginRegistry() { openKeyedStore: () => { throw new Error("registry plugin runtime proxy should bind openKeyedStore"); }, + openSyncKeyedStore: () => { + throw new Error("registry plugin runtime proxy should bind openSyncKeyedStore"); + }, }, } as unknown as PluginRuntime, }); @@ -91,6 +94,13 @@ describe("plugin runtime state proxy", () => { }); await expect(telegramStore.lookup("k")).resolves.toBeUndefined(); await expect(store.lookup("k")).resolves.toEqual({ plugin: "discord" }); + + const syncStore = api.runtime.state.openSyncKeyedStore<{ plugin: string }>({ + namespace: "sync-runtime", + maxEntries: 10, + }); + expect(syncStore.registerIfAbsent("k", { plugin: "discord" })).toBe(true); + expect(syncStore.lookup("k")).toEqual({ plugin: "discord" }); }); }); @@ -119,6 +129,9 @@ describe("plugin runtime state proxy", () => { expect(() => api.runtime.state.openKeyedStore({ namespace: "runtime", maxEntries: 10 }), ).toThrow("openKeyedStore is only available for trusted plugins"); + expect(() => + api.runtime.state.openSyncKeyedStore({ namespace: "runtime", maxEntries: 10 }), + ).toThrow("openKeyedStore is only available for trusted plugins"); }); it("rejects untrusted global plugins", () => { diff --git a/src/plugin-state/plugin-state-store.sqlite.ts b/src/plugin-state/plugin-state-store.sqlite.ts index bba375535a84..cdd7cc492e2e 100644 --- a/src/plugin-state/plugin-state-store.sqlite.ts +++ b/src/plugin-state/plugin-state-store.sqlite.ts @@ -17,7 +17,7 @@ const PLUGIN_STATE_DIR_MODE = 0o700; const PLUGIN_STATE_FILE_MODE = 0o600; const PLUGIN_STATE_SIDECAR_SUFFIXES = ["", "-shm", "-wal"] as const; // Plugin-wide fuse only; namespace maxEntries still owns normal cache eviction. -const MAX_ENTRIES_PER_PLUGIN = 6_000; +const MAX_ENTRIES_PER_PLUGIN = 50_000; export const MAX_PLUGIN_STATE_VALUE_BYTES = 65_536; export const MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN = MAX_ENTRIES_PER_PLUGIN; @@ -98,7 +98,7 @@ function wrapPluginStateError( operation: PluginStateStoreOperation, fallbackCode: PluginStateStoreErrorCode, message: string, - pathname = resolvePluginStateSqlitePath(process.env), + pathname: string = resolvePluginStateSqlitePath(process.env), ): PluginStateStoreError { if (error instanceof PluginStateStoreError) { return error; @@ -282,8 +282,8 @@ function createStatements(db: DatabaseSync): PluginStateStatements { }; } -function ensurePluginStatePermissions(pathname: string) { - const dir = resolvePluginStateDir(process.env); +function ensurePluginStatePermissions(pathname: string, env: NodeJS.ProcessEnv = process.env) { + const dir = resolvePluginStateDir(env); mkdirSync(dir, { recursive: true, mode: PLUGIN_STATE_DIR_MODE }); chmodSync(dir, PLUGIN_STATE_DIR_MODE); for (const suffix of PLUGIN_STATE_SIDECAR_SUFFIXES) { @@ -294,9 +294,12 @@ function ensurePluginStatePermissions(pathname: string) { } } -function ensurePluginStatePermissionsBestEffort(pathname: string): void { +function ensurePluginStatePermissionsBestEffort( + pathname: string, + env: NodeJS.ProcessEnv = process.env, +): void { try { - ensurePluginStatePermissions(pathname); + ensurePluginStatePermissions(pathname, env); } catch { // The write already committed. Permission hardening is best-effort from here. } @@ -304,8 +307,9 @@ function ensurePluginStatePermissionsBestEffort(pathname: string): void { function openPluginStateDatabase( operation: PluginStateStoreOperation = "open", + env: NodeJS.ProcessEnv = process.env, ): PluginStateDatabase { - const pathname = resolvePluginStateSqlitePath(process.env); + const pathname = resolvePluginStateSqlitePath(env); if (cachedDatabase && cachedDatabase.path === pathname) { return cachedDatabase; } @@ -316,7 +320,7 @@ function openPluginStateDatabase( } try { - ensurePluginStatePermissions(pathname); + ensurePluginStatePermissions(pathname, env); } catch (error) { throw createPluginStateError({ code: "PLUGIN_STATE_OPEN_FAILED", @@ -346,7 +350,7 @@ function openPluginStateDatabase( db.exec("PRAGMA synchronous = NORMAL;"); db.exec("PRAGMA busy_timeout = 5000;"); ensureSchema(db, pathname); - ensurePluginStatePermissions(pathname); + ensurePluginStatePermissions(pathname, env); cachedDatabase = { db, path: pathname, @@ -373,14 +377,15 @@ function countRow(row: CountRow | undefined): number { function runWriteTransaction( operation: PluginStateStoreOperation, write: (store: PluginStateDatabase) => T, + env: NodeJS.ProcessEnv = process.env, ): T { - const store = openPluginStateDatabase(operation); - ensurePluginStatePermissions(store.path); + const store = openPluginStateDatabase(operation, env); + ensurePluginStatePermissions(store.path, env); store.db.exec("BEGIN IMMEDIATE"); try { const result = write(store); store.db.exec("COMMIT"); - ensurePluginStatePermissionsBestEffort(store.path); + ensurePluginStatePermissionsBestEffort(store.path, env); return result; } catch (error) { try { @@ -456,35 +461,41 @@ export function pluginStateRegister(params: { valueJson: string; maxEntries: number; ttlMs?: number; + env?: NodeJS.ProcessEnv; }): void { try { - runWriteTransaction("register", (store) => { - const now = Date.now(); - const expiresAt = params.ttlMs == null ? null : now + params.ttlMs; - store.statements.pruneExpiredNamespace.run(params.pluginId, params.namespace, now); - store.statements.upsertEntry.run({ - plugin_id: params.pluginId, - namespace: params.namespace, - entry_key: params.key, - value_json: params.valueJson, - created_at: now, - expires_at: expiresAt, - }); - enforcePostRegisterLimits({ - store, - pluginId: params.pluginId, - namespace: params.namespace, - maxEntries: params.maxEntries, - now, - protectedKey: params.key, - }); - }); + runWriteTransaction( + "register", + (store) => { + const now = Date.now(); + const expiresAt = params.ttlMs == null ? null : now + params.ttlMs; + store.statements.pruneExpiredNamespace.run(params.pluginId, params.namespace, now); + store.statements.upsertEntry.run({ + plugin_id: params.pluginId, + namespace: params.namespace, + entry_key: params.key, + value_json: params.valueJson, + created_at: now, + expires_at: expiresAt, + }); + enforcePostRegisterLimits({ + store, + pluginId: params.pluginId, + namespace: params.namespace, + maxEntries: params.maxEntries, + now, + protectedKey: params.key, + }); + }, + params.env, + ); } catch (error) { throw wrapPluginStateError( error, "register", "PLUGIN_STATE_WRITE_FAILED", "Failed to register plugin state entry.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -496,39 +507,45 @@ export function pluginStateRegisterIfAbsent(params: { valueJson: string; maxEntries: number; ttlMs?: number; + env?: NodeJS.ProcessEnv; }): boolean { try { - return runWriteTransaction("register", (store) => { - const now = Date.now(); - const expiresAt = params.ttlMs == null ? null : now + params.ttlMs; - store.statements.pruneExpiredNamespace.run(params.pluginId, params.namespace, now); - const result = store.statements.insertEntryIfAbsent.run({ - plugin_id: params.pluginId, - namespace: params.namespace, - entry_key: params.key, - value_json: params.valueJson, - created_at: now, - expires_at: expiresAt, - }); - if (result.changes === 0) { - return false; - } - enforcePostRegisterLimits({ - store, - pluginId: params.pluginId, - namespace: params.namespace, - maxEntries: params.maxEntries, - now, - protectedKey: params.key, - }); - return true; - }); + return runWriteTransaction( + "register", + (store) => { + const now = Date.now(); + const expiresAt = params.ttlMs == null ? null : now + params.ttlMs; + store.statements.pruneExpiredNamespace.run(params.pluginId, params.namespace, now); + const result = store.statements.insertEntryIfAbsent.run({ + plugin_id: params.pluginId, + namespace: params.namespace, + entry_key: params.key, + value_json: params.valueJson, + created_at: now, + expires_at: expiresAt, + }); + if (result.changes === 0) { + return false; + } + enforcePostRegisterLimits({ + store, + pluginId: params.pluginId, + namespace: params.namespace, + maxEntries: params.maxEntries, + now, + protectedKey: params.key, + }); + return true; + }, + params.env, + ); } catch (error) { throw wrapPluginStateError( error, "register", "PLUGIN_STATE_WRITE_FAILED", "Failed to register plugin state entry.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -537,9 +554,10 @@ export function pluginStateLookup(params: { pluginId: string; namespace: string; key: string; + env?: NodeJS.ProcessEnv; }): unknown { try { - const { statements } = openPluginStateDatabase("lookup"); + const { statements } = openPluginStateDatabase("lookup", params.env); const row = statements.selectEntry.get( params.pluginId, params.namespace, @@ -553,6 +571,7 @@ export function pluginStateLookup(params: { "lookup", "PLUGIN_STATE_READ_FAILED", "Failed to read plugin state entry.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -561,27 +580,33 @@ export function pluginStateConsume(params: { pluginId: string; namespace: string; key: string; + env?: NodeJS.ProcessEnv; }): unknown { try { - return runWriteTransaction("consume", (store) => { - const row = store.statements.selectEntry.get( - params.pluginId, - params.namespace, - params.key, - Date.now(), - ) as PluginStateRow | undefined; - if (!row) { - return undefined; - } - store.statements.deleteEntry.run(params.pluginId, params.namespace, params.key); - return parseStoredJson(row.value_json, "consume"); - }); + return runWriteTransaction( + "consume", + (store) => { + const row = store.statements.selectEntry.get( + params.pluginId, + params.namespace, + params.key, + Date.now(), + ) as PluginStateRow | undefined; + if (!row) { + return undefined; + } + store.statements.deleteEntry.run(params.pluginId, params.namespace, params.key); + return parseStoredJson(row.value_json, "consume"); + }, + params.env, + ); } catch (error) { throw wrapPluginStateError( error, "consume", "PLUGIN_STATE_READ_FAILED", "Failed to consume plugin state entry.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -590,9 +615,10 @@ export function pluginStateDelete(params: { pluginId: string; namespace: string; key: string; + env?: NodeJS.ProcessEnv; }): boolean { try { - const { statements } = openPluginStateDatabase("delete"); + const { statements } = openPluginStateDatabase("delete", params.env); const result = statements.deleteEntry.run(params.pluginId, params.namespace, params.key); return result.changes > 0; } catch (error) { @@ -601,6 +627,7 @@ export function pluginStateDelete(params: { "delete", "PLUGIN_STATE_WRITE_FAILED", "Failed to delete plugin state entry.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -608,9 +635,10 @@ export function pluginStateDelete(params: { export function pluginStateEntries(params: { pluginId: string; namespace: string; + env?: NodeJS.ProcessEnv; }): PluginStateEntry[] { try { - const { statements } = openPluginStateDatabase("entries"); + const { statements } = openPluginStateDatabase("entries", params.env); const rows = statements.selectEntries.all( params.pluginId, params.namespace, @@ -623,6 +651,7 @@ export function pluginStateEntries(params: { "entries", "PLUGIN_STATE_READ_FAILED", "Failed to list plugin state entries.", + resolvePluginStateSqlitePath(params.env), ); } } @@ -641,9 +670,13 @@ export function countPluginStateLiveEntries(pluginId: string): number { } } -export function pluginStateClear(params: { pluginId: string; namespace: string }): void { +export function pluginStateClear(params: { + pluginId: string; + namespace: string; + env?: NodeJS.ProcessEnv; +}): void { try { - const { statements } = openPluginStateDatabase("clear"); + const { statements } = openPluginStateDatabase("clear", params.env); statements.clearNamespace.run(params.pluginId, params.namespace); } catch (error) { throw wrapPluginStateError( @@ -651,6 +684,7 @@ export function pluginStateClear(params: { pluginId: string; namespace: string } "clear", "PLUGIN_STATE_WRITE_FAILED", "Failed to clear plugin state namespace.", + resolvePluginStateSqlitePath(params.env), ); } } diff --git a/src/plugin-state/plugin-state-store.test.ts b/src/plugin-state/plugin-state-store.test.ts index d2c9d2f365aa..ffa1a43e403b 100644 --- a/src/plugin-state/plugin-state-store.test.ts +++ b/src/plugin-state/plugin-state-store.test.ts @@ -6,10 +6,12 @@ import { type OpenClawTestState, } from "../test-utils/openclaw-test-state.js"; import { + MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, clearPluginStateStoreForTests, closePluginStateSqliteStore, createCorePluginStateKeyedStore, createPluginStateKeyedStore, + createPluginStateSyncKeyedStore, PluginStateStoreError, probePluginStateStore, resetPluginStateStoreForTests, @@ -78,6 +80,22 @@ describe("plugin state keyed store", () => { }); }); + it("supports synchronous keyed store callers", async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateSyncKeyedStore<{ count: number }>("discord", { + namespace: "sync-components", + maxEntries: 10, + }); + + expect(store.registerIfAbsent("interaction:1", { count: 1 })).toBe(true); + expect(store.registerIfAbsent("interaction:1", { count: 2 })).toBe(false); + expect(store.lookup("interaction:1")).toEqual({ count: 1 }); + expect(store.entries()).toMatchObject([{ key: "interaction:1", value: { count: 1 } }]); + expect(store.consume("interaction:1")).toEqual({ count: 1 }); + expect(store.lookup("interaction:1")).toBeUndefined(); + }); + }); + it("upserts values and refreshes deterministic entry ordering", async () => { await withPluginStateTestState(async () => { vi.useFakeTimers(); @@ -210,7 +228,7 @@ describe("plugin state keyed store", () => { expect((await evicting.entries()).map((entry) => entry.key)).toEqual(["b", "c"]); seedPluginStateEntriesForTests([ - ...Array.from({ length: 5_999 }, (_, entryIndex) => ({ + ...Array.from({ length: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN - 1 }, (_, entryIndex) => ({ pluginId: "limited-plugin", namespace: "limit", key: `k-${entryIndex}`, @@ -225,7 +243,7 @@ describe("plugin state keyed store", () => { ]); const limited = createPluginStateKeyedStore("limited-plugin", { namespace: "limit", - maxEntries: 6_001, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN + 1, }); const sibling = createPluginStateKeyedStore("limited-plugin", { namespace: "sibling", @@ -342,7 +360,7 @@ describe("plugin state keyed store", () => { it("evicts current namespace rows when sibling namespaces consume plugin row budget", async () => { await withPluginStateTestState(async () => { seedPluginStateEntriesForTests([ - ...Array.from({ length: 5_989 }, (_, entryIndex) => ({ + ...Array.from({ length: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN - 11 }, (_, entryIndex) => ({ pluginId: "telegram", namespace: "telegram.message-cache", key: `k-${entryIndex}`, @@ -358,7 +376,7 @@ describe("plugin state keyed store", () => { const messageStore = createPluginStateKeyedStore("telegram", { namespace: "telegram.message-cache", - maxEntries: 6_000, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, }); const topicStore = createPluginStateKeyedStore("telegram", { namespace: "telegram.topic-name-cache", @@ -378,7 +396,9 @@ describe("plugin state keyed store", () => { kind: "topic", entryIndex: 0, }); - await expect(messageStore.entries()).resolves.toHaveLength(5_989); + await expect(messageStore.entries()).resolves.toHaveLength( + MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN - 11, + ); await expect(topicStore.entries()).resolves.toHaveLength(11); }); }); @@ -436,7 +456,7 @@ describe("plugin state keyed store", () => { it("rejects plugin overflow when the current namespace cannot shed old rows", async () => { await withPluginStateTestState(async () => { seedPluginStateEntriesForTests( - Array.from({ length: 6_000 }, (_, entryIndex) => ({ + Array.from({ length: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN }, (_, entryIndex) => ({ pluginId: "telegram", namespace: "telegram.topic-name-cache", key: `topic-${entryIndex}`, @@ -446,11 +466,11 @@ describe("plugin state keyed store", () => { const messageStore = createPluginStateKeyedStore("telegram", { namespace: "telegram.message-cache", - maxEntries: 6_000, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, }); const topicStore = createPluginStateKeyedStore("telegram", { namespace: "telegram.topic-name-cache", - maxEntries: 6_000, + maxEntries: MAX_PLUGIN_STATE_ENTRIES_PER_PLUGIN, }); await expectPluginStateStoreError(messageStore.register("new-message", { fresh: true }), { diff --git a/src/plugin-state/plugin-state-store.ts b/src/plugin-state/plugin-state-store.ts index 3b227432be1d..8685c3d4502b 100644 --- a/src/plugin-state/plugin-state-store.ts +++ b/src/plugin-state/plugin-state-store.ts @@ -14,6 +14,7 @@ import type { OpenKeyedStoreOptions, PluginStateEntry, PluginStateKeyedStore, + PluginStateSyncKeyedStore, PluginStateStoreOperation, } from "./plugin-state-store.types.js"; import { PluginStateStoreError } from "./plugin-state-store.types.js"; @@ -22,6 +23,7 @@ export type { OpenKeyedStoreOptions, PluginStateEntry, PluginStateKeyedStore, + PluginStateSyncKeyedStore, PluginStateStoreErrorCode, PluginStateStoreOperation, PluginStateStoreProbeResult, @@ -47,6 +49,12 @@ type StoreOptionSignature = { defaultTtlMs?: number; }; +type PreparedRegisterParams = { + key: string; + valueJson: string; + ttlMs?: number; +}; + const namespaceOptionSignatures = new Map(); const textEncoder = new TextEncoder(); @@ -190,6 +198,27 @@ function assertValueSize(json: string): void { } } +function prepareRegisterParams( + key: string, + value: unknown, + defaultTtlMs?: number, + opts?: { ttlMs?: number }, +): PreparedRegisterParams { + const normalizedKey = validateKey(key, "register"); + assertJsonSerializable(value); + const json = JSON.stringify(value); + if (json === undefined) { + throw invalidInput("plugin state value must be JSON-serializable", "register"); + } + assertValueSize(json); + const ttlMs = validateOptionalTtlMs(opts?.ttlMs, "register") ?? defaultTtlMs; + return { + key: normalizedKey, + valueJson: json, + ...(ttlMs != null ? { ttlMs } : {}), + }; +} + function assertConsistentOptions( pluginId: string, namespace: string, @@ -219,65 +248,145 @@ function createKeyedStoreForPluginId( const namespace = validateNamespace(options.namespace); const maxEntries = validateMaxEntries(options.maxEntries); const defaultTtlMs = validateOptionalTtlMs(options.defaultTtlMs); + const env = options.env; assertConsistentOptions(pluginId, namespace, { maxEntries, defaultTtlMs }); - const prepareRegisterParams = ( - key: string, - value: T, - opts?: { ttlMs?: number }, - ): { key: string; valueJson: string; ttlMs?: number } => { - const normalizedKey = validateKey(key, "register"); - assertJsonSerializable(value); - const json = JSON.stringify(value); - assertValueSize(json); - const ttlMs = validateOptionalTtlMs(opts?.ttlMs, "register") ?? defaultTtlMs; - return { - key: normalizedKey, - valueJson: json, - ...(ttlMs != null ? { ttlMs } : {}), - }; - }; - return { async register(key, value, opts) { - const params = prepareRegisterParams(key, value, opts); + const params = prepareRegisterParams(key, value, defaultTtlMs, opts); pluginStateRegister({ pluginId, namespace, key: params.key, valueJson: params.valueJson, maxEntries, + ...(env ? { env } : {}), ...(params.ttlMs != null ? { ttlMs: params.ttlMs } : {}), }); }, async registerIfAbsent(key, value, opts) { - const params = prepareRegisterParams(key, value, opts); + const params = prepareRegisterParams(key, value, defaultTtlMs, opts); return pluginStateRegisterIfAbsent({ pluginId, namespace, key: params.key, valueJson: params.valueJson, maxEntries, + ...(env ? { env } : {}), ...(params.ttlMs != null ? { ttlMs: params.ttlMs } : {}), }); }, async lookup(key) { const normalizedKey = validateKey(key, "lookup"); - return pluginStateLookup({ pluginId, namespace, key: normalizedKey }) as T | undefined; + return pluginStateLookup({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }) as T | undefined; }, async consume(key) { const normalizedKey = validateKey(key, "consume"); - return pluginStateConsume({ pluginId, namespace, key: normalizedKey }) as T | undefined; + return pluginStateConsume({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }) as T | undefined; }, async delete(key) { const normalizedKey = validateKey(key, "delete"); - return pluginStateDelete({ pluginId, namespace, key: normalizedKey }); + return pluginStateDelete({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }); }, async entries() { - return pluginStateEntries({ pluginId, namespace }) as PluginStateEntry[]; + return pluginStateEntries({ + pluginId, + namespace, + ...(env ? { env } : {}), + }) as PluginStateEntry[]; }, async clear() { - pluginStateClear({ pluginId, namespace }); + pluginStateClear({ pluginId, namespace, ...(env ? { env } : {}) }); + }, + }; +} + +function createSyncKeyedStoreForPluginId( + pluginId: string, + options: OpenKeyedStoreOptions, +): PluginStateSyncKeyedStore { + const namespace = validateNamespace(options.namespace); + const maxEntries = validateMaxEntries(options.maxEntries); + const defaultTtlMs = validateOptionalTtlMs(options.defaultTtlMs); + const env = options.env; + assertConsistentOptions(pluginId, namespace, { maxEntries, defaultTtlMs }); + + return { + register(key, value, opts) { + const params = prepareRegisterParams(key, value, defaultTtlMs, opts); + pluginStateRegister({ + pluginId, + namespace, + key: params.key, + valueJson: params.valueJson, + maxEntries, + ...(env ? { env } : {}), + ...(params.ttlMs != null ? { ttlMs: params.ttlMs } : {}), + }); + }, + registerIfAbsent(key, value, opts) { + const params = prepareRegisterParams(key, value, defaultTtlMs, opts); + return pluginStateRegisterIfAbsent({ + pluginId, + namespace, + key: params.key, + valueJson: params.valueJson, + maxEntries, + ...(env ? { env } : {}), + ...(params.ttlMs != null ? { ttlMs: params.ttlMs } : {}), + }); + }, + lookup(key) { + const normalizedKey = validateKey(key, "lookup"); + return pluginStateLookup({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }) as T | undefined; + }, + consume(key) { + const normalizedKey = validateKey(key, "consume"); + return pluginStateConsume({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }) as T | undefined; + }, + delete(key) { + const normalizedKey = validateKey(key, "delete"); + return pluginStateDelete({ + pluginId, + namespace, + key: normalizedKey, + ...(env ? { env } : {}), + }); + }, + entries() { + return pluginStateEntries({ + pluginId, + namespace, + ...(env ? { env } : {}), + }) as PluginStateEntry[]; + }, + clear() { + pluginStateClear({ pluginId, namespace, ...(env ? { env } : {}) }); }, }; } @@ -292,12 +401,28 @@ export function createPluginStateKeyedStore( return createKeyedStoreForPluginId(pluginId, options); } +export function createPluginStateSyncKeyedStore( + pluginId: string, + options: OpenKeyedStoreOptions, +): PluginStateSyncKeyedStore { + if (pluginId.startsWith("core:")) { + throw invalidInput("Plugin ids starting with 'core:' are reserved for core consumers.", "open"); + } + return createSyncKeyedStoreForPluginId(pluginId, options); +} + export function createCorePluginStateKeyedStore( options: OpenKeyedStoreOptions & { ownerId: `core:${string}` }, ): PluginStateKeyedStore { return createKeyedStoreForPluginId(options.ownerId, options); } +export function createCorePluginStateSyncKeyedStore( + options: OpenKeyedStoreOptions & { ownerId: `core:${string}` }, +): PluginStateSyncKeyedStore { + return createSyncKeyedStoreForPluginId(options.ownerId, options); +} + export function clearPluginStateStoreForTests(): void { clearPluginStateSqliteStoreForTests(); namespaceOptionSignatures.clear(); diff --git a/src/plugin-state/plugin-state-store.types.ts b/src/plugin-state/plugin-state-store.types.ts index 9c07e95a7b36..79613aa21486 100644 --- a/src/plugin-state/plugin-state-store.types.ts +++ b/src/plugin-state/plugin-state-store.types.ts @@ -15,10 +15,21 @@ export type PluginStateKeyedStore = { clear(): Promise; }; +export type PluginStateSyncKeyedStore = { + register(key: string, value: T, opts?: { ttlMs?: number }): void; + registerIfAbsent(key: string, value: T, opts?: { ttlMs?: number }): boolean; + lookup(key: string): T | undefined; + consume(key: string): T | undefined; + delete(key: string): boolean; + entries(): PluginStateEntry[]; + clear(): void; +}; + export type OpenKeyedStoreOptions = { namespace: string; maxEntries: number; defaultTtlMs?: number; + env?: NodeJS.ProcessEnv; }; export type PluginStateStoreErrorCode = diff --git a/src/plugins/registry.ts b/src/plugins/registry.ts index 563ffb99d4c0..0d75775afbb0 100644 --- a/src/plugins/registry.ts +++ b/src/plugins/registry.ts @@ -28,8 +28,10 @@ import { } from "../infra/node-commands.js"; import { createPluginStateKeyedStore, + createPluginStateSyncKeyedStore, type OpenKeyedStoreOptions, type PluginStateKeyedStore, + type PluginStateSyncKeyedStore, } from "../plugin-state/plugin-state-store.js"; import { normalizePluginGatewayMethodScope } from "../shared/gateway-method-policy.js"; import { resolveGlobalSingleton } from "../shared/global-singleton.js"; @@ -2573,19 +2575,28 @@ export function createPluginRegistry(registryParams: PluginRegistryParams) { }; if (prop === "state") { const baseState = getRuntimeProperty(); + const assertPluginStateAllowed = () => { + const record = + pluginRuntimeRecordById.get(pluginId) ?? + registry.plugins.find((entry) => entry.id === pluginId); + if (record?.origin !== "bundled" && record?.trustedOfficialInstall !== true) { + throw new Error( + "openKeyedStore is only available for trusted plugins in this release.", + ); + } + }; return { ...baseState, openKeyedStore: (options: OpenKeyedStoreOptions): PluginStateKeyedStore => { - const record = - pluginRuntimeRecordById.get(pluginId) ?? - registry.plugins.find((entry) => entry.id === pluginId); - if (record?.origin !== "bundled" && record?.trustedOfficialInstall !== true) { - throw new Error( - "openKeyedStore is only available for trusted plugins in this release.", - ); - } + assertPluginStateAllowed(); return createPluginStateKeyedStore(pluginId, options); }, + openSyncKeyedStore: ( + options: OpenKeyedStoreOptions, + ): PluginStateSyncKeyedStore => { + assertPluginStateAllowed(); + return createPluginStateSyncKeyedStore(pluginId, options); + }, } satisfies PluginRuntime["state"]; } if (prop === "config") { diff --git a/src/plugins/runtime/index.ts b/src/plugins/runtime/index.ts index f704f410e6c5..ea9632e66c03 100644 --- a/src/plugins/runtime/index.ts +++ b/src/plugins/runtime/index.ts @@ -253,6 +253,9 @@ export function createPluginRuntime(_options: CreatePluginRuntimeOptions = {}): openKeyedStore: () => { throw new Error("openKeyedStore is only available through the plugin runtime proxy."); }, + openSyncKeyedStore: () => { + throw new Error("openSyncKeyedStore is only available through the plugin runtime proxy."); + }, }, tasks, taskFlow, diff --git a/src/plugins/runtime/types-core.ts b/src/plugins/runtime/types-core.ts index 0277d6942f62..dc9fb755c673 100644 --- a/src/plugins/runtime/types-core.ts +++ b/src/plugins/runtime/types-core.ts @@ -311,6 +311,9 @@ export type PluginRuntimeCore = { openKeyedStore: ( options: import("../../plugin-state/plugin-state-store.types.js").OpenKeyedStoreOptions, ) => import("../../plugin-state/plugin-state-store.types.js").PluginStateKeyedStore; + openSyncKeyedStore: ( + options: import("../../plugin-state/plugin-state-store.types.js").OpenKeyedStoreOptions, + ) => import("../../plugin-state/plugin-state-store.types.js").PluginStateSyncKeyedStore; }; tasks: { runs: PluginRuntimeTaskRuns;