diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 9028ab67547..d1e94f5dc8a 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -53,7 +53,7 @@ fa2df02bede6ed8843e5c2bd605c6ea5cd20313c305593fbd371dba6f1b931c3 module/command eb4c757fe0086c1dbfa4c3f3caf3dcff0d3cab3924c608237f08f740a6ee5f59 module/command-status 235a9e4d983042c3db156efcab0e333984cbbf13ebdfcc44a3a5ab40cf3edb4f module/config-contracts 20f3f8042de53e4eee61b64de9102c8c202b9299e6a29235647a4729f70145f2 module/config-mutation -0c0bb80f81d0c769f343e3be95f01b3d734077c12e831367440952052aafbeb7 module/config-runtime +6f23fd2d777f7189c38b34a51c6899abdad4ee4e6d74458eb3183c8ce03b8677 module/config-runtime c1ea9510dfda047609a99d5d2cd1f1560f5d469a36e6b695766213d695c25b0f module/conversation-runtime d9267aacc65aeebf0046d4eef691e5a510764063e6c7942b1d6bc7a282ca67a1 module/core 4af19d59c2f18674e7d7f7dc1b358b644dc707e6bd601dc47168bd9e4a669940 module/dedupe-runtime @@ -116,7 +116,7 @@ b6b8edc50ecab8386c9acd8f374a207212b5a99c8f518538bbcf0c458dda3881 module/runtime 17a6a199714ba8308e62928c0491bcf9fd214c923c3879aa99a047a30138253d module/secret-ref-runtime 596a315d426121c9620b314e3a9a7f523840b46e007d94d0d5e83cdedf789d15 module/security-runtime 31b785e74f1f8f56241b7756ef6a5d86199c5ce177cbb1c234a261866972f270 module/session-discussion -f59099aa2d536246d4b1297f01bcea66796351a9259debdd45909afeb7a42d86 module/session-store-runtime +9d7d884330397701c7de9f5b6800b970b2b349027491704184cb1bd6cda1fd00 module/session-store-runtime 695971d31b3e16f0bf9b643acc30df312fe94b3c0dfb27a2e6156c8a00b5e261 module/setup 9129c17df1523903f34beffd348ce177fcaa2983e155a8c854a50cf1bd7e99c4 module/setup-runtime cd431f6ba8327b81438b7a63b1963120f200f5abd145fb6aa7c5c561339cb0b1 module/setup-tools diff --git a/extensions/zalouser/doctor-contract-api.test.ts b/extensions/zalouser/doctor-contract-api.test.ts index 692349d2b3f..31d0656316c 100644 --- a/extensions/zalouser/doctor-contract-api.test.ts +++ b/extensions/zalouser/doctor-contract-api.test.ts @@ -149,10 +149,28 @@ describe("zalouser doctor state migration", () => { await expect(fs.access(`${filePath}.migrated`)).resolves.toBeUndefined(); }); + it("does not inspect agent session stores when zalouser has never been configured", async () => { + const migration = findMigration("zalouser-direct-session-keys"); + const context = createDoctorContext(env); + const config = { agents: { list: [{ id: "worker-1" }] } }; + + await expect( + migration.detectLegacyState({ config, env, stateDir, oauthDir: stateDir, context }), + ).resolves.toBeNull(); + for (const agentId of ["main", "worker-1"]) { + await expect( + fs.access(path.join(stateDir, "agents", agentId, "agent", "openclaw-agent.sqlite")), + ).rejects.toThrow(); + } + }); + it("moves legacy group-shaped DM sessions to canonical direct keys", async () => { const legacyKey = "agent:main:zalouser:group:user-1"; const canonicalKey = "agent:main:zalouser:direct:user-1"; - const config = { session: { store: storePath, dmScope: "per-channel-peer" as const } }; + const config = { + channels: { zalouser: {} }, + session: { store: storePath, dmScope: "per-channel-peer" as const }, + }; await upsertSessionEntry({ agentId: "main", env, @@ -198,6 +216,7 @@ describe("zalouser doctor state migration", () => { const secondLegacyKey = "agent:main:zalouser:group:user-2"; const canonicalKey = "agent:main:zalouser:direct:alice"; const config = { + channels: { zalouser: {} }, session: { store: storePath, dmScope: "per-channel-peer" as const, diff --git a/extensions/zalouser/doctor-contract-api.ts b/extensions/zalouser/doctor-contract-api.ts index fff4e4e961e..50bfede317c 100644 --- a/extensions/zalouser/doctor-contract-api.ts +++ b/extensions/zalouser/doctor-contract-api.ts @@ -84,6 +84,7 @@ async function collectLegacyZalouserCredentialSources( function collectLegacyZalouserDmEntries( config: OpenClawConfig, env: NodeJS.ProcessEnv, + options: { readOnly?: boolean } = {}, ): LegacyZalouserDmEntry[] { const entries = new Map(); const fallbackAccountId = config.channels?.zalouser?.defaultAccount?.trim() || "default"; @@ -93,7 +94,11 @@ function collectLegacyZalouserDmEntries( ]); for (const agentId of agentIds) { const storePath = resolveStorePath(config.session?.store, { agentId, env }); - const storedEntries = listSessionEntries({ agentId, storePath }); + const storedEntries = listSessionEntries({ + agentId, + storePath, + ...(options.readOnly ? { readOnly: true } : {}), + }); const entryByKey = new Map(storedEntries.map(({ sessionKey, entry }) => [sessionKey, entry])); for (const { sessionKey, entry } of storedEntries) { const parsed = parseAgentSessionKey(sessionKey); @@ -228,10 +233,17 @@ export const stateMigrations: PluginDoctorStateMigration[] = [ { id: "zalouser-direct-session-keys", label: "Zalo Personal direct-message sessions", - detectLegacyState({ config, env }) { - const count = collectLegacyZalouserDmEntries(config, env).flatMap( - ({ legacyKeys }) => legacyKeys, - ).length; + async detectLegacyState({ config, env }) { + // A never-configured channel cannot own legacy DMs, so do not scan every agent DB at startup. + // Removed config defers leftover-row detection until zalouser is configured again. + if ( + config.channels?.zalouser === undefined && + (await collectLegacyZalouserCredentialSources(env)).length === 0 + ) { + return null; + } + const pending = collectLegacyZalouserDmEntries(config, env, { readOnly: true }); + const count = pending.flatMap(({ legacyKeys }) => legacyKeys).length; return count > 0 ? { preview: [`- Zalo Personal direct-message session keys: ${count} legacy row(s)`] } : null; diff --git a/src/commands/health.plugins.test.ts b/src/commands/health.plugins.test.ts index 1c4963e1461..604371f142c 100644 --- a/src/commands/health.plugins.test.ts +++ b/src/commands/health.plugins.test.ts @@ -25,6 +25,7 @@ describe("getHealthSnapshot plugin state", () => { })); vi.doMock("../config/sessions/session-accessor.js", () => ({ listSessionEntries: () => [], + listSessionEntriesReadOnly: () => [], })); vi.doMock("../channels/plugins/read-only.js", () => ({ listReadOnlyChannelPluginsForConfig: () => [], diff --git a/src/commands/health.snapshot.test.ts b/src/commands/health.snapshot.test.ts index 362c2b2760a..ad9520d9516 100644 --- a/src/commands/health.snapshot.test.ts +++ b/src/commands/health.snapshot.test.ts @@ -73,7 +73,7 @@ async function loadFreshHealthModulesForTest() { loadSessionStore: () => testStore, })); vi.doMock("../config/sessions/session-accessor.js", () => ({ - listSessionEntries: (scope?: { agentId?: string; storePath?: string }) => { + listSessionEntriesReadOnly: (scope?: { agentId?: string; storePath?: string }) => { listHealthSessionEntriesCalls.push(scope ?? {}); return Object.entries(testStore).map(([sessionKey, entry]) => ({ sessionKey, entry })); }, diff --git a/src/commands/health.ts b/src/commands/health.ts index 2624d6a649a..1223cc31b81 100644 --- a/src/commands/health.ts +++ b/src/commands/health.ts @@ -366,11 +366,22 @@ const resolveAgentOrder = (cfg: OpenClawConfig) => { }; const buildSessionSummary = async (storePath: string, agentId?: string) => { - const { listSessionEntries } = await import("../config/sessions/session-accessor.js"); - const sessions = listSessionEntries({ - ...(agentId ? { agentId } : {}), - storePath, - }) + const { listSessionEntriesReadOnly } = await import("../config/sessions/session-accessor.js"); + const { isTransientSqliteError } = await import("../infra/unhandled-rejections.js"); + let listed: ReturnType; + try { + listed = listSessionEntriesReadOnly({ + ...(agentId ? { agentId } : {}), + storePath, + }); + } catch (error) { + if (!isTransientSqliteError(error)) { + throw error; + } + // Health is best-effort: an empty snapshot beats failing on a transient lock. + listed = []; + } + const sessions = listed .filter(({ sessionKey }) => sessionKey !== "global" && sessionKey !== "unknown") .map(({ sessionKey, entry }) => ({ key: sessionKey, updatedAt: entry?.updatedAt ?? 0 })) .toSorted((a, b) => b.updatedAt - a.updatedAt); diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index 7cf1e8d5f15..9532f0caab1 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -11,6 +11,7 @@ import { resolveStorePath } from "./paths.js"; import { clearPluginOwnedSessionState } from "./plugin-host-cleanup.js"; import { listSqliteSessionEntries, + listSqliteSessionEntriesReadOnly, loadExactSqliteSessionEntry, loadSqliteSessionEntry, loadSqliteSessionEntryReadOnly, @@ -319,6 +320,16 @@ export function listSessionEntries(scope: SessionEntryListScope = {}): SessionEn return listSqliteSessionEntries(scope); } +/** + * Health/status introspection must not join the writable lifecycle or register databases; + * doing so churns fleet-wide agent handles on every health tick. + */ +export function listSessionEntriesReadOnly( + scope: SessionEntryListScope = {}, +): SessionEntrySummary[] { + return listSqliteSessionEntriesReadOnly(scope); +} + /** * Borrowed keyed view over one resolved store for synchronous read-only hot paths. * Unlike loadSessionEntry, `get` is a raw exact persisted-key probe with no alias diff --git a/src/config/sessions/session-accessor.readonly.test.ts b/src/config/sessions/session-accessor.readonly.test.ts new file mode 100644 index 00000000000..568dbd63f3a --- /dev/null +++ b/src/config/sessions/session-accessor.readonly.test.ts @@ -0,0 +1,88 @@ +import fs from "node:fs"; +import { afterEach, describe, expect, it } from "vitest"; +import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js"; +import { + closeOpenClawAgentDatabasesForTest, + isOpenClawAgentDatabaseOpen, + resolveOpenClawAgentSqlitePath, +} from "../../state/openclaw-agent-db.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../../state/openclaw-state-db.js"; +import { + listSessionEntries, + listSessionEntriesReadOnly, + upsertSessionEntry, +} from "./session-accessor.js"; + +const tempDirs: string[] = []; + +function countRegisteredAgentDatabases(env: NodeJS.ProcessEnv): number { + const row = openOpenClawStateDatabase({ env }) + .db.prepare("SELECT count(*) AS count FROM agent_databases") + .get() as { count: number }; + return row.count; +} + +function clearRegisteredAgentDatabases(env: NodeJS.ProcessEnv): void { + openOpenClawStateDatabase({ env }).db.prepare("DELETE FROM agent_databases").run(); +} + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + cleanupTempDirs(tempDirs); +}); + +describe("session accessor readonly listing", () => { + it("returns the same entries as the writable listing for a populated agent database", async () => { + const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-populated-"); + const env = { OPENCLAW_STATE_DIR: stateDir }; + const listScope = { agentId: "worker-1", env }; + + await upsertSessionEntry( + { ...listScope, sessionKey: "agent:worker-1:main" }, + { sessionId: "session-1", updatedAt: 10 }, + ); + await upsertSessionEntry( + { ...listScope, sessionKey: "agent:worker-1:telegram:dm:42" }, + { sessionId: "session-2", updatedAt: 20 }, + ); + const writableEntries = listSessionEntries(listScope); + closeOpenClawAgentDatabasesForTest(); + + expect(listSessionEntriesReadOnly(listScope)).toEqual(writableEntries); + }); + + it("returns an empty list without creating or registering a missing agent database", () => { + const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-missing-"); + const env = { OPENCLAW_STATE_DIR: stateDir }; + const agentId = "worker-1"; + const databasePath = resolveOpenClawAgentSqlitePath({ agentId, env }); + clearRegisteredAgentDatabases(env); + + expect(listSessionEntriesReadOnly({ agentId, env })).toEqual([]); + expect(fs.existsSync(databasePath)).toBe(false); + expect(countRegisteredAgentDatabases(env)).toBe(0); + }); + + it("does not register a populated database during readonly health-style listing", async () => { + const stateDir = makeTempDir(tempDirs, "openclaw-session-readonly-registry-"); + const env = { OPENCLAW_STATE_DIR: stateDir }; + const agentId = "worker-1"; + const scope = { agentId, env }; + + await upsertSessionEntry( + { ...scope, sessionKey: "agent:worker-1:main" }, + { sessionId: "session-1", updatedAt: 10 }, + ); + const databasePath = resolveOpenClawAgentSqlitePath({ agentId, env }); + closeOpenClawAgentDatabasesForTest(); + clearRegisteredAgentDatabases(env); + + expect(listSessionEntriesReadOnly(scope)).toHaveLength(1); + expect(countRegisteredAgentDatabases(env)).toBe(0); + expect(isOpenClawAgentDatabaseOpen(databasePath)).toBe(false); + }); +}); diff --git a/src/config/sessions/session-accessor.sqlite-entry.ts b/src/config/sessions/session-accessor.sqlite-entry.ts index 00de71a2f1e..e4380cdb5a5 100644 --- a/src/config/sessions/session-accessor.sqlite-entry.ts +++ b/src/config/sessions/session-accessor.sqlite-entry.ts @@ -1,3 +1,4 @@ +import type { DatabaseSync } from "node:sqlite"; import type { MsgContext } from "../../auto-reply/templating.js"; import { executeSqliteQuerySync, @@ -131,6 +132,26 @@ export function listSqliteSessionEntries( ): SessionEntrySummary[] { const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + return listSqliteSessionEntriesFromDatabase(database); +} + +/** + * Lists session entries without opening the agent database writable. + * Transient lock errors propagate: only the caller knows whether "empty" is an + * acceptable degradation (health snapshots) or hides real state (migration detection). + */ +export function listSqliteSessionEntriesReadOnly( + scope: Partial> = {}, +): SessionEntrySummary[] { + const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); + const result = withOpenClawAgentDatabaseReadOnly( + (database) => listSqliteSessionEntriesFromDatabase(database), + toDatabaseOptions(resolved), + ); + return result.found ? result.value : []; +} + +function listSqliteSessionEntriesFromDatabase(database: { db: DatabaseSync }) { const db = getSessionKysely(database.db); const rows = executeSqliteQuerySync( database.db, diff --git a/src/config/sessions/session-accessor.sqlite.ts b/src/config/sessions/session-accessor.sqlite.ts index f0697f17b47..83f4a51d98f 100644 --- a/src/config/sessions/session-accessor.sqlite.ts +++ b/src/config/sessions/session-accessor.sqlite.ts @@ -1,6 +1,7 @@ // Stable SQLite accessor surface. Domain owners live in the focused modules below. export { listSqliteSessionEntries, + listSqliteSessionEntriesReadOnly, listSqliteSessionEntriesByStatus, listSqliteSessionTranscriptInstances, loadExactSqliteSessionEntry, diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index e00e0cda218..7f6abfd5ef8 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -122,6 +122,7 @@ export type { export { clearPluginOwnedSessionState, listSessionEntries, + listSessionEntriesReadOnly, loadExactSessionEntry, loadSessionEntry, loadSessionEntryReadOnly, diff --git a/src/gateway/server-methods/usage.test.ts b/src/gateway/server-methods/usage.test.ts index f4ab4ff8086..2c9df2be9c0 100644 --- a/src/gateway/server-methods/usage.test.ts +++ b/src/gateway/server-methods/usage.test.ts @@ -460,6 +460,34 @@ describe("gateway usage helpers", () => { ); }); + it("keeps refreshing cost summaries fresh for the TTL window", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-02-05T00:00:00.000Z")); + vi.mocked(loadCostUsageSummaryFromCache).mockResolvedValueOnce({ + ...costSummary({ totalTokens: 1, totalCost: 0 }), + cacheStatus: { + status: "refreshing", + cachedFiles: 1, + pendingFiles: 1, + staleFiles: 1, + }, + }); + + const config = {} as OpenClawConfig; + await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config }); + + const entry = testApi.costUsageCache.get("agent:__default__:1-2:gateway"); + expect(entry?.updatedAt).toBe(Date.now()); + + await vi.advanceTimersByTimeAsync(29_999); + await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config }); + expect(vi.mocked(loadCostUsageSummaryFromCache)).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(1); + await testApi.loadCostUsageSummaryCached({ startMs: 1, endMs: 2, config }); + expect(vi.mocked(loadCostUsageSummaryFromCache)).toHaveBeenCalledTimes(2); + }); + it("keeps cost usage cache entries scoped by agentId", async () => { const config = {} as OpenClawConfig; diff --git a/src/gateway/server-methods/usage.ts b/src/gateway/server-methods/usage.ts index 655348ba825..6ac7a1fe77c 100644 --- a/src/gateway/server-methods/usage.ts +++ b/src/gateway/server-methods/usage.ts @@ -905,12 +905,7 @@ async function loadCostUsageSummaryCached(params: { const cacheKey = `${params.agentScope === "all" ? "all" : `agent:${params.agentId ?? "__default__"}`}:${params.startMs}-${params.endMs}:${dayBucketKey}`; const now = Date.now(); const cached = costUsageCache.get(cacheKey); - if ( - cached?.summary && - cached.updatedAt && - now - cached.updatedAt < COST_USAGE_CACHE_TTL_MS && - cached.summary.cacheStatus?.status !== "refreshing" - ) { + if (cached?.summary && cached.updatedAt && now - cached.updatedAt < COST_USAGE_CACHE_TTL_MS) { return cached.summary; } @@ -941,9 +936,11 @@ async function loadCostUsageSummaryCached(params: { }) ) .then((summary) => { + // Refresh work is independent; retaining freshness prevents fleet rescans while it runs. + // The short TTL still picks up a completed refresh promptly. setCostUsageCache(cacheKey, { summary, - updatedAt: summary.cacheStatus?.status === "refreshing" ? undefined : Date.now(), + updatedAt: Date.now(), }); return summary; }) diff --git a/src/infra/session-cost-usage-cache.sqlite.test.ts b/src/infra/session-cost-usage-cache.sqlite.test.ts new file mode 100644 index 00000000000..09e2cea0b13 --- /dev/null +++ b/src/infra/session-cost-usage-cache.sqlite.test.ts @@ -0,0 +1,82 @@ +import fs from "node:fs"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, + resolveOpenClawAgentSqlitePath, +} from "../state/openclaw-agent-db.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../state/openclaw-state-db.js"; +import { withEnv } from "../test-utils/env.js"; +import { + isSessionCostUsageRefreshRunning, + readSessionCostUsageRollupRows, + writeSessionCostUsageRollup, +} from "./session-cost-usage-cache.sqlite.js"; + +const tempDirs: string[] = []; + +function countRegisteredAgentDatabases(): number { + const row = openOpenClawStateDatabase() + .db.prepare("SELECT count(*) AS count FROM agent_databases") + .get() as { + count: number; + }; + return row.count; +} + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + cleanupTempDirs(tempDirs); +}); + +describe("session cost usage SQLite cache", () => { + it("returns empty values without creating a missing agent database", () => { + const stateDir = makeTempDir(tempDirs, "openclaw-usage-cache-missing-"); + + withEnv({ OPENCLAW_STATE_DIR: stateDir }, () => { + const databasePath = resolveOpenClawAgentSqlitePath({ agentId: "worker-1" }); + + expect(readSessionCostUsageRollupRows("worker-1", databasePath)).toEqual([]); + expect(isSessionCostUsageRefreshRunning("worker-1", databasePath)).toBe(false); + expect(fs.existsSync(databasePath)).toBe(false); + expect(fs.existsSync(path.join(stateDir, "state", "openclaw.sqlite"))).toBe(false); + }); + }); + + it("does not register readonly cache reads while writes still register", () => { + const stateDir = makeTempDir(tempDirs, "openclaw-usage-cache-registry-"); + + withEnv({ OPENCLAW_STATE_DIR: stateDir }, () => { + const agentId = "worker-1"; + const database = openOpenClawAgentDatabase({ agentId }); + const databasePath = database.path; + closeOpenClawAgentDatabasesForTest(); + + const stateDatabase = openOpenClawStateDatabase(); + stateDatabase.db.prepare("DELETE FROM agent_databases").run(); + expect(countRegisteredAgentDatabases()).toBe(0); + + expect(readSessionCostUsageRollupRows(agentId, databasePath)).toEqual([]); + expect(isSessionCostUsageRefreshRunning(agentId, databasePath)).toBe(false); + expect(countRegisteredAgentDatabases()).toBe(0); + + expect( + writeSessionCostUsageRollup({ + agentId, + databasePath, + rollupId: "session.jsonl", + previousValueJson: null, + valueJson: "{}", + updatedAt: 1, + }), + ).toBe(true); + expect(countRegisteredAgentDatabases()).toBe(1); + }); + }); +}); diff --git a/src/infra/session-cost-usage-cache.sqlite.ts b/src/infra/session-cost-usage-cache.sqlite.ts index eb1b7075744..e363defc07d 100644 --- a/src/infra/session-cost-usage-cache.sqlite.ts +++ b/src/infra/session-cost-usage-cache.sqlite.ts @@ -1,11 +1,11 @@ +import type { DatabaseSync } from "node:sqlite"; import { normalizeAgentId } from "../routing/session-key.js"; +import { withOpenClawAgentDatabaseReadOnly } from "../state/openclaw-agent-db-readonly.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../state/openclaw-agent-db.generated.js"; -import { - openOpenClawAgentDatabase, - runOpenClawAgentWriteTransaction, -} from "../state/openclaw-agent-db.js"; +import { runOpenClawAgentWriteTransaction } from "../state/openclaw-agent-db.js"; // Per-agent SQLite storage for rebuildable per-session usage rollups. import { executeSqliteQuerySync, getNodeSqliteKysely } from "./kysely-sync.js"; +import { isTransientSqliteError } from "./unhandled-rejections.js"; const LEGACY_CACHE_SCOPE = "session-cost-usage"; const LEGACY_CACHE_KEY = "cache"; @@ -26,11 +26,24 @@ type SessionCostUsageRollupRow = { valueJson: string; }; -function openCacheDatabase(agentId: string | undefined, databasePath?: string) { - return openOpenClawAgentDatabase({ - agentId: normalizeAgentId(agentId), - ...(databasePath ? { path: databasePath } : {}), - }); +function readCacheDatabase( + agentId: string | undefined, + databasePath: string | undefined, + operation: (database: { db: DatabaseSync }) => T, +): T | undefined { + try { + const result = withOpenClawAgentDatabaseReadOnly(operation, { + agentId: normalizeAgentId(agentId), + ...(databasePath ? { path: databasePath } : {}), + }); + return result.found ? result.value : undefined; + } catch (error) { + if (!isTransientSqliteError(error)) { + throw error; + } + // Usage rollups are rebuildable cache; stale or empty data beats failing the dashboard. + return undefined; + } } function readCacheValue( @@ -39,18 +52,21 @@ function readCacheValue( key: string, databasePath?: string, ): string | null { - const database = openCacheDatabase(agentId, databasePath); - const kysely = getNodeSqliteKysely(database.db); - const row = executeSqliteQuerySync( - database.db, - kysely - .selectFrom("cache_entries") - .select("value_json") - .where("scope", "=", scope) - .where("key", "=", key) - .limit(1), - ).rows[0]; - return row?.value_json ?? null; + return ( + readCacheDatabase(agentId, databasePath, (database) => { + const kysely = getNodeSqliteKysely(database.db); + const row = executeSqliteQuerySync( + database.db, + kysely + .selectFrom("cache_entries") + .select("value_json") + .where("scope", "=", scope) + .where("key", "=", key) + .limit(1), + ).rows[0]; + return row?.value_json ?? null; + }) ?? null + ); } function deleteCacheValueIfUnchanged(params: { @@ -84,18 +100,21 @@ export function readSessionCostUsageRollupRows( agentId?: string, databasePath?: string, ): SessionCostUsageRollupRow[] { - const database = openCacheDatabase(agentId, databasePath); - const kysely = getNodeSqliteKysely(database.db); - return executeSqliteQuerySync( - database.db, - kysely - .selectFrom("cache_entries") - .select(["key", "value_json", "updated_at"]) - .where("scope", "=", ROLLUP_SCOPE), - ).rows.flatMap((row) => - row.value_json === null - ? [] - : [{ key: row.key, valueJson: row.value_json, updatedAt: row.updated_at }], + return ( + readCacheDatabase(agentId, databasePath, (database) => { + const kysely = getNodeSqliteKysely(database.db); + return executeSqliteQuerySync( + database.db, + kysely + .selectFrom("cache_entries") + .select(["key", "value_json", "updated_at"]) + .where("scope", "=", ROLLUP_SCOPE), + ).rows.flatMap((row) => + row.value_json === null + ? [] + : [{ key: row.key, valueJson: row.value_json, updatedAt: row.updated_at }], + ); + }) ?? [] ); } diff --git a/src/infra/session-cost-usage-refresh.test.ts b/src/infra/session-cost-usage-refresh.test.ts new file mode 100644 index 00000000000..fcb26a3e0a2 --- /dev/null +++ b/src/infra/session-cost-usage-refresh.test.ts @@ -0,0 +1,69 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { testing as testApi } from "./session-cost-usage.test-support.js"; + +describe("session cost usage refresh backoff", () => { + beforeEach(() => { + vi.useFakeTimers(); + testApi.clearUsageCostRefreshesForTest(); + }); + + afterEach(() => { + testApi.clearUsageCostRefreshesForTest(); + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("doubles consecutive busy delays, caps them, and resets after success", async () => { + let calls = 0; + const refresh = vi + .spyOn(testApi.usageCostRefreshRuntime, "refreshCostUsageCacheForAgent") + .mockImplementation(async () => { + calls += 1; + if (calls <= 10) { + return "busy"; + } + if (calls === 11) { + testApi.requestCostUsageCacheRefresh({ + agentId: "backoff-test", + sessionFiles: ["next-session.jsonl"], + }); + return "refreshed"; + } + if (calls === 12) { + return "busy"; + } + return "refreshed"; + }); + + testApi.requestCostUsageCacheRefresh({ agentId: "backoff-test" }); + await vi.advanceTimersByTimeAsync(0); + expect(refresh).toHaveBeenCalledTimes(1); + + for (const [delayMs, expectedCalls] of [ + [50, 2], + [100, 3], + [200, 4], + [400, 5], + [800, 6], + [1_600, 7], + [3_200, 8], + [5_000, 9], + [5_000, 10], + ] as const) { + await vi.advanceTimersByTimeAsync(delayMs - 1); + expect(refresh).toHaveBeenCalledTimes(expectedCalls - 1); + await vi.advanceTimersByTimeAsync(1); + expect(refresh).toHaveBeenCalledTimes(expectedCalls); + } + + await vi.advanceTimersByTimeAsync(4_999); + expect(refresh).toHaveBeenCalledTimes(10); + await vi.advanceTimersByTimeAsync(1); + expect(refresh).toHaveBeenCalledTimes(12); + + await vi.advanceTimersByTimeAsync(49); + expect(refresh).toHaveBeenCalledTimes(12); + await vi.advanceTimersByTimeAsync(1); + expect(refresh).toHaveBeenCalledTimes(13); + }); +}); diff --git a/src/infra/session-cost-usage.test-support.ts b/src/infra/session-cost-usage.test-support.ts new file mode 100644 index 00000000000..4a983960aa7 --- /dev/null +++ b/src/infra/session-cost-usage.test-support.ts @@ -0,0 +1,32 @@ +import "./session-cost-usage.js"; + +type UsageCostRefreshParams = { + agentId?: string; + sessionFiles?: string[]; +}; + +type SessionCostUsageTestApi = { + clearUsageCostRefreshesForTest(): void; + requestCostUsageCacheRefresh(params?: UsageCostRefreshParams): void; + usageCostRefreshRuntime: { + refreshCostUsageCacheForAgent(params?: UsageCostRefreshParams): Promise<"busy" | "refreshed">; + }; +}; + +function getTestApi(): SessionCostUsageTestApi { + return (globalThis as Record)[ + Symbol.for("openclaw.sessionCostUsageTestApi") + ] as SessionCostUsageTestApi; +} + +export const testing: SessionCostUsageTestApi = { + clearUsageCostRefreshesForTest() { + getTestApi().clearUsageCostRefreshesForTest(); + }, + requestCostUsageCacheRefresh(params) { + getTestApi().requestCostUsageCacheRefresh(params); + }, + get usageCostRefreshRuntime() { + return getTestApi().usageCostRefreshRuntime; + }, +}; diff --git a/src/infra/session-cost-usage.ts b/src/infra/session-cost-usage.ts index db78226fad9..66e21c6cca5 100644 --- a/src/infra/session-cost-usage.ts +++ b/src/infra/session-cost-usage.ts @@ -112,6 +112,8 @@ const USAGE_COST_ROLLUP_VERSION = 2; const USAGE_COST_TRANSCRIPT_STAT_CONCURRENCY = 32; const USAGE_COST_FILE_ANCHOR_BYTES = 4096; const USAGE_COST_DIRECT_REFRESH_RETRY_MS = 25; +const USAGE_COST_REFRESH_RETRY_MIN_MS = 50; +const USAGE_COST_REFRESH_RETRY_MAX_MS = 5_000; const logger = createSubsystemLogger("usage-cost-cache"); type UsageCostRefreshState = { @@ -122,6 +124,7 @@ type UsageCostRefreshState = { pendingSessionFiles: Set; running: boolean; sessionsDir: string; + busyRetryDelayMs: number; timer?: ReturnType; }; @@ -1618,6 +1621,8 @@ async function refreshCostUsageCacheForAgent(params?: { } } +const usageCostRefreshRuntime = { refreshCostUsageCacheForAgent }; + async function refreshCostUsageCache(params?: { config?: OpenClawConfig; agentId?: string; @@ -1768,6 +1773,7 @@ function requestCostUsageCacheRefresh(params?: { pendingSessionFiles: new Set(), running: false, sessionsDir: resolveSessionTranscriptsDirForAgent(params?.agentId), + busyRetryDelayMs: USAGE_COST_REFRESH_RETRY_MIN_MS, }; mergeUsageCostRefreshRequest(state, params); usageCostRefreshes.set(refreshKey, state); @@ -1827,7 +1833,7 @@ async function runQueuedUsageCostRefresh( state.pendingSessionFiles.clear(); } state.fullRefreshRequested = false; - const result = await refreshCostUsageCacheForAgent({ + const result = await usageCostRefreshRuntime.refreshCostUsageCacheForAgent({ config: state.config, agentId: state.agentId, databasePath: state.databasePath, @@ -1842,9 +1848,15 @@ async function runQueuedUsageCostRefresh( state.pendingSessionFiles.add(sessionFile); } } - retryDelayMs = 50; + retryDelayMs = state.busyRetryDelayMs; + // Contention among many per-agent refreshes must degrade to polling, not a 20Hz spin. + state.busyRetryDelayMs = Math.min( + state.busyRetryDelayMs * 2, + USAGE_COST_REFRESH_RETRY_MAX_MS, + ); break; } + state.busyRetryDelayMs = USAGE_COST_REFRESH_RETRY_MIN_MS; } } catch (error) { logger.warn(`background refresh failed: ${formatErrorMessage(error)}`, { error }); @@ -1858,6 +1870,23 @@ async function runQueuedUsageCostRefresh( } } +function clearUsageCostRefreshesForTest(): void { + for (const state of usageCostRefreshes.values()) { + if (state.timer) { + clearTimeout(state.timer); + } + } + usageCostRefreshes.clear(); +} + +if (process.env.VITEST || process.env.NODE_ENV === "test") { + (globalThis as Record)[Symbol.for("openclaw.sessionCostUsageTestApi")] = { + requestCostUsageCacheRefresh, + usageCostRefreshRuntime, + clearUsageCostRefreshesForTest, + }; +} + /** * Scan all transcript files to discover sessions not in the session store. * Returns basic metadata for each discovered session. diff --git a/src/infra/sqlite-wal.ts b/src/infra/sqlite-wal.ts index 8a3871746fc..0eb72a0f315 100644 --- a/src/infra/sqlite-wal.ts +++ b/src/infra/sqlite-wal.ts @@ -34,7 +34,7 @@ type MountEntry = { mountPoint: string; fsType: string; source?: string }; export type SqliteWalMaintenance = { checkpoint: () => boolean; - close: () => boolean; + close: (options?: { checkpointMode?: SqliteWalCheckpointMode }) => boolean; }; /** Options controlling WAL autocheckpoint and periodic checkpoint behavior. */ @@ -508,12 +508,15 @@ export function configureSqliteWalMaintenance( return { checkpoint, - close: () => { + close: (closeOptions) => { if (timer) { clearInterval(timer); timer = null; } - return checkpoint(); + // Cache eviction passes PASSIVE: a TRUNCATE close-checkpoint waits on + // readers and has starved the event loop for seconds under fleet churn. + // Orderly dispose/delete keeps TRUNCATE so sidecars are flushed for unlink. + return runCheckpoint(closeOptions?.checkpointMode ?? checkpointMode); }, }; } diff --git a/src/plugin-sdk/session-store-runtime.ts b/src/plugin-sdk/session-store-runtime.ts index 3feb3d88d8e..4791a2c2bdf 100644 --- a/src/plugin-sdk/session-store-runtime.ts +++ b/src/plugin-sdk/session-store-runtime.ts @@ -16,6 +16,7 @@ import { deleteSessionEntryLifecycle as deleteAccessorSessionEntryLifecycle, loadTranscriptEventsSync as loadAccessorTranscriptEventsSync, listSessionEntries as listAccessorSessionEntries, + listSessionEntriesReadOnly as listAccessorSessionEntriesReadOnly, loadSessionEntry, patchSessionEntry as patchAccessorSessionEntry, readSessionUpdatedAt as readAccessorSessionUpdatedAt, @@ -360,11 +361,17 @@ export function getSessionEntry(params: SessionStoreReadParams): SessionEntry | return entry ? projectPluginSessionEntry(entry) : undefined; } -/** Lists session entries for one agent. */ +/** + * Lists session entries for one agent. `readOnly` reads without joining the + * agent database writable lifecycle (no create/register/migrate) — required + * for detection/introspection paths that may run across the whole fleet. + * One flagged entry instead of a second export keeps the SDK surface budget flat. + */ export function listSessionEntries( - params: SessionStoreListParams = {}, + params: SessionStoreListParams & { readOnly?: boolean } = {}, ): SessionStoreEntrySummary[] { - return listAccessorSessionEntries({ + const list = params.readOnly ? listAccessorSessionEntriesReadOnly : listAccessorSessionEntries; + return list({ ...(params.agentId !== undefined ? { agentId: params.agentId } : {}), ...(params.env !== undefined ? { env: params.env } : {}), ...(params.hydrateSkillPromptRefs !== undefined diff --git a/src/state/openclaw-agent-db.cache-eviction.test.ts b/src/state/openclaw-agent-db.cache-eviction.test.ts new file mode 100644 index 00000000000..668973eb7da --- /dev/null +++ b/src/state/openclaw-agent-db.cache-eviction.test.ts @@ -0,0 +1,180 @@ +// Agent database cache tests cover bounded process-local SQLite handle ownership. +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + closeOpenClawAgentDatabasesForTest, + disposeOpenClawAgentDatabaseByPath, + isOpenClawAgentDatabaseOpen, + listOpenClawRegisteredAgentDatabases, + OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP, + openOpenClawAgentDatabase, +} from "./openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "./openclaw-state-db.js"; + +const tempStateDirs: string[] = []; + +function createTempStateDir(): string { + const stateDir = fs.realpathSync( + fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-agent-db-cache-")), + ); + tempStateDirs.push(stateDir); + return stateDir; +} + +function fillAgentDatabaseCache(env: NodeJS.ProcessEnv, prefix: string): void { + for (let index = 0; index < OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP; index += 1) { + openOpenClawAgentDatabase({ agentId: `${prefix}-${index}`, env }); + } +} + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + for (const stateDir of tempStateDirs.splice(0)) { + fs.rmSync(stateDir, { force: true, recursive: true }); + } +}); + +describe("openclaw agent database handle cache", () => { + it("keeps only the capped number of open handles", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const databases = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP + 1 }, (_, index) => + openOpenClawAgentDatabase({ agentId: `worker-${index}`, env }), + ); + const leastRecentlyUsed = databases[0]!; + + expect(databases.filter((database) => database.db.isOpen)).toHaveLength( + OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP, + ); + expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false); + expect(leastRecentlyUsed.db.isOpen).toBe(false); + }); + + it("refreshes cache-hit recency before evicting the true LRU handle", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const recentlyUsed = openOpenClawAgentDatabase({ agentId: "recently-used", env }); + const untouched = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP - 1 }, (_, index) => + openOpenClawAgentDatabase({ agentId: `untouched-${index}`, env }), + ); + const leastRecentlyUsed = untouched[0]!; + + expect(openOpenClawAgentDatabase({ agentId: "recently-used", env })).toBe(recentlyUsed); + openOpenClawAgentDatabase({ agentId: "newest", env }); + + expect(recentlyUsed.db.isOpen).toBe(true); + expect(isOpenClawAgentDatabaseOpen(recentlyUsed.path)).toBe(true); + expect(leastRecentlyUsed.db.isOpen).toBe(false); + expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false); + }); + + it("never evicts an LRU handle with an open transaction", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const transactionOwner = openOpenClawAgentDatabase({ agentId: "transaction-owner", env }); + transactionOwner.db.exec("BEGIN IMMEDIATE"); + try { + const untouched = Array.from({ length: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP - 1 }, (_, index) => + openOpenClawAgentDatabase({ agentId: `untouched-${index}`, env }), + ); + const leastRecentlyUsed = untouched[0]!; + openOpenClawAgentDatabase({ agentId: "newest", env }); + + expect(transactionOwner.db.isOpen).toBe(true); + expect(transactionOwner.db.isTransaction).toBe(true); + expect(isOpenClawAgentDatabaseOpen(transactionOwner.path)).toBe(true); + expect(leastRecentlyUsed.db.isOpen).toBe(false); + expect(isOpenClawAgentDatabaseOpen(leastRecentlyUsed.path)).toBe(false); + } finally { + transactionOwner.db.exec("ROLLBACK"); + } + }); + + it("reopens an evicted database without losing durable rows", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const evicted = openOpenClawAgentDatabase({ agentId: "evicted", env }); + evicted.db + .prepare( + "INSERT INTO auth_profile_state (state_key, state_json, updated_at) VALUES (?, ?, ?)", + ) + .run("cache-eviction", JSON.stringify({ preserved: true }), 42); + + fillAgentDatabaseCache(env, "filler"); + expect(evicted.db.isOpen).toBe(false); + + const reopened = openOpenClawAgentDatabase({ agentId: "evicted", env }); + expect(reopened).not.toBe(evicted); + expect( + reopened.db + .prepare("SELECT state_json, updated_at FROM auth_profile_state WHERE state_key = ?") + .get("cache-eviction"), + ).toEqual({ state_json: JSON.stringify({ preserved: true }), updated_at: 42 }); + }); + + it("registers a first open without refreshing registry metadata after eviction", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000); + try { + const evicted = openOpenClawAgentDatabase({ agentId: "evicted", env }); + expect( + listOpenClawRegisteredAgentDatabases({ env }).find( + (entry) => entry.agentId === "evicted" && entry.path === evicted.path, + ), + ).toMatchObject({ lastSeenAt: 1_000 }); + + nowSpy.mockReturnValue(2_000); + fillAgentDatabaseCache(env, "registry-filler"); + expect(evicted.db.isOpen).toBe(false); + + openOpenClawAgentDatabase({ agentId: "evicted", env }); + expect( + listOpenClawRegisteredAgentDatabases({ env }).find( + (entry) => entry.agentId === "evicted" && entry.path === evicted.path, + ), + ).toMatchObject({ lastSeenAt: 1_000 }); + } finally { + nowSpy.mockRestore(); + } + }); + + it("validates ownership when an evicted path is requested for another agent", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const evicted = openOpenClawAgentDatabase({ agentId: "worker-a", env }); + fillAgentDatabaseCache(env, "ownership-filler"); + expect(evicted.db.isOpen).toBe(false); + + expect(() => + openOpenClawAgentDatabase({ agentId: "worker-b", env, path: evicted.path }), + ).toThrow(/belongs to agent worker-a/); + }); + + it("revalidates and registers a database after explicit disposal", () => { + const env = { OPENCLAW_STATE_DIR: createTempStateDir() }; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_000); + try { + const disposed = openOpenClawAgentDatabase({ agentId: "disposed", env }); + expect( + listOpenClawRegisteredAgentDatabases({ env }).find( + (entry) => entry.agentId === "disposed" && entry.path === disposed.path, + ), + ).toMatchObject({ lastSeenAt: 1_000 }); + + expect(disposeOpenClawAgentDatabaseByPath(disposed.path, { env })).toBe(true); + expect( + listOpenClawRegisteredAgentDatabases({ env }).some( + (entry) => entry.agentId === "disposed" && entry.path === disposed.path, + ), + ).toBe(false); + + nowSpy.mockReturnValue(2_000); + openOpenClawAgentDatabase({ agentId: "disposed", env, path: disposed.path }); + expect( + listOpenClawRegisteredAgentDatabases({ env }).find( + (entry) => entry.agentId === "disposed" && entry.path === disposed.path, + ), + ).toMatchObject({ lastSeenAt: 2_000 }); + } finally { + nowSpy.mockRestore(); + } + }); +}); diff --git a/src/state/openclaw-agent-db.ts b/src/state/openclaw-agent-db.ts index 6ede759a646..b1499331c19 100644 --- a/src/state/openclaw-agent-db.ts +++ b/src/state/openclaw-agent-db.ts @@ -70,14 +70,22 @@ export { resolveOpenClawAgentSqlitePath } from "./openclaw-agent-db.paths.js"; * OpenClaw state database for discovery and maintenance. */ const OPENCLAW_AGENT_DB_SLOW_OPEN_MS = 1_000; +// Each WAL database consumes roughly three file descriptors, so the fixed cap +// satisfies the bounded-cache policy within a predictable FD budget, without config. +export const OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP = 64; const agentDbLog = createSubsystemLogger("state/agent-db"); const cachedDatabases = new Map(); +// External schema changes under a live process are unsupported: doctor migrations +// require restart, so successful owner/schema validation is process-stable. +const validatedAgentDatabasePaths = new Map(); const terminalOpenLatch = createSqliteTerminalOpenLatch({ closeByPath: closeOpenClawAgentDatabaseByPath, }); /** Latch background verification damage so later opens fail without rescanning. */ export function recordOpenClawAgentDatabaseOpenFailure(pathname: string, error: Error): void { + // Quarantine revokes this process's trust because doctor may replace the file. + validatedAgentDatabasePaths.delete(path.resolve(pathname)); terminalOpenLatch.record(pathname, error); } @@ -153,6 +161,8 @@ export function openOpenClawAgentDatabase( `OpenClaw agent database ${pathname} is already open for agent ${cached.agentId}; requested agent ${agentId}.`, ); } + cachedDatabases.delete(pathname); + cachedDatabases.set(pathname, cached); return cached; } // Latched paths are quarantined; every fresh open fails fast here until @@ -187,14 +197,24 @@ export function openOpenClawAgentDatabase( } const openStartedAt = Date.now(); ensureOpenClawAgentDatabasePermissions(pathname, databaseOptions); + // Free a slot before constructing the new handle: under real descriptor + // pressure the 65th open would otherwise fail before eviction could run. + evictLruAgentDatabaseHandles(); const sqlite = requireNodeSqlite(); const db = new sqlite.DatabaseSync(pathname); + // Eviction churn must avoid schema/registry busy waits on the event loop while + // reconcile workers hold write transactions on these same agent databases. + const isValidatedReopen = validatedAgentDatabasePaths.get(pathname) === agentId; const walMaintenance = (() => { let maintenance: OpenClawAgentDatabase["walMaintenance"] | undefined; try { db.exec(`PRAGMA busy_timeout = ${OPENCLAW_SQLITE_BUSY_TIMEOUT_MS};`); - assertSupportedAgentSchemaVersion(db, pathname); - assertExistingAgentSchemaOwner(readExistingAgentSchemaMeta(db), agentId, pathname); + if (!isValidatedReopen) { + assertSupportedAgentSchemaVersion(db, pathname); + assertExistingAgentSchemaOwner(readExistingAgentSchemaMeta(db), agentId, pathname); + } + // Integrity is not process-stable: the file can be damaged while evicted. + // This guard is read-only (no busy waits), so every physical open pays it. assertAgentDatabaseIntegrityBeforeMutation(db, pathname); configureSqlitePreSchemaPragmas(db, { busyTimeoutMs: OPENCLAW_SQLITE_BUSY_TIMEOUT_MS, @@ -206,7 +226,9 @@ export function openOpenClawAgentDatabase( foreignKeys: true, synchronous: "NORMAL", }); - ensureOpenClawAgentSchema(db, agentId, pathname); + if (!isValidatedReopen) { + ensureOpenClawAgentSchema(db, agentId, pathname); + } return maintenance; } catch (err) { maintenance?.close(); @@ -222,11 +244,14 @@ export function openOpenClawAgentDatabase( })(); ensureOpenClawAgentDatabasePermissions(pathname, databaseOptions); const database = { agentId, db, path: pathname, walMaintenance }; - try { - registerOpenClawAgentDatabase({ agentId, path: pathname, env: options.env }); - } catch (error) { - closeCachedOpenClawAgentDatabase(database); - throw error; + if (!isValidatedReopen) { + try { + registerOpenClawAgentDatabase({ agentId, path: pathname, env: options.env }); + } catch (error) { + closeCachedOpenClawAgentDatabase(database); + throw error; + } + validatedAgentDatabasePaths.set(pathname, agentId); } cachedDatabases.set(pathname, database); terminalOpenLatch.clear(pathname); @@ -312,14 +337,59 @@ export function runOpenClawAgentWriteTransaction( let unregisterExitClose: (() => void) | null = null; -function closeCachedOpenClawAgentDatabase(database: OpenClawAgentDatabase): void { - database.walMaintenance.close(); +function closeCachedOpenClawAgentDatabase( + database: OpenClawAgentDatabase, + options: { eviction?: boolean } = {}, +): void { + // Eviction must stay cheap: PASSIVE skips waiting on concurrent readers, + // whose drained TRUNCATE checkpoints blocked the event loop for seconds. + database.walMaintenance.close(options.eviction ? { checkpointMode: "PASSIVE" } : undefined); clearNodeSqliteKyselyCacheForDatabase(database.db); if (database.db.isOpen) { database.db.close(); } } +function evictLruAgentDatabaseHandles(): void { + // Callers re-fetch handles from this cache at each operation entry and use + // them within one synchronous section, so eviction can never close a handle + // mid-use; a handle retained across an eviction-triggering open goes stale. + while (cachedDatabases.size >= OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP) { + let evicted = false; + for (const [pathname, database] of cachedDatabases) { + // A synchronous transaction owns its handle through COMMIT or ROLLBACK; + // closing it here would violate the transaction commit-section contract. + if (database.db.isTransaction) { + continue; + } + // Registry rows are durable discovery metadata; only explicit disposal + // unregisters them, while eviction closes this process-local handle. + closeCachedOpenClawAgentDatabase(database, { eviction: true }); + cachedDatabases.delete(pathname); + agentDbLog.debug("evicted OpenClaw agent database handle", { + agentId: database.agentId, + openHandles: cachedDatabases.size, + path: pathname, + }); + evicted = true; + break; + } + if (!evicted) { + // Every handle is mid-transaction: sync commit sections bound concurrent + // transactions at call-nesting depth, so this stays a pathological safety + // valve; exceeding the cap beats failing an unrelated agent's open. + agentDbLog.warn( + "agent database handle cap exceeded; all cached handles are in transactions", + { + cap: OPENCLAW_AGENT_DB_OPEN_HANDLE_CAP, + openHandles: cachedDatabases.size, + }, + ); + return; + } + } +} + /** Return whether the exact cached agent database pathname is still open. */ export function isOpenClawAgentDatabaseOpen(pathname: string): boolean { const database = cachedDatabases.get(path.resolve(pathname)); @@ -352,6 +422,8 @@ export function disposeOpenClawAgentDatabaseByPath( // Require the cache's exact lexical owner. Following a symlink or accepting // an uncached path could unregister a database another process now owns. const resolvedPath = path.resolve(pathname); + // Disposal can be followed by file deletion or recreation, so revalidate next open. + validatedAgentDatabasePaths.delete(resolvedPath); const database = cachedDatabases.get(resolvedPath); if (!database || database.path !== resolvedPath) { return false; @@ -383,5 +455,6 @@ export function closeOpenClawAgentDatabases(): void { /** Close cached agent handles and clear terminal failure latches for test isolation. */ export function closeOpenClawAgentDatabasesForTest(): void { closeOpenClawAgentDatabases(); + validatedAgentDatabasePaths.clear(); terminalOpenLatch.clearAll(); }