diff --git a/scripts/deadcode-exports.baseline.mjs b/scripts/deadcode-exports.baseline.mjs index 6ec3a6bc12df..e2ba15eea184 100644 --- a/scripts/deadcode-exports.baseline.mjs +++ b/scripts/deadcode-exports.baseline.mjs @@ -3509,6 +3509,7 @@ export const KNIP_UNUSED_EXPORT_BASELINE = [ "src/infra/runtime-guard.ts: parseMinimumNodeEngine", "src/infra/runtime-guard.ts: RuntimeDetails", "src/infra/runtime-guard.ts: runtimeSatisfies", + "src/infra/session-cost-usage-cache.sqlite.ts: sessionCostUsageCacheTestApi", "src/infra/session-cost-usage.ts: loadSessionCostSummaryFromCache", "src/infra/session-cost-usage.ts: refreshCostUsageCache", "src/infra/session-cost-usage.ts: requestCostUsageCacheRefresh", diff --git a/scripts/ts-max-loc-baseline-v2.json b/scripts/ts-max-loc-baseline-v2.json index da28bb5e8a1c..45549761508b 100644 --- a/scripts/ts-max-loc-baseline-v2.json +++ b/scripts/ts-max-loc-baseline-v2.json @@ -994,7 +994,7 @@ "src/infra/push-apns.ts": 1202, "src/infra/restart-stale-pids.ts": 673, "src/infra/restart.ts": 1261, - "src/infra/session-cost-usage.ts": 3036, + "src/infra/session-cost-usage.ts": 2873, "src/infra/sqlite-schema-contract.ts": 541, "src/infra/sqlite-snapshot.ts": 864, "src/infra/state-migrations.debug-proxy.ts": 578, diff --git a/src/commands/doctor-usage-cost-cache.test.ts b/src/commands/doctor-usage-cost-cache.test.ts new file mode 100644 index 000000000000..d4baa96d03f4 --- /dev/null +++ b/src/commands/doctor-usage-cost-cache.test.ts @@ -0,0 +1,58 @@ +// Covers doctor cleanup of legacy usage-cost cache sidecars. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { maybeRepairLegacyRuntimeFiles } from "./doctor-usage-cost-cache.js"; + +let root: string | undefined; + +afterEach(async () => { + if (root) { + await fs.rm(root, { recursive: true, force: true }); + root = undefined; + } +}); + +describe("legacy usage-cost cache cleanup", () => { + it("removes only rebuildable usage-cost cache sidecars", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-usage-cost-doctor-")); + const sessionsDir = path.join(root, "agents", "main", "sessions"); + await fs.mkdir(sessionsDir, { recursive: true }); + const cacheFiles = [ + path.join(sessionsDir, ".usage-cost-cache.json"), + path.join(sessionsDir, ".usage-cost-cache.json.lock"), + ]; + const staleTempFiles = [ + path.join(sessionsDir, ".usage-cost-cache.123.3b241101-e2bb-4255-8caf-4136c566a962.tmp"), + path.join(sessionsDir, ".usage-cost-cache.json.lock.123.456.tmp"), + path.join(sessionsDir, ".usage-cost-cache.json.123.tmp"), + path.join(sessionsDir, ".usage-cost-cache.123.tmp"), + path.join(sessionsDir, ".usage-cost-cache.json.lock.123.tmp"), + ]; + const recentTemp = path.join(sessionsDir, ".usage-cost-cache.456.tmp"); + const transcript = path.join(sessionsDir, "session.jsonl"); + const unrelatedFiles = [ + transcript, + path.join(sessionsDir, ".usage-cost-cache.backup"), + path.join(sessionsDir, ".usage-cost-cache.notes"), + ]; + await Promise.all( + [...cacheFiles, ...staleTempFiles, recentTemp, ...unrelatedFiles].map((filePath) => + fs.writeFile(filePath, "x"), + ), + ); + const staleTime = new Date(Date.now() - 60_000); + await Promise.all(staleTempFiles.map((filePath) => fs.utimes(filePath, staleTime, staleTime))); + const env = { OPENCLAW_STATE_DIR: root } as NodeJS.ProcessEnv; + + await maybeRepairLegacyRuntimeFiles(true, env); + + for (const filePath of [...cacheFiles, ...staleTempFiles]) { + await expect(fs.readFile(filePath, "utf-8")).rejects.toMatchObject({ code: "ENOENT" }); + } + for (const filePath of [recentTemp, ...unrelatedFiles]) { + await expect(fs.readFile(filePath, "utf-8")).resolves.toBe("x"); + } + }); +}); diff --git a/src/commands/doctor-usage-cost-cache.ts b/src/commands/doctor-usage-cost-cache.ts new file mode 100644 index 000000000000..9b6ee0248667 --- /dev/null +++ b/src/commands/doctor-usage-cost-cache.ts @@ -0,0 +1,98 @@ +/** Doctor cleanup for rebuildable legacy usage-cost cache sidecars. */ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { note } from "../../packages/terminal-core/src/note.js"; +import { resolveStateDir } from "../config/paths.js"; +import { maybeScrubConfigAuditLog } from "./doctor-config-audit-scrub.js"; + +const LEGACY_USAGE_COST_TEMP_GRACE_MS = 10_000; + +function isLegacyUsageCostCacheTempName(name: string): boolean { + return ( + /^\.usage-cost-cache\.\d+\.[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.tmp$/u.test( + name, + ) || + /^\.usage-cost-cache(?:\.json)?\.\d+\.tmp$/u.test(name) || + /^\.usage-cost-cache\.json\.lock\.\d+(?:\.\d+)?\.tmp$/u.test(name) + ); +} + +async function detectLegacyUsageCostCacheFiles(params?: { + env?: NodeJS.ProcessEnv; + homedir?: () => string; +}): Promise { + const stateDir = resolveStateDir(params?.env ?? process.env, params?.homedir ?? os.homedir); + const sessionDirs = [path.join(stateDir, "sessions")]; + const agentsDir = path.join(stateDir, "agents"); + const agentEntries = await fs.readdir(agentsDir, { withFileTypes: true }).catch(() => []); + for (const entry of agentEntries) { + if (entry.isDirectory()) { + sessionDirs.push(path.join(agentsDir, entry.name, "sessions")); + } + } + const files: string[] = []; + for (const sessionDir of sessionDirs) { + const entries = await fs.readdir(sessionDir, { withFileTypes: true }).catch(() => []); + for (const entry of entries) { + if (!entry.isFile()) { + continue; + } + const filePath = path.join(sessionDir, entry.name); + if (entry.name === ".usage-cost-cache.json" || entry.name === ".usage-cost-cache.json.lock") { + files.push(filePath); + continue; + } + if (isLegacyUsageCostCacheTempName(entry.name)) { + const stats = await fs.stat(filePath).catch(() => null); + if (stats && Date.now() - stats.mtimeMs >= LEGACY_USAGE_COST_TEMP_GRACE_MS) { + files.push(filePath); + } + } + } + } + return files.toSorted(); +} + +async function maybeRemoveLegacyUsageCostCacheFiles(params: { + shouldRepair: boolean; + env?: NodeJS.ProcessEnv; + homedir?: () => string; +}): Promise { + const files = await detectLegacyUsageCostCacheFiles(params); + if (files.length === 0) { + return; + } + if (!params.shouldRepair) { + note( + `${files.length} rebuildable usage-cost cache ${files.length === 1 ? "file remains" : "files remain"}. Run \`openclaw doctor --fix\` to remove ${files.length === 1 ? "it" : "them"}.`, + "Usage cost cache", + ); + return; + } + const failures: string[] = []; + for (const filePath of files) { + await fs.rm(filePath, { force: true }).catch((error: unknown) => { + failures.push(`${filePath}: ${String(error)}`); + }); + } + if (failures.length > 0) { + note( + `Failed removing legacy usage-cost cache files:\n${failures.join("\n")}`, + "Usage cost cache", + ); + return; + } + note( + `Removed ${files.length} rebuildable legacy usage-cost cache ${files.length === 1 ? "file" : "files"}; SQLite rebuilds the cache on demand.`, + "Usage cost cache", + ); +} + +export async function maybeRepairLegacyRuntimeFiles( + shouldRepair: boolean, + env?: NodeJS.ProcessEnv, +): Promise { + await maybeScrubConfigAuditLog({ shouldRepair, env }); + await maybeRemoveLegacyUsageCostCacheFiles({ shouldRepair, env }); +} diff --git a/src/commands/doctor.fast-path-mocks.ts b/src/commands/doctor.fast-path-mocks.ts index ceb0affe7322..201fcda6d10a 100644 --- a/src/commands/doctor.fast-path-mocks.ts +++ b/src/commands/doctor.fast-path-mocks.ts @@ -49,6 +49,10 @@ vi.mock("./doctor-config-audit-scrub.js", () => ({ maybeScrubConfigAuditLog: vi.fn().mockResolvedValue(undefined), })); +vi.mock("./doctor-usage-cost-cache.js", () => ({ + maybeRepairLegacyRuntimeFiles: vi.fn().mockResolvedValue(undefined), +})); + vi.mock("./doctor/cron/index.js", () => ({ maybeRepairLegacyCronStore: vi.fn().mockResolvedValue(undefined), noteLegacyWhatsAppCrontabHealthCheck: vi.fn().mockResolvedValue(undefined), diff --git a/src/commitments/runtime.test.ts b/src/commitments/runtime.test.ts index 59cb6769ae92..6379fe0e7dfd 100644 --- a/src/commitments/runtime.test.ts +++ b/src/commitments/runtime.test.ts @@ -31,6 +31,7 @@ function requireFirstEmbeddedAgentRequest(): { provider?: string; model?: string; disableTools?: boolean; + sessionFile?: string; } { const [call] = runEmbeddedAgentMock.mock.calls; if (!call) { @@ -40,7 +41,12 @@ function requireFirstEmbeddedAgentRequest(): { if (!request || typeof request !== "object" || Array.isArray(request)) { throw new Error("expected embedded OpenClaw agent extraction request"); } - return request as { provider?: string; model?: string; disableTools?: boolean }; + return request as { + provider?: string; + model?: string; + disableTools?: boolean; + sessionFile?: string; + }; } describe("commitment extraction runtime", () => { @@ -268,6 +274,7 @@ describe("commitment extraction runtime", () => { expect(request.provider).toBe("openai"); expect(request.model).toBe("gpt-5.5"); expect(request.disableTools).toBe(true); + expect(request.sessionFile).toBeUndefined(); }); it("backs off hidden extraction after terminal model or auth failures", async () => { diff --git a/src/commitments/runtime.ts b/src/commitments/runtime.ts index 9d779724d51b..6b03841731fc 100644 --- a/src/commitments/runtime.ts +++ b/src/commitments/runtime.ts @@ -1,11 +1,9 @@ // Runs commitment extraction, scheduling, and follow-up lifecycle work. import { randomUUID } from "node:crypto"; -import path from "node:path"; import { resolveExpiresAtMsFromDurationMs } from "@openclaw/normalization-core/number-coercion"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { resolveAgentWorkspaceDir } from "../agents/agent-scope.js"; import type { OpenClawConfig } from "../config/config.js"; -import { resolveStateDir } from "../config/paths.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { resolveCommitmentTimezone, resolveCommitmentsConfig } from "./config.js"; import { @@ -211,16 +209,6 @@ function openTerminalFailureCooldown( }); } -function resolveExtractionSessionFile(agentId: string, runId: string): string { - return path.join( - resolveStateDir(), - "commitments", - "extractor-sessions", - agentId, - `${runId}.jsonl`, - ); -} - function joinPayloadText(result: EmbeddedAgentPayloadResult): string { return ( result.payloads @@ -260,7 +248,6 @@ async function defaultExtractBatch(params: { sessionKey: `agent:${first.agentId}:commitments:${runId}`, agentId: first.agentId, trigger: "manual", - sessionFile: resolveExtractionSessionFile(first.agentId, runId), workspaceDir: resolveAgentWorkspaceDir(cfg, first.agentId), config: cfg, provider: modelRef.provider, diff --git a/src/flows/doctor-health-contributions.ts b/src/flows/doctor-health-contributions.ts index 4cce511732b6..a3f74b362e12 100644 --- a/src/flows/doctor-health-contributions.ts +++ b/src/flows/doctor-health-contributions.ts @@ -686,8 +686,8 @@ async function runSessionSnapshotsHealth(ctx: DoctorHealthFlowContext): Promise< } async function runConfigAuditScrubHealth(ctx: DoctorHealthFlowContext): Promise { - const { maybeScrubConfigAuditLog } = await import("../commands/doctor-config-audit-scrub.js"); - await maybeScrubConfigAuditLog({ shouldRepair: ctx.prompter.shouldRepair }); + const legacyFiles = await import("../commands/doctor-usage-cost-cache.js"); + await legacyFiles.maybeRepairLegacyRuntimeFiles(ctx.prompter.shouldRepair, ctx.env); } async function runLegacyCronHealth(ctx: DoctorHealthFlowContext): Promise { diff --git a/src/infra/session-cost-usage-cache.sqlite.ts b/src/infra/session-cost-usage-cache.sqlite.ts new file mode 100644 index 000000000000..cbfd446d2250 --- /dev/null +++ b/src/infra/session-cost-usage-cache.sqlite.ts @@ -0,0 +1,279 @@ +import { normalizeAgentId } from "../routing/session-key.js"; +import type { DB as OpenClawAgentKyselyDatabase } from "../state/openclaw-agent-db.generated.js"; +import { + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, +} from "../state/openclaw-agent-db.js"; +// Per-agent SQLite storage for the rebuildable session cost/usage cache. +import { executeSqliteQuerySync, getNodeSqliteKysely } from "./kysely-sync.js"; + +const CACHE_SCOPE = "session-cost-usage"; +const CACHE_KEY = "cache"; +const REFRESH_LOCK_KEY = "refresh-lock"; + +type AgentCacheDatabase = Pick; + +type SessionCostUsageRefreshLock = { + pid: number; + startedAt: number; + ownerNonce: string; +}; + +function readCacheValue( + agentId: string | undefined, + key: string, + databasePath?: string, +): string | null { + const database = openOpenClawAgentDatabase({ + agentId: normalizeAgentId(agentId), + ...(databasePath ? { path: databasePath } : {}), + }); + const kysely = getNodeSqliteKysely(database.db); + const row = executeSqliteQuerySync( + database.db, + kysely + .selectFrom("cache_entries") + .select("value_json") + .where("scope", "=", CACHE_SCOPE) + .where("key", "=", key) + .limit(1), + ).rows[0]; + return row?.value_json ?? null; +} + +function upsertCacheValue(params: { + agentId?: string; + databasePath?: string; + key: string; + valueJson: string; + updatedAt: number; +}): void { + runOpenClawAgentWriteTransaction( + (database) => { + const kysely = getNodeSqliteKysely(database.db); + executeSqliteQuerySync( + database.db, + kysely + .insertInto("cache_entries") + .values({ + scope: CACHE_SCOPE, + key: params.key, + value_json: params.valueJson, + blob: null, + expires_at: null, + updated_at: params.updatedAt, + }) + .onConflict((conflict) => + conflict.columns(["scope", "key"]).doUpdateSet({ + value_json: params.valueJson, + blob: null, + expires_at: null, + updated_at: params.updatedAt, + }), + ), + ); + }, + { + agentId: normalizeAgentId(params.agentId), + ...(params.databasePath ? { path: params.databasePath } : {}), + }, + { operationLabel: `session-cost-usage.${params.key}.write` }, + ); +} + +function deleteCacheValueIfUnchanged(params: { + agentId?: string; + databasePath?: string; + key: string; + valueJson: string; +}): void { + runOpenClawAgentWriteTransaction( + (database) => { + const kysely = getNodeSqliteKysely(database.db); + executeSqliteQuerySync( + database.db, + kysely + .deleteFrom("cache_entries") + .where("scope", "=", CACHE_SCOPE) + .where("key", "=", params.key) + .where("value_json", "=", params.valueJson), + ); + }, + { + agentId: normalizeAgentId(params.agentId), + ...(params.databasePath ? { path: params.databasePath } : {}), + }, + { operationLabel: `session-cost-usage.${params.key}.delete` }, + ); +} + +export function readSessionCostUsageCacheJson( + agentId?: string, + databasePath?: string, +): string | null { + return readCacheValue(agentId, CACHE_KEY, databasePath); +} + +export function writeSessionCostUsageCacheJson(params: { + agentId?: string; + databasePath?: string; + valueJson: string; + updatedAt: number; +}): void { + upsertCacheValue({ ...params, key: CACHE_KEY }); +} + +function parseRefreshLock(raw: string | null): SessionCostUsageRefreshLock | null { + if (!raw) { + return null; + } + try { + const value = JSON.parse(raw) as Partial | null; + if ( + !value || + typeof value.pid !== "number" || + !Number.isInteger(value.pid) || + value.pid <= 0 || + typeof value.startedAt !== "number" || + !Number.isFinite(value.startedAt) || + typeof value.ownerNonce !== "string" || + !value.ownerNonce + ) { + return null; + } + return { pid: value.pid, startedAt: value.startedAt, ownerNonce: value.ownerNonce }; + } catch { + return null; + } +} + +function isProcessRunning(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +export function isSessionCostUsageRefreshRunning(agentId?: string, databasePath?: string): boolean { + const raw = readCacheValue(agentId, REFRESH_LOCK_KEY, databasePath); + const lock = parseRefreshLock(raw); + if (lock && isProcessRunning(lock.pid)) { + return true; + } + if (raw !== null) { + deleteCacheValueIfUnchanged({ + agentId, + databasePath, + key: REFRESH_LOCK_KEY, + valueJson: raw, + }); + } + return false; +} + +export function acquireSessionCostUsageRefreshLock( + agentId?: string, + databasePath?: string, +): { + acquired: boolean; + release: () => void; +} { + const previousRaw = readCacheValue(agentId, REFRESH_LOCK_KEY, databasePath); + const previousLock = parseRefreshLock(previousRaw); + // Process liveness is resolved before BEGIN. The transaction only compares + // the authoritative row and commits the prepared replacement synchronously. + const previousOwnerIsRunning = previousLock ? isProcessRunning(previousLock.pid) : false; + const lock: SessionCostUsageRefreshLock = { + pid: process.pid, + startedAt: Date.now(), + ownerNonce: `${process.pid}:${Date.now()}:${process.hrtime.bigint()}`, + }; + const lockJson = JSON.stringify(lock); + const acquired = runOpenClawAgentWriteTransaction( + (database) => { + const kysely = getNodeSqliteKysely(database.db); + const currentRaw = + executeSqliteQuerySync( + database.db, + kysely + .selectFrom("cache_entries") + .select("value_json") + .where("scope", "=", CACHE_SCOPE) + .where("key", "=", REFRESH_LOCK_KEY) + .limit(1), + ).rows[0]?.value_json ?? null; + if (currentRaw !== previousRaw || previousOwnerIsRunning) { + return false; + } + executeSqliteQuerySync( + database.db, + kysely + .insertInto("cache_entries") + .values({ + scope: CACHE_SCOPE, + key: REFRESH_LOCK_KEY, + value_json: lockJson, + blob: null, + expires_at: null, + updated_at: lock.startedAt, + }) + .onConflict((conflict) => + conflict.columns(["scope", "key"]).doUpdateSet({ + value_json: lockJson, + blob: null, + expires_at: null, + updated_at: lock.startedAt, + }), + ), + ); + return true; + }, + { + agentId: normalizeAgentId(agentId), + ...(databasePath ? { path: databasePath } : {}), + }, + { operationLabel: "session-cost-usage.refresh-lock.acquire" }, + ); + return { + acquired, + release: () => { + if (acquired) { + deleteCacheValueIfUnchanged({ + agentId, + databasePath, + key: REFRESH_LOCK_KEY, + valueJson: lockJson, + }); + } + }, + }; +} + +export const sessionCostUsageCacheTestApi = { + readCacheJson: readSessionCostUsageCacheJson, + writeCacheJson(agentId: string | undefined, valueJson: string): void { + writeSessionCostUsageCacheJson({ agentId, valueJson, updatedAt: Date.now() }); + }, + readRefreshLock(agentId?: string): SessionCostUsageRefreshLock | null { + return parseRefreshLock(readCacheValue(agentId, REFRESH_LOCK_KEY)); + }, + writeRefreshLock(agentId: string | undefined, value: unknown): void { + upsertCacheValue({ + agentId, + key: REFRESH_LOCK_KEY, + valueJson: JSON.stringify(value), + updatedAt: Date.now(), + }); + }, + writeMalformedRefreshLock(agentId: string | undefined, valueJson: string): void { + upsertCacheValue({ agentId, key: REFRESH_LOCK_KEY, valueJson, updatedAt: Date.now() }); + }, + clearRefreshLock(agentId?: string): void { + const raw = readCacheValue(agentId, REFRESH_LOCK_KEY); + if (raw !== null) { + deleteCacheValueIfUnchanged({ agentId, key: REFRESH_LOCK_KEY, valueJson: raw }); + } + }, +}; diff --git a/src/infra/session-cost-usage.stream-errors.test.ts b/src/infra/session-cost-usage.stream-errors.test.ts index fa173ada38a8..0c63aa091532 100644 --- a/src/infra/session-cost-usage.stream-errors.test.ts +++ b/src/infra/session-cost-usage.stream-errors.test.ts @@ -7,6 +7,7 @@ import { PassThrough } from "node:stream"; import { afterEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import { withEnvAsync } from "../test-utils/env.js"; +import { sessionCostUsageCacheTestApi } from "./session-cost-usage-cache.sqlite.js"; import { loadCostUsageSummaryFromCache, loadSessionLogs, @@ -71,8 +72,7 @@ describe("session cost usage stream errors", () => { await withEnvAsync({ OPENCLAW_STATE_DIR: tempDir }, async () => { await refreshCostUsageCache(); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cacheBefore = await fs.readFile(cachePath, "utf-8"); + const cacheBefore = sessionCostUsageCacheTestApi.readCacheJson(); const appendedEntry = `${usageEntry("2026-07-06T12:01:00.000Z", 20)}\n`; await fs.appendFile(sessionFile, appendedEntry, "utf-8"); @@ -86,7 +86,7 @@ describe("session cost usage stream errors", () => { }); await expect(refreshCostUsageCache()).rejects.toThrow("stream read failed"); - expect(await fs.readFile(cachePath, "utf-8")).toBe(cacheBefore); + expect(sessionCostUsageCacheTestApi.readCacheJson()).toBe(cacheBefore); const summary = await loadCostUsageSummaryFromCache({ startMs: Date.UTC(2026, 6, 6), diff --git a/src/infra/session-cost-usage.test.ts b/src/infra/session-cost-usage.test.ts index fb6be18a863a..4f6e7579d57d 100644 --- a/src/infra/session-cost-usage.test.ts +++ b/src/infra/session-cost-usage.test.ts @@ -16,6 +16,7 @@ import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; import { withEnvAsync } from "../test-utils/env.js"; import * as usageFormat from "../utils/usage-format.js"; import * as formatDatetime from "./format-time/format-datetime.js"; +import { sessionCostUsageCacheTestApi } from "./session-cost-usage-cache.sqlite.js"; import { discoverAllSessions, loadCostUsageSummary, @@ -54,6 +55,16 @@ describe("session cost usage", () => { } return value; }; + const readUsageCostCache = (parse: (raw: string) => T = (raw) => JSON.parse(raw) as T): T => { + const raw = sessionCostUsageCacheTestApi.readCacheJson(); + if (!raw) { + throw new Error("expected SQLite usage-cost cache row"); + } + return parse(raw); + }; + const writeUsageCostCache = (value: unknown, agentId?: string): void => { + sessionCostUsageCacheTestApi.writeCacheJson(agentId, JSON.stringify(value)); + }; beforeAll(async () => { await suiteRootTracker.setup(); @@ -804,10 +815,9 @@ describe("session cost usage", () => { // Simulate a durable cache written by a build from before the current cache // semantics: refresh under the current code, then stamp an older version. await refreshCostUsageCache({ sessionFiles: [sessionFile] }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { version: number }; + const cache = readUsageCostCache<{ version: number }>(); cache.version = 6; - await fs.writeFile(cachePath, `${JSON.stringify(cache)}\n`, "utf-8"); + writeUsageCostCache(cache); // The pre-upgrade cache must be treated as stale (not served), forcing a rebuild // under current cost semantics instead of reusing old complete-$0 totals. @@ -1364,10 +1374,9 @@ describe("session cost usage", () => { await withStateDir(root, async () => { await refreshCostUsageCache(); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const beforeCache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const beforeCache = readUsageCostCache<{ files: Record; - }; + }>(); expect(beforeCache.files[sessionFile]?.parsedRecords).toBe(1); await fs.appendFile( @@ -1385,9 +1394,9 @@ describe("session cost usage", () => { endMs: Date.UTC(2026, 1, 5) + 24 * 60 * 60 * 1000 - 1, requestRefresh: false, }); - const afterCache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const afterCache = readUsageCostCache<{ files: Record; - }; + }>(); expect(summary.totals.totalTokens).toBe(60); expect(summary.totals.totalCost).toBeCloseTo(0.06, 5); @@ -1560,11 +1569,10 @@ describe("session cost usage", () => { await withStateDir(root, async () => { await refreshCostUsageCache({ config: configFor(1, 1) }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const cache = readUsageCostCache<{ pricingFingerprint?: unknown; files: Record>; - }; + }>(); expect(typeof cache.pricingFingerprint).toBe("string"); expect(cache.files[sessionFile]).not.toHaveProperty("pricingFingerprint"); expect(cache.files[sessionFile]).not.toHaveProperty("filePath"); @@ -1591,14 +1599,14 @@ describe("session cost usage", () => { }); }); - it("reclaims stale usage cache temp files before refreshing", async () => { - const root = await makeSessionCostRoot("cost-cache-temp-cleanup"); + it("does not create usage cache sidecar files", async () => { + const root = await makeSessionCostRoot("cost-cache-no-sidecars"); const sessionsDir = path.join(root, "agents", "main", "sessions"); await fs.mkdir(sessionsDir, { recursive: true }); - const sessionFile = path.join(sessionsDir, "sess-cache-temp-cleanup.jsonl"); + const sessionFile = path.join(sessionsDir, "sess-cache-no-sidecars.jsonl"); await fs.writeFile( sessionFile, - transcriptText("sess-cache-temp-cleanup", { + transcriptText("sess-cache-no-sidecars", { type: "message", timestamp: "2026-02-05T12:00:00.000Z", message: { @@ -1616,36 +1624,12 @@ describe("session cost usage", () => { "utf-8", ); - const staleLegacyTempPath = path.join(sessionsDir, ".usage-cost-cache.json.12345.tmp"); - const staleCurrentTempPath = path.join(sessionsDir, ".usage-cost-cache.12345.tmp"); - const recentTempPath = path.join(sessionsDir, ".usage-cost-cache.67890.tmp"); - const lockTempPath = path.join(sessionsDir, ".usage-cost-cache.json.lock.12345.tmp"); - await Promise.all( - [staleLegacyTempPath, staleCurrentTempPath, recentTempPath, lockTempPath].map((tempPath) => - fs.writeFile(tempPath, "partial\n", "utf-8"), - ), - ); - const staleTime = new Date(Date.now() - 60_000); - await Promise.all( - [staleLegacyTempPath, staleCurrentTempPath, lockTempPath].map((tempPath) => - fs.utimes(tempPath, staleTime, staleTime), - ), - ); - - const exists = async (filePath: string): Promise => - await fs.stat(filePath).then( - () => true, - () => false, - ); - await withStateDir(root, async () => { const result = await refreshCostUsageCache(); expect(result).toBe("refreshed"); - - expect(await exists(staleLegacyTempPath)).toBe(false); - expect(await exists(staleCurrentTempPath)).toBe(false); - expect(await exists(recentTempPath)).toBe(true); - expect(await exists(lockTempPath)).toBe(true); + expect( + (await fs.readdir(sessionsDir)).filter((name) => name.includes("usage-cost-cache")), + ).toEqual([]); const summary = await loadCostUsageSummaryFromCache({ startMs: Date.UTC(2026, 1, 5), @@ -1657,7 +1641,7 @@ describe("session cost usage", () => { }); }); - it("keeps queued durable aggregate refresh state scoped to the cache path", async () => { + it("keeps queued durable aggregate refresh state scoped to the agent database", async () => { const firstRoot = await makeSessionCostRoot("cost-cache-queued-first"); const secondRoot = await makeSessionCostRoot("cost-cache-queued-second"); const writeSession = async (root: string, sessionId: string) => { @@ -1800,11 +1784,10 @@ describe("session cost usage", () => { }); expect(summary.totals.totalTokens).toBe(30); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); await waitFor(async () => { - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const cache = readUsageCostCache<{ files: Record; - }; + }>(); return Boolean(cache.files[oldSessionFile]); }); }); @@ -1975,10 +1958,9 @@ describe("session cost usage", () => { await withStateDir(root, async () => { await refreshCostUsageCache({ sessionFiles: [sessionFile] }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { version: number }; + const cache = readUsageCostCache<{ version: number }>(); cache.version = 2; - await fs.writeFile(cachePath, `${JSON.stringify(cache)}\n`, "utf-8"); + writeUsageCostCache(cache); const summary = await loadSessionCostSummaryFromCache({ sessionId: "sess-cache-version", @@ -2232,10 +2214,9 @@ describe("session cost usage", () => { await withStateDir(root, async () => { await refreshCostUsageCache(); await refreshCostUsageCache({ sessionFiles: [sessionFile] }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const cache = readUsageCostCache<{ files: Record; - }; + }>(); expect(cache.files[sessionFile]).toHaveProperty("sessionSummary"); expect(cache.files[otherSessionFile]?.sessionSummary).toBeUndefined(); @@ -2266,24 +2247,20 @@ describe("session cost usage", () => { ); await withStateDir(root, async () => { - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const lockPath = `${cachePath}.lock`; - await fs.writeFile( - lockPath, - `${JSON.stringify({ - pid: process.pid, - startedAt: Date.now() - 60 * 60 * 1000, - })}\n`, - "utf-8", - ); + sessionCostUsageCacheTestApi.writeRefreshLock(undefined, { + pid: process.pid, + startedAt: Date.now() - 60 * 60 * 1000, + ownerNonce: "live-owner", + }); const result = await refreshCostUsageCache(); expect(result).toBe("busy"); - expect(await fs.readFile(lockPath, "utf-8")).toContain(String(process.pid)); + expect(sessionCostUsageCacheTestApi.readRefreshLock()).toMatchObject({ pid: process.pid }); + sessionCostUsageCacheTestApi.clearRefreshLock(); }); }); - it("treats in-progress usage cache lock writes as busy", async () => { + it("reclaims malformed SQLite usage cache locks", async () => { const root = await makeSessionCostRoot("cost-cache-malformed-lock-recent"); const sessionsDir = path.join(root, "agents", "main", "sessions"); await fs.mkdir(sessionsDir, { recursive: true }); @@ -2307,17 +2284,10 @@ describe("session cost usage", () => { ); await withStateDir(root, async () => { - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const lockPath = `${cachePath}.lock`; - await fs.writeFile(lockPath, "", "utf-8"); - - try { - const result = await refreshCostUsageCache(); - expect(result).toBe("busy"); - expect(await fs.readFile(lockPath, "utf-8")).toBe(""); - } finally { - await fs.rm(lockPath, { force: true }); - } + sessionCostUsageCacheTestApi.writeMalformedRefreshLock(undefined, "{"); + const result = await refreshCostUsageCache(); + expect(result).toBe("refreshed"); + expect(sessionCostUsageCacheTestApi.readRefreshLock()).toBeNull(); }); }); @@ -2345,16 +2315,11 @@ describe("session cost usage", () => { ); await withStateDir(root, async () => { - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const lockPath = `${cachePath}.lock`; - await fs.writeFile( - lockPath, - `${JSON.stringify({ - pid: 2_147_483_647, - startedAt: Date.now(), - })}\n`, - "utf-8", - ); + sessionCostUsageCacheTestApi.writeRefreshLock(undefined, { + pid: 2_147_483_647, + startedAt: Date.now(), + ownerNonce: "abandoned-owner", + }); const result = await refreshCostUsageCache(); expect(result).toBe("refreshed"); @@ -2400,11 +2365,7 @@ describe("session cost usage", () => { ); await withStateDir(root, async () => { - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const lockPath = `${cachePath}.lock`; - await fs.writeFile(lockPath, "{", "utf-8"); - const old = new Date(Date.now() - 60_000); - await fs.utimes(lockPath, old, old); + sessionCostUsageCacheTestApi.writeMalformedRefreshLock(undefined, "{"); const result = await refreshCostUsageCache(); expect(result).toBe("refreshed"); @@ -2426,11 +2387,10 @@ describe("session cost usage", () => { }); }); - it("throttles cache writes during a large stale refresh and skips writes when nothing changed", async () => { + it("checkpoints a large SQLite cache refresh and skips writes when nothing changed", async () => { const root = await makeSessionCostRoot("cost-cache-throttle"); const sessionsDir = path.join(root, "agents", "main", "sessions"); await fs.mkdir(sessionsDir, { recursive: true }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); const sessionCount = 300; // > USAGE_COST_CACHE_CHECKPOINT_FILES (256) const baseTimestamp = "2026-02-05T12:00:00.000Z"; @@ -2461,40 +2421,19 @@ describe("session cost usage", () => { } await withStateDir(root, async () => { - const renameSpy = vi.spyOn(fs, "rename"); - const cacheRenamesBefore = renameSpy.mock.calls.length; - try { - await refreshCostUsageCache(); - const cacheRenamesAfterCold = renameSpy.mock.calls.filter( - ([, dest]) => dest === cachePath, - ).length; + await refreshCostUsageCache(); + const updatedAt = readUsageCostCache<{ updatedAt: number }>().updatedAt; - // Without throttling this cold refresh would rewrite once per file plus a final flush. - // With checkpointing it must be far fewer than the file count. - expect(cacheRenamesAfterCold).toBeGreaterThan(0); - expect(cacheRenamesAfterCold).toBeLessThan(sessionCount / 4); + const summary = await loadCostUsageSummaryFromCache({ + startMs: Date.UTC(2026, 1, 5), + endMs: Date.UTC(2026, 1, 5) + 24 * 60 * 60 * 1000 - 1, + requestRefresh: false, + }); + expect(summary.totals.totalTokens).toBe(sessionCount); + expect(summary.cacheStatus?.status).toBe("fresh"); - const summary = await loadCostUsageSummaryFromCache({ - startMs: Date.UTC(2026, 1, 5), - endMs: Date.UTC(2026, 1, 5) + 24 * 60 * 60 * 1000 - 1, - requestRefresh: false, - }); - expect(summary.totals.totalTokens).toBe(sessionCount); - expect(summary.cacheStatus?.status).toBe("fresh"); - - // No-op refresh: nothing stale, nothing deleted -> no cache rewrite. - const renamesBeforeNoOp = renameSpy.mock.calls.filter( - ([, dest]) => dest === cachePath, - ).length; - await refreshCostUsageCache(); - const renamesAfterNoOp = renameSpy.mock.calls.filter( - ([, dest]) => dest === cachePath, - ).length; - expect(renamesAfterNoOp).toBe(renamesBeforeNoOp); - } finally { - renameSpy.mockRestore(); - void cacheRenamesBefore; - } + await refreshCostUsageCache(); + expect(readUsageCostCache<{ updatedAt: number }>().updatedAt).toBe(updatedAt); }); }); @@ -2618,10 +2557,9 @@ describe("session cost usage", () => { ); }); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const cache = JSON.parse(await fs.readFile(cachePath, "utf-8")) as { + const cache = readUsageCostCache<{ files: Record; - }; + }>(); expect(cache.files).toHaveProperty(firstSessionFile); expect(cache.files).toHaveProperty(secondSessionFile); expect(cache.files[firstSessionFile]).toHaveProperty("sessionSummary"); @@ -2656,13 +2594,11 @@ describe("session cost usage", () => { try { await withStateDir(root, async () => { await refreshCostUsageCache(); - const cachePath = path.join(sessionsDir, ".usage-cost-cache.json"); - const lockPath = `${cachePath}.lock`; - await fs.writeFile( - lockPath, - `${JSON.stringify({ pid: process.pid, startedAt: Date.now() })}\n`, - "utf-8", - ); + sessionCostUsageCacheTestApi.writeRefreshLock(undefined, { + pid: process.pid, + startedAt: Date.now(), + ownerNonce: "queued-owner", + }); try { const cold = await loadSessionCostSummaryFromCache({ @@ -2679,7 +2615,7 @@ describe("session cost usage", () => { }); expect(stillMissing.summary).toBeNull(); } finally { - await fs.rm(lockPath, { force: true }); + sessionCostUsageCacheTestApi.clearRefreshLock(); } await vi.waitFor( diff --git a/src/infra/session-cost-usage.ts b/src/infra/session-cost-usage.ts index 50810490d975..0c277142c06d 100644 --- a/src/infra/session-cost-usage.ts +++ b/src/infra/session-cost-usage.ts @@ -39,7 +39,9 @@ import { selectVisibleTranscriptEvents } from "../config/sessions/transcript-vis import type { SessionEntry } from "../config/sessions/types.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; +import { normalizeAgentId } from "../routing/session-key.js"; import { stripEnvelope, stripMessageIdHints } from "../shared/chat-envelope.js"; +import { resolveOpenClawAgentSqlitePath } from "../state/openclaw-agent-db.js"; import { runTasksWithConcurrency } from "../utils/run-with-concurrency.js"; import { countToolResults, extractToolCallNames } from "../utils/transcript-tools.js"; import { @@ -49,7 +51,12 @@ import { } from "../utils/usage-format.js"; import { formatErrorMessage } from "./errors.js"; import { createTimeZoneDayKeyFormatter } from "./format-time/format-datetime.js"; -import { replaceFileAtomic } from "./replace-file.js"; +import { + acquireSessionCostUsageRefreshLock, + isSessionCostUsageRefreshRunning, + readSessionCostUsageCacheJson, + writeSessionCostUsageCacheJson, +} from "./session-cost-usage-cache.sqlite.js"; import { addCostUsageTotals as addTotals, cloneCostUsageTotals as cloneTotals, @@ -98,9 +105,6 @@ export type { // Bump when the durable cache schema or the meaning of cached totals changes, so // older builds are rebuilt instead of served stale. const USAGE_COST_CACHE_VERSION = 7; -const USAGE_COST_CACHE_FILE = ".usage-cost-cache.json"; -const USAGE_COST_CACHE_LOCK_WRITE_GRACE_MS = 10_000; -const USAGE_COST_CACHE_TEMP_FILE_GRACE_MS = USAGE_COST_CACHE_LOCK_WRITE_GRACE_MS; const USAGE_COST_TRANSCRIPT_STAT_CONCURRENCY = 32; // Checkpoint policy for refreshCostUsageCache: bound the cost of full cache // serialization when scanning thousands of session files. Smaller of the two @@ -111,8 +115,8 @@ const logger = createSubsystemLogger("usage-cost-cache"); type UsageCostRefreshState = { agentId?: string; - cachePath: string; config?: OpenClawConfig; + databasePath: string; fullRefreshRequested: boolean; pendingSessionFiles: Set; running: boolean; @@ -124,6 +128,10 @@ type UsageCostRefreshResult = "refreshed" | "busy"; const usageCostRefreshes = new Map(); +function resolveUsageCostCacheDatabasePath(agentId?: string): string { + return resolveOpenClawAgentSqlitePath({ agentId: normalizeAgentId(agentId) }); +} + type UsageCostCachedUsageEntry = CostUsageTotals & { timestamp: number; provider?: string; @@ -168,25 +176,10 @@ type UsageCostTranscriptFile = { sessionId?: string; }; -type UsageCostCacheLock = { - pid: number; - startedAt: number; - token?: string; -}; - -type UsageCostCacheLockReadResult = - | { state: "missing" } - | { state: "valid"; lock: UsageCostCacheLock } - | { state: "malformed"; mtimeMs: number }; - function resolveUsageCostPricingFingerprint(config?: OpenClawConfig): string { return resolveModelCostConfigFingerprint(config); } -function resolveUsageCostCachePath(agentId?: string): string { - return path.join(resolveSessionTranscriptsDirForAgent(agentId), USAGE_COST_CACHE_FILE); -} - function resolveUsageCostSessionStorePath(params?: { agentId?: string; sessionsDir?: string; @@ -196,141 +189,6 @@ function resolveUsageCostSessionStorePath(params?: { : resolveDefaultSessionStorePath(params?.agentId); } -function resolveUsageCostCacheLockPath(cachePath: string): string { - return `${cachePath}.lock`; -} - -function parseUsageCostCacheLock(raw: string): UsageCostCacheLock | null { - const parsed = JSON.parse(raw) as unknown; - if (!parsed || typeof parsed !== "object") { - return null; - } - const lock = parsed as Partial; - if ( - typeof lock.pid !== "number" || - !Number.isInteger(lock.pid) || - lock.pid <= 0 || - typeof lock.startedAt !== "number" || - !Number.isFinite(lock.startedAt) || - (lock.token !== undefined && typeof lock.token !== "string") - ) { - return null; - } - return { pid: lock.pid, startedAt: lock.startedAt, token: lock.token }; -} - -async function readUsageCostCacheLockState( - lockPath: string, -): Promise { - try { - const lock = parseUsageCostCacheLock(await fs.promises.readFile(lockPath, "utf-8")); - if (lock) { - return { state: "valid", lock }; - } - } catch (err) { - if ((err as NodeJS.ErrnoException).code === "ENOENT") { - return { state: "missing" }; - } - } - const stats = await fs.promises.stat(lockPath).catch(() => null); - if (!stats) { - return { state: "missing" }; - } - return { state: "malformed", mtimeMs: stats.mtimeMs }; -} - -async function readUsageCostCacheLock(lockPath: string): Promise { - const result = await readUsageCostCacheLockState(lockPath); - return result.state === "valid" ? result.lock : null; -} - -function isMalformedUsageCostCacheLockRecent(mtimeMs: number): boolean { - return Date.now() - mtimeMs < USAGE_COST_CACHE_LOCK_WRITE_GRACE_MS; -} - -async function writeUsageCostCacheLockAtomically( - lockPath: string, - lock: UsageCostCacheLock, -): Promise { - const tempPath = `${lockPath}.${process.pid}.${process.hrtime.bigint()}.tmp`; - await fs.promises.writeFile(tempPath, `${JSON.stringify(lock)}\n`, { flag: "wx" }); - try { - await fs.promises.link(tempPath, lockPath); - } finally { - await fs.promises.rm(tempPath, { force: true }).catch(() => undefined); - } -} - -function isProcessRunning(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch (err) { - const code = (err as NodeJS.ErrnoException).code; - return code === "EPERM"; - } -} - -async function isUsageCostCacheRefreshRunning(cachePath: string): Promise { - const lockPath = resolveUsageCostCacheLockPath(cachePath); - const result = await readUsageCostCacheLockState(lockPath); - if (result.state === "missing") { - return false; - } - if (result.state === "malformed") { - if (isMalformedUsageCostCacheLockRecent(result.mtimeMs)) { - return true; - } - await fs.promises.rm(lockPath, { force: true }).catch(() => undefined); - return false; - } - const lock = result.lock; - if (isProcessRunning(lock.pid)) { - return true; - } - await fs.promises.rm(lockPath, { force: true }).catch(() => undefined); - return false; -} - -async function acquireUsageCostCacheRefreshLock(cachePath: string): Promise<{ - acquired: boolean; - release: () => Promise; -}> { - const lockPath = resolveUsageCostCacheLockPath(cachePath); - await fs.promises.mkdir(path.dirname(lockPath), { recursive: true }); - const lock: UsageCostCacheLock = { - pid: process.pid, - startedAt: Date.now(), - token: `${process.pid}:${Date.now()}:${process.hrtime.bigint()}`, - }; - try { - await writeUsageCostCacheLockAtomically(lockPath, lock); - return { - acquired: true, - release: async () => { - const current = await readUsageCostCacheLock(lockPath); - if ( - current?.pid === lock.pid && - current.startedAt === lock.startedAt && - current.token === lock.token - ) { - await fs.promises.rm(lockPath, { force: true }).catch(() => undefined); - } - }, - }; - } catch (err) { - const code = (err as NodeJS.ErrnoException).code; - if (code !== "EEXIST") { - throw err; - } - if (await isUsageCostCacheRefreshRunning(cachePath)) { - return { acquired: false, release: async () => undefined }; - } - await fs.promises.rm(lockPath, { force: true }).catch(() => undefined); - return acquireUsageCostCacheRefreshLock(cachePath); - } -} - function createEmptyUsageCostCache(pricingFingerprint: string): UsageCostCacheFile { return { version: USAGE_COST_CACHE_VERSION, updatedAt: 0, pricingFingerprint, files: {} }; } @@ -357,52 +215,36 @@ function normalizeUsageCostCache(raw: unknown, pricingFingerprint: string): Usag }; } -async function readUsageCostCache( - cachePath: string, +function readUsageCostCache( + agentId: string | undefined, pricingFingerprint: string, -): Promise { + databasePath?: string, +): UsageCostCacheFile { try { - const raw = await fs.promises.readFile(cachePath, "utf-8"); + const raw = readSessionCostUsageCacheJson(agentId, databasePath); + if (!raw) { + return createEmptyUsageCostCache(pricingFingerprint); + } return normalizeUsageCostCache(JSON.parse(raw), pricingFingerprint); } catch { return createEmptyUsageCostCache(pricingFingerprint); } } -async function writeUsageCostCache(cachePath: string, cache: UsageCostCacheFile): Promise { - await replaceFileAtomic({ - filePath: cachePath, - content: `${JSON.stringify(cache)}\n`, - tempPrefix: ".usage-cost-cache", +function writeUsageCostCache( + agentId: string | undefined, + cache: UsageCostCacheFile, + databasePath?: string, +): void { + const valueJson = JSON.stringify(cache); + writeSessionCostUsageCacheJson({ + agentId, + databasePath, + valueJson, + updatedAt: cache.updatedAt, }); } -function isUsageCostCacheTempFileName(name: string): boolean { - if (!name.endsWith(".tmp") || name.startsWith(`${USAGE_COST_CACHE_FILE}.lock.`)) { - return false; - } - return name.startsWith(".usage-cost-cache.") || name.startsWith(`${USAGE_COST_CACHE_FILE}.`); -} - -async function cleanupStaleUsageCostCacheTempFiles(cachePath: string): Promise { - const dir = path.dirname(cachePath); - const cutoffMs = Date.now() - USAGE_COST_CACHE_TEMP_FILE_GRACE_MS; - const entries = await fs.promises.readdir(dir, { withFileTypes: true }).catch(() => []); - await Promise.all( - entries.map(async (entry) => { - if (!entry.isFile() || !isUsageCostCacheTempFileName(entry.name)) { - return; - } - const tempPath = path.join(dir, entry.name); - const stats = await fs.promises.stat(tempPath).catch(() => null); - if (!stats || stats.mtimeMs > cutoffMs) { - return; - } - await fs.promises.rm(tempPath, { force: true }).catch(() => undefined); - }), - ); -} - async function listUsageCountedTranscriptFileStats( agentId?: string, params?: { minMtimeMs?: number; sessionsDir?: string }, @@ -1834,28 +1676,29 @@ async function scanUsageFileForCache(params: { }; } -async function refreshCostUsageCacheForPath(params?: { +async function refreshCostUsageCacheForAgent(params?: { config?: OpenClawConfig; agentId?: string; - cachePath?: string; + databasePath?: string; maxFiles?: number; sessionsDir?: string; sessionFiles?: string[]; startMs?: number; }): Promise { - const cachePath = params?.cachePath ?? resolveUsageCostCachePath(params?.agentId); - const lock = await acquireUsageCostCacheRefreshLock(cachePath); + const databasePath = + params?.databasePath ?? + resolveOpenClawAgentSqlitePath({ agentId: normalizeAgentId(params?.agentId) }); + const lock = acquireSessionCostUsageRefreshLock(params?.agentId, databasePath); if (!lock.acquired) { return "busy"; } try { - await cleanupStaleUsageCostCacheTempFiles(cachePath); const pricingFingerprint = resolveUsageCostPricingFingerprint(params?.config); - const cache = await readUsageCostCache(cachePath, pricingFingerprint); + const cache = readUsageCostCache(params?.agentId, pricingFingerprint, databasePath); const files = await listUsageCountedTranscriptFiles(params?.agentId, { sessionsDir: params?.sessionsDir, }); - // Empty caches come from missing/corrupt files and version/pricing mismatches. + // Empty caches come from missing/corrupt rows and version/pricing mismatches. // Persist the empty current-shape cache even when this refresh scans no files. let cacheMutated = cache.updatedAt === 0; const sessionSummaryFiles = new Set(params?.sessionFiles ?? []); @@ -1915,7 +1758,7 @@ async function refreshCostUsageCacheForPath(params?: { now - lastCheckpointMs >= USAGE_COST_CACHE_CHECKPOINT_INTERVAL_MS ) { cache.updatedAt = now; - await writeUsageCostCache(cachePath, cache); + writeUsageCostCache(params?.agentId, cache, databasePath); dirtyCount = 0; lastCheckpointMs = Date.now(); } @@ -1923,11 +1766,11 @@ async function refreshCostUsageCacheForPath(params?: { if (cacheMutated || dirtyCount > 0) { cache.updatedAt = Date.now(); - await writeUsageCostCache(cachePath, cache); + writeUsageCostCache(params?.agentId, cache, databasePath); } return "refreshed"; } finally { - await lock.release(); + lock.release(); } } @@ -1938,7 +1781,7 @@ export async function refreshCostUsageCache(params?: { sessionFiles?: string[]; startMs?: number; }): Promise { - return await refreshCostUsageCacheForPath(params); + return await refreshCostUsageCacheForAgent(params); } export async function loadCostUsageSummaryFromCache(params: { @@ -1950,12 +1793,11 @@ export async function loadCostUsageSummaryFromCache(params: { requestRefresh?: boolean; refreshMode?: "background" | "sync-when-empty"; }): Promise { - const cachePath = resolveUsageCostCachePath(params.agentId); + const databasePath = resolveUsageCostCacheDatabasePath(params.agentId); + const refreshKey = databasePath; const pricingFingerprint = resolveUsageCostPricingFingerprint(params.config); - let [cache, files] = await Promise.all([ - readUsageCostCache(cachePath, pricingFingerprint), - listUsageCountedTranscriptFiles(params.agentId), - ]); + let cache = readUsageCostCache(params.agentId, pricingFingerprint, databasePath); + let files = await listUsageCountedTranscriptFiles(params.agentId); const staleFiles = getUsageCostStaleFiles({ cache, files, @@ -1971,10 +1813,8 @@ export async function loadCostUsageSummaryFromCache(params: { agentId: params.agentId, startMs: params.startMs, }); - [cache, files] = await Promise.all([ - readUsageCostCache(cachePath, pricingFingerprint), - listUsageCountedTranscriptFiles(params.agentId), - ]); + cache = readUsageCostCache(params.agentId, pricingFingerprint, databasePath); + files = await listUsageCountedTranscriptFiles(params.agentId); if (result === "refreshed") { const remainingStaleFiles = getUsageCostStaleFiles({ cache, @@ -1988,14 +1828,14 @@ export async function loadCostUsageSummaryFromCache(params: { requestCostUsageCacheRefresh({ config: params.config, agentId: params.agentId }); } } - const refreshRunning = await isUsageCostCacheRefreshRunning(cachePath); + const refreshRunning = isSessionCostUsageRefreshRunning(params.agentId, databasePath); return buildCostUsageSummaryFromCache({ cache, files, startMs: params.startMs, endMs: params.endMs, dayBucket: params.dayBucket, - refreshing: usageCostRefreshes.has(cachePath) || refreshRunning, + refreshing: usageCostRefreshes.has(refreshKey) || refreshRunning, }); } @@ -2011,12 +1851,11 @@ export async function loadSessionCostSummaryFromCache(params: { requestRefresh?: boolean; refreshMode?: "background" | "sync-when-empty"; }): Promise<{ summary: SessionCostSummary | null; cacheStatus: UsageCacheStatus }> { - const cachePath = resolveUsageCostCachePath(params.agentId); + const databasePath = resolveUsageCostCacheDatabasePath(params.agentId); + const refreshKey = databasePath; const pricingFingerprint = resolveUsageCostPricingFingerprint(params.config); - let [cache, file] = await Promise.all([ - readUsageCostCache(cachePath, pricingFingerprint), - resolveUsageCostTranscriptFile(params.sessionFile), - ]); + let cache = readUsageCostCache(params.agentId, pricingFingerprint, databasePath); + let file = await resolveUsageCostTranscriptFile(params.sessionFile); let entry = cache.files[params.sessionFile]; let stale = !file || @@ -2034,10 +1873,8 @@ export async function loadSessionCostSummaryFromCache(params: { sessionFiles: [params.sessionFile], }); if (result === "refreshed") { - [cache, file] = await Promise.all([ - readUsageCostCache(cachePath, pricingFingerprint), - resolveUsageCostTranscriptFile(params.sessionFile), - ]); + cache = readUsageCostCache(params.agentId, pricingFingerprint, databasePath); + file = await resolveUsageCostTranscriptFile(params.sessionFile); entry = cache.files[params.sessionFile]; stale = !file || @@ -2064,7 +1901,8 @@ export async function loadSessionCostSummaryFromCache(params: { } } const refreshRunning = - usageCostRefreshes.has(cachePath) || (await isUsageCostCacheRefreshRunning(cachePath)); + usageCostRefreshes.has(refreshKey) || + isSessionCostUsageRefreshRunning(params.agentId, databasePath); let summary = stale ? null : (entry?.sessionSummary ?? null); // Persisted summaries use Gateway-local day keys. Explicit request calendars // must rebuild daily projections from the cached transcript entries. @@ -2126,7 +1964,7 @@ export async function loadSessionCostSummariesFromCache(params: { dayBucket?: UsageDailyBucket; requestRefresh?: boolean; }): Promise<{ summaries: Array; cacheStatus: UsageCacheStatus }> { - const cachePath = resolveUsageCostCachePath(params.agentId); + const databasePath = resolveUsageCostCacheDatabasePath(params.agentId); const pricingFingerprint = resolveUsageCostPricingFingerprint(params.config); const fileTasks = params.sessions.map( (session) => async () => await resolveUsageCostTranscriptFile(session.sessionFile), @@ -2135,11 +1973,9 @@ export async function loadSessionCostSummariesFromCache(params: { tasks: fileTasks, limit: USAGE_COST_TRANSCRIPT_STAT_CONCURRENCY, }).then(({ results }) => results); - const [cache, files, refreshRunning] = await Promise.all([ - readUsageCostCache(cachePath, pricingFingerprint), - filesPromise, - isUsageCostCacheRefreshRunning(cachePath), - ]); + const cache = readUsageCostCache(params.agentId, pricingFingerprint, databasePath); + const refreshRunning = isSessionCostUsageRefreshRunning(params.agentId, databasePath); + const files = await filesPromise; const staleFiles = new Set(); let cachedFiles = 0; const requiresDailyRebucket = params.dayBucket !== undefined; @@ -2217,8 +2053,9 @@ export function requestCostUsageCacheRefresh(params?: { agentId?: string; sessionFiles?: string[]; }): void { - const cachePath = resolveUsageCostCachePath(params?.agentId); - const existing = usageCostRefreshes.get(cachePath); + const databasePath = resolveUsageCostCacheDatabasePath(params?.agentId); + const refreshKey = databasePath; + const existing = usageCostRefreshes.get(refreshKey); if (existing) { mergeUsageCostRefreshRequest(existing, params); return; @@ -2226,16 +2063,16 @@ export function requestCostUsageCacheRefresh(params?: { const state: UsageCostRefreshState = { agentId: params?.agentId, - cachePath, config: params?.config, + databasePath, fullRefreshRequested: false, pendingSessionFiles: new Set(), running: false, - sessionsDir: path.dirname(cachePath), + sessionsDir: resolveSessionTranscriptsDirForAgent(params?.agentId), }; mergeUsageCostRefreshRequest(state, params); - usageCostRefreshes.set(cachePath, state); - scheduleUsageCostRefresh(cachePath, state); + usageCostRefreshes.set(refreshKey, state); + scheduleUsageCostRefresh(refreshKey, state); } function mergeUsageCostRefreshRequest( @@ -2291,10 +2128,10 @@ async function runQueuedUsageCostRefresh( state.pendingSessionFiles.clear(); } state.fullRefreshRequested = false; - const result = await refreshCostUsageCacheForPath({ - cachePath: state.cachePath, + const result = await refreshCostUsageCacheForAgent({ config: state.config, agentId: state.agentId, + databasePath: state.databasePath, sessionsDir: state.sessionsDir, sessionFiles: fullRefreshRequested ? undefined : sessionFiles, });