From 09aff1b94d2904460c8eeceefad546df00c40b3d Mon Sep 17 00:00:00 2001 From: outofplace <69466055+outof-place@users.noreply.github.com> Date: Sat, 10 Oct 2026 02:19:29 +0200 Subject: [PATCH] perf(usage): keep per-source usage caches off the main thread (#26980) * perf(usage): keep per-source usage caches off the main thread The usage stores kept every provider's per-source scan cache (processedFiles / processedDatabases) inside the one JSON file main loads at startup. On a large Claude history that file is 56 MB, 43 MB of it per-source records only a scan reads, so main paid a sync read+parse at launch, a structured clone of the whole cache to and from the scan worker on every scan, and a pretty-printed rewrite after every scan. The usage scan worker now owns those records in a sidecar (`-sources.json`): it reads it before a scan, writes it after (compact, durable rename), and returns only sessions and daily aggregates. Main keeps and writes only that report, compact. A cache still carrying inline records is split on the worker the first time it is loaded (above 8 MB); smaller ones parse on main and rescan cold. Sync IPC getters and automation lookups wait for the load. Synthetic 56 MB cache, same machine, interleaved runs: - first launch: 108 ms sync parse -> 18 ms main busy, 4 ms longest block - later launches: ~69 ms -> ~7 ms - forced scan, longest main-thread block: 89 ms -> 11 ms (real built app, headless: 84-94 ms -> 11-13 ms) Co-Authored-By: Claude Opus 5.5 * fix(usage): preserve cache attribution and legacy warm scans * fix(usage): reclaim cache temps after worker retirement --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Neil --- src/main/claude-usage/store.test.ts | 20 +- src/main/claude-usage/store.ts | 10 +- .../codex-usage-long-context-billing.test.ts | 3 +- src/main/codex-usage/codex-usage-provider.ts | 9 +- .../store-automation-usage.test.ts | 3 +- .../codex-usage/store-model-pricing.test.ts | 3 +- src/main/codex-usage/store-orca-scope.test.ts | 3 +- .../codex-usage/store-persistence.test.ts | 43 +++-- src/main/codex-usage/store-test-harness.ts | 1 - src/main/codex-usage/store.ts | 5 +- src/main/durable-file-write.test.ts | 61 ++++++- src/main/durable-file-write.ts | 23 ++- src/main/ipc/usage-provider-handlers.test.ts | 71 ++++++-- src/main/ipc/usage-provider-handlers.ts | 13 +- src/main/muse-usage/muse-usage-provider.ts | 9 +- .../persisted-file-normalization.ts | 16 ++ .../persisted-state-normalization.ts | 11 +- src/main/muse-usage/store.ts | 4 +- .../opencode-usage/opencode-usage-provider.ts | 13 +- .../persisted-database-normalization.ts | 46 +++++ .../persisted-state-normalization.ts | 43 +---- src/main/opencode-usage/store.test.ts | 15 +- src/main/opencode-usage/store.ts | 5 +- src/main/usage/usage-provider-contract.ts | 18 +- .../usage-provider-store-lifecycle.test.ts | 171 ++++++++++++++++-- .../usage/usage-provider-store-lifecycle.ts | 123 ++++++++++--- .../usage-scan-worker-cache-cleanup.test.ts | 102 +++++++++++ .../usage/usage-scan-worker-cache-cleanup.ts | 39 ++++ .../usage/usage-scan-worker-client.test.ts | 84 ++++++--- src/main/usage/usage-scan-worker-client.ts | 138 +++++++------- src/main/usage/usage-scan-worker-entry.ts | 74 ++++++-- .../usage-scan-worker-event-loop.test.ts | 18 +- src/main/usage/usage-scan-worker-protocol.ts | 80 ++++---- src/main/usage/usage-scan-worker-spawn.ts | 118 +++++------- .../usage/usage-source-cache-file.test.ts | 113 ++++++++++++ src/main/usage/usage-source-cache-file.ts | 96 ++++++++++ 36 files changed, 1175 insertions(+), 429 deletions(-) create mode 100644 src/main/muse-usage/persisted-file-normalization.ts create mode 100644 src/main/opencode-usage/persisted-database-normalization.ts create mode 100644 src/main/usage/usage-scan-worker-cache-cleanup.test.ts create mode 100644 src/main/usage/usage-scan-worker-cache-cleanup.ts create mode 100644 src/main/usage/usage-source-cache-file.test.ts create mode 100644 src/main/usage/usage-source-cache-file.ts diff --git a/src/main/claude-usage/store.test.ts b/src/main/claude-usage/store.test.ts index 565fdefab60..88105e3ccc7 100644 --- a/src/main/claude-usage/store.test.ts +++ b/src/main/claude-usage/store.test.ts @@ -15,7 +15,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanClaudeUsageFilesViaWorker: vi.fn() + scanClaudeUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) import { ClaudeUsageStore, initClaudeUsagePath } from './store' @@ -94,7 +95,6 @@ describe('ClaudeUsageStore', () => { initClaudeUsagePath() vi.mocked(scanClaudeUsageFilesViaWorker).mockReset() vi.mocked(scanClaudeUsageFilesViaWorker).mockResolvedValue({ - processedFiles: [], sessions: [], dailyAggregates: [] }) @@ -693,7 +693,7 @@ describe('ClaudeUsageStore', () => { expect(usage.unavailableReason).toBe('scan_failed') }) - it('adapts Claude scans to pretty-printed cache persistence', async () => { + it('hands Claude scans the worker-owned source cache and persists a compact report', async () => { const store = createStoreWithState({ schemaVersion: 5, scanState: { @@ -705,9 +705,17 @@ describe('ClaudeUsageStore', () => { }) await store.refresh(true) + await store.flush() - expect(scanClaudeUsageFilesViaWorker).toHaveBeenCalledWith([], []) - expect(readFileSync(join(tempUserData, 'orca-claude-usage.json'), 'utf-8')).toContain('\n') + expect(scanClaudeUsageFilesViaWorker).toHaveBeenCalledWith([], { + path: join(tempUserData, 'orca-claude-usage-sources.json'), + schemaVersion: 6, + worktreeFingerprint: '[]', + reuse: false + }) + const persisted = readFileSync(join(tempUserData, 'orca-claude-usage.json'), 'utf-8') + expect(persisted).not.toContain('\n') + expect(JSON.parse(persisted)).not.toHaveProperty('processedFiles') }) it('joins a scan that is already in flight when the run finished before it started', async () => { @@ -722,7 +730,6 @@ describe('ClaudeUsageStore', () => { }) // Prime the worktree fingerprint so an unforced refresh can return early. vi.mocked(scanClaudeUsageFilesViaWorker).mockResolvedValue({ - processedFiles: [], sessions: [], dailyAggregates: [] }) @@ -743,7 +750,6 @@ describe('ClaudeUsageStore', () => { startScan() await scanFinished return { - processedFiles: [], sessions: [createWorktreeUsageSession(worktreeId)], dailyAggregates: [] } diff --git a/src/main/claude-usage/store.ts b/src/main/claude-usage/store.ts index 686b32ab827..ac4d3264f08 100644 --- a/src/main/claude-usage/store.ts +++ b/src/main/claude-usage/store.ts @@ -14,7 +14,10 @@ import type { import type { AutomationRunUsage } from '../../shared/automations-types' import type { Store } from '../persistence' import type { ClaudeUsagePersistedState } from './types' -import { scanClaudeUsageFilesViaWorker } from '../usage/usage-scan-worker-spawn' +import { + scanClaudeUsageFilesViaWorker, + splitUsageCacheFileViaWorker +} from '../usage/usage-scan-worker-spawn' import { UsageProviderStoreLifecycle } from '../usage/usage-provider-store-lifecycle' import { buildBreakdown, buildDaily, buildSummary } from './claude-usage-report-aggregation' import { buildRecentSessions } from './claude-usage-session-rows' @@ -90,8 +93,8 @@ export class ClaudeUsageStore extends UsageProviderStoreLifecycle< normalizeState: normalizePersistedState, sourceKey: 'processedFiles', dataPresenceKey: 'hasAnyClaudeData', - jsonIndent: 2, - scan: scanClaudeUsageFilesViaWorker + scan: scanClaudeUsageFilesViaWorker, + splitCacheFile: splitUsageCacheFileViaWorker }) } @@ -142,6 +145,7 @@ export class ClaudeUsageStore extends UsageProviderStoreLifecycle< } async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise { + await this.whenLoaded() return resolveAutomationRunUsage(input, { getState: () => this.state, refresh: (force) => this.refresh(force), diff --git a/src/main/codex-usage/codex-usage-long-context-billing.test.ts b/src/main/codex-usage/codex-usage-long-context-billing.test.ts index d11ed5144e6..9a6afb662f0 100644 --- a/src/main/codex-usage/codex-usage-long-context-billing.test.ts +++ b/src/main/codex-usage/codex-usage-long-context-billing.test.ts @@ -15,7 +15,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanCodexUsageFilesViaWorker: vi.fn() + scanCodexUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) import { createUsageWorktreeResolver } from '../usage/usage-worktree-resolver' diff --git a/src/main/codex-usage/codex-usage-provider.ts b/src/main/codex-usage/codex-usage-provider.ts index e4422a4c1be..ada5cd8b6c6 100644 --- a/src/main/codex-usage/codex-usage-provider.ts +++ b/src/main/codex-usage/codex-usage-provider.ts @@ -1,6 +1,6 @@ import type { UsageProvider } from '../usage/usage-provider-contract' import { scanCodexUsageFilesViaWorker } from '../usage/usage-scan-worker-spawn' -import type { CodexUsageDailyAggregate, CodexUsagePersistedFile, CodexUsageSession } from './types' +import type { CodexUsageDailyAggregate, CodexUsageSession } from './types' // Why: v5 keys Codex ownership on raw token_count identity without session id // so forks that rewrite session_meta still match. Older caches used session- @@ -13,9 +13,4 @@ export const codexUsageProvider = { label: 'Codex', schemaVersion: CODEX_USAGE_SCHEMA_VERSION, scan: scanCodexUsageFilesViaWorker -} satisfies UsageProvider< - 'processedFiles', - CodexUsagePersistedFile, - CodexUsageSession, - CodexUsageDailyAggregate -> +} satisfies UsageProvider diff --git a/src/main/codex-usage/store-automation-usage.test.ts b/src/main/codex-usage/store-automation-usage.test.ts index b32eb030f46..e2602e122f4 100644 --- a/src/main/codex-usage/store-automation-usage.test.ts +++ b/src/main/codex-usage/store-automation-usage.test.ts @@ -18,7 +18,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanCodexUsageFilesViaWorker: vi.fn() + scanCodexUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) describe('CodexUsageStore', () => { diff --git a/src/main/codex-usage/store-model-pricing.test.ts b/src/main/codex-usage/store-model-pricing.test.ts index 50053c9975f..5bfaa4fd346 100644 --- a/src/main/codex-usage/store-model-pricing.test.ts +++ b/src/main/codex-usage/store-model-pricing.test.ts @@ -12,7 +12,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanCodexUsageFilesViaWorker: vi.fn() + scanCodexUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) describe('CodexUsageStore', () => { diff --git a/src/main/codex-usage/store-orca-scope.test.ts b/src/main/codex-usage/store-orca-scope.test.ts index ab838bffd4e..3109488a276 100644 --- a/src/main/codex-usage/store-orca-scope.test.ts +++ b/src/main/codex-usage/store-orca-scope.test.ts @@ -12,7 +12,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanCodexUsageFilesViaWorker: vi.fn() + scanCodexUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) describe('CodexUsageStore', () => { diff --git a/src/main/codex-usage/store-persistence.test.ts b/src/main/codex-usage/store-persistence.test.ts index 882ef59aed9..81bcb55a983 100644 --- a/src/main/codex-usage/store-persistence.test.ts +++ b/src/main/codex-usage/store-persistence.test.ts @@ -15,10 +15,12 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanCodexUsageFilesViaWorker: vi.fn() + scanCodexUsageFilesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) import { CodexUsageStore, normalizePersistedState } from './store' +import { CODEX_USAGE_SCHEMA_VERSION } from './codex-usage-provider' import { scanCodexUsageFilesViaWorker } from '../usage/usage-scan-worker-spawn' describe('CodexUsageStore', () => { @@ -41,7 +43,12 @@ describe('CodexUsageStore', () => { join(storeEnv.tempUserData, 'orca-codex-usage.json'), 'utf-8' ) - expect(scanCodexUsageFilesViaWorker).toHaveBeenCalledWith([], []) + expect(scanCodexUsageFilesViaWorker).toHaveBeenCalledWith([], { + path: join(storeEnv.tempUserData, 'orca-codex-usage-sources.json'), + schemaVersion: CODEX_USAGE_SCHEMA_VERSION, + worktreeFingerprint: '[]', + reuse: false + }) expect(persistedJson).toBe(JSON.stringify(JSON.parse(persistedJson))) }) @@ -96,22 +103,8 @@ describe('CodexUsageStore', () => { }) it('rescans from scratch when the cache predates per-request long-context counts', async () => { - const processedFile = { - path: '/codex/sessions/rollout-1.jsonl', - mtimeMs: 1, - size: 10, - sessions: [], - dailyAggregates: [], - ownedEventKeys: [], - hasDeferredClaims: false, - parseResumeState: null - } const scanMock = vi.mocked(scanCodexUsageFilesViaWorker) - scanMock.mockResolvedValue({ - processedFiles: [processedFile], - sessions: [], - dailyAggregates: [] - }) + const sourceCachePath = join(storeEnv.tempUserData, 'orca-codex-usage-sources.json') const cacheFile = join(storeEnv.tempUserData, 'orca-codex-usage.json') const seeded = new CodexUsageStore({ getRepos: () => [], getAllWorktreeMeta: () => ({}) }) await seeded.setEnabled(true) @@ -121,12 +114,22 @@ describe('CodexUsageStore', () => { scanMock.mockClear() await new CodexUsageStore({ getRepos: () => [], getAllWorktreeMeta: () => ({}) }).refresh(true) - // Control: a current-schema cache hands its processed files back for reuse. - expect(scanMock).toHaveBeenLastCalledWith([], [processedFile]) + // Control: a current-schema cache lets the worker reuse its processed files. + expect(scanMock).toHaveBeenLastCalledWith([], { + path: sourceCachePath, + schemaVersion: CODEX_USAGE_SCHEMA_VERSION, + worktreeFingerprint: '[]', + reuse: true + }) writeFileSync(cacheFile, JSON.stringify({ ...JSON.parse(current), schemaVersion: 5 })) scanMock.mockClear() await new CodexUsageStore({ getRepos: () => [], getAllWorktreeMeta: () => ({}) }).refresh(false) - expect(scanMock).toHaveBeenLastCalledWith([], []) + expect(scanMock).toHaveBeenLastCalledWith([], { + path: sourceCachePath, + schemaVersion: CODEX_USAGE_SCHEMA_VERSION, + worktreeFingerprint: '[]', + reuse: false + }) }) }) diff --git a/src/main/codex-usage/store-test-harness.ts b/src/main/codex-usage/store-test-harness.ts index fb178f38225..a2fc1f4fce8 100644 --- a/src/main/codex-usage/store-test-harness.ts +++ b/src/main/codex-usage/store-test-harness.ts @@ -9,7 +9,6 @@ import type { CodexUsagePersistedState } from './types' export function createEmptyScanResult() { return { - processedFiles: [], sessions: [], dailyAggregates: [] } diff --git a/src/main/codex-usage/store.ts b/src/main/codex-usage/store.ts index 7e91fb3945a..1f93663edfd 100644 --- a/src/main/codex-usage/store.ts +++ b/src/main/codex-usage/store.ts @@ -20,6 +20,7 @@ import { resolveCodexAutomationRunUsage } from './codex-automation-run-attributi import { buildRecentSessions } from './codex-usage-session-rows' import { buildBreakdown, buildDaily, buildSummary } from './codex-usage-rollup-projections' import { UsageProviderStoreLifecycle } from '../usage/usage-provider-store-lifecycle' +import { splitUsageCacheFileViaWorker } from '../usage/usage-scan-worker-spawn' const SCHEMA_VERSION = CODEX_USAGE_SCHEMA_VERSION @@ -95,7 +96,8 @@ export class CodexUsageStore extends UsageProviderStoreLifecycle< normalizeState: normalizePersistedState, sourceKey: 'processedFiles', dataPresenceKey: 'hasAnyCodexData', - scan: codexUsageProvider.scan + scan: codexUsageProvider.scan, + splitCacheFile: splitUsageCacheFileViaWorker }) } @@ -143,6 +145,7 @@ export class CodexUsageStore extends UsageProviderStoreLifecycle< } async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise { + await this.whenLoaded() return resolveCodexAutomationRunUsage(input, { getState: () => this.state, refresh: (force) => this.refresh(force), diff --git a/src/main/durable-file-write.test.ts b/src/main/durable-file-write.test.ts index c2a44bf384c..5ea2d85b959 100644 --- a/src/main/durable-file-write.test.ts +++ b/src/main/durable-file-write.test.ts @@ -1,8 +1,13 @@ -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' -import { writeFileDurable, writeFileDurableSync } from './durable-file-write' +import { + durableWriteTempPath, + removeStaleDurableWriteTempFiles, + writeFileDurable, + writeFileDurableSync +} from './durable-file-write' describe('durable file write', () => { let dir: string @@ -84,4 +89,56 @@ describe('durable file write', () => { writeFileDurableSync(`${final}.b.tmp`, final, 'from-sync') expect(readFileSync(final, 'utf-8')).toBe('from-sync') }) + + it('preserves unowned temp names and tags owned writes', () => { + const final = join(dir, 'state.json') + const unowned = durableWriteTempPath(final) + const owned = durableWriteTempPath(final, 'worker-1') + + expect(unowned.startsWith(`${final}.${process.pid}.`)).toBe(true) + expect(unowned.endsWith('.tmp')).toBe(true) + expect(unowned).not.toContain('.owner-') + expect(owned.startsWith(`${final}.${process.pid}.`)).toBe(true) + expect(owned.endsWith('.owner-worker-1.tmp')).toBe(true) + }) + + it('reclaims only the retired owner in this process', async () => { + const final = join(dir, 'state.json') + const retired = durableWriteTempPath(final, 'worker-1') + const replacement = durableWriteTempPath(final, 'worker-2') + const unowned = durableWriteTempPath(final) + const otherProcess = `${final}.${process.pid + 1}.1.test.owner-worker-1.tmp` + const otherTarget = durableWriteTempPath(join(dir, 'other-state.json'), 'worker-1') + const temps = [retired, replacement, unowned, otherProcess, otherTarget] + for (const temp of temps) { + writeFileSync(temp, 'pending') + } + writeFileSync(final, 'committed') + + await removeStaleDurableWriteTempFiles(final, { retiredOwner: 'worker-1' }) + + expect(existsSync(retired)).toBe(false) + for (const live of [replacement, unowned, otherProcess, otherTarget]) { + expect(existsSync(live)).toBe(true) + } + expect(readFileSync(final, 'utf-8')).toBe('committed') + }) + + it('keeps current-process temps during the default stale sweep', async () => { + const final = join(dir, 'state.json') + const unowned = durableWriteTempPath(final) + const owned = durableWriteTempPath(final, 'worker-1') + const otherUnowned = `${final}.${process.pid + 1}.1.test.tmp` + const otherOwned = `${final}.${process.pid + 1}.1.test.owner-worker-1.tmp` + for (const temp of [unowned, owned, otherUnowned, otherOwned]) { + writeFileSync(temp, 'pending') + } + + await removeStaleDurableWriteTempFiles(final) + + expect(existsSync(unowned)).toBe(true) + expect(existsSync(owned)).toBe(true) + expect(existsSync(otherUnowned)).toBe(false) + expect(existsSync(otherOwned)).toBe(false) + }) }) diff --git a/src/main/durable-file-write.ts b/src/main/durable-file-write.ts index e573b4ecd7f..6e15a76f872 100644 --- a/src/main/durable-file-write.ts +++ b/src/main/durable-file-write.ts @@ -167,29 +167,38 @@ export async function writeFileDurableIfCurrent( } /** Temp path for a durable write. Shared shape so `removeStaleDurableWriteTempFiles` can reclaim orphans. */ -export function durableWriteTempPath(finalPath: string): string { - return `${finalPath}.${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}.tmp` +export function durableWriteTempPath(finalPath: string, owner?: string): string { + const ownerSuffix = owner === undefined ? '' : `.owner-${owner}` + return `${finalPath}.${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}${ownerSuffix}.tmp` } /** * Sweep temp files orphaned by a death between write and rename — for multi-MB payloads they would * otherwise accumulate forever. Callers can require a minimum age to spare another live instance's - * write. This process's own temps are always skipped because deleting one would fail its rename. + * write. This process's own temps are skipped unless their owner is known to have exited. */ export async function removeStaleDurableWriteTempFiles( finalPath: string, - options: { minimumAgeMs?: number } = {} + options: { minimumAgeMs?: number; retiredOwner?: string } = {} ): Promise { const directory = dirname(finalPath) const prefix = `${basename(finalPath)}.` const ownPrefix = `${prefix}${process.pid}.` + const retiredOwnerSuffix = + options.retiredOwner === undefined ? null : `.owner-${options.retiredOwner}.tmp` try { const names = await readdir(directory) await Promise.all( names - .filter( - (name) => name.startsWith(prefix) && name.endsWith('.tmp') && !name.startsWith(ownPrefix) - ) + .filter((name) => { + if (!name.startsWith(prefix) || !name.endsWith('.tmp')) { + return false + } + // The caller must await owner exit before reclaiming its current-process files. + return retiredOwnerSuffix === null + ? !name.startsWith(ownPrefix) + : name.startsWith(ownPrefix) && name.endsWith(retiredOwnerSuffix) + }) .map(async (name) => { const path = join(directory, name) if (options.minimumAgeMs) { diff --git a/src/main/ipc/usage-provider-handlers.test.ts b/src/main/ipc/usage-provider-handlers.test.ts index 7309e9b0429..a8d15f7a66b 100644 --- a/src/main/ipc/usage-provider-handlers.test.ts +++ b/src/main/ipc/usage-provider-handlers.test.ts @@ -7,8 +7,9 @@ vi.mock('electron', () => ({ ipcMain: { handle } })) import { registerUsageProviderHandlers } from './usage-provider-handlers' describe('usage provider IPC handlers', () => { - it('registers every provider route and forwards query arguments', () => { + it('registers every provider route and forwards query arguments', async () => { const createUsage = () => ({ + whenLoaded: vi.fn(() => Promise.resolve()), getScanState: vi.fn(), setEnabled: vi.fn(), refresh: vi.fn(), @@ -31,7 +32,7 @@ describe('usage provider IPC handlers', () => { }) const prefixes = ['claudeUsage', 'codexUsage', 'openCodeUsage', 'museUsage'] - const suffixes = Object.keys(claudeUsage) + const suffixes = Object.keys(claudeUsage).filter((key) => key !== 'whenLoaded') expect(handle.mock.calls.map(([channel]) => channel)).toEqual( prefixes.flatMap((prefix) => suffixes.map((suffix) => `${prefix}:${suffix}`)) ) @@ -42,18 +43,18 @@ describe('usage provider IPC handlers', () => { )?.[1] as ((event: unknown, args?: unknown) => unknown) | undefined return handler?.({}, args) } - call('claudeUsage', 'getScanState') - call('codexUsage', 'getScanState') - call('openCodeUsage', 'getScanState') - call('museUsage', 'getScanState') - call('claudeUsage', 'setEnabled', { enabled: true }) - call('claudeUsage', 'refresh') - call('claudeUsage', 'refresh', { force: true }) - call('claudeUsage', 'getSnapshot', { scope: 'orca', range: '30d', limit: 7 }) - call('claudeUsage', 'getSummary', { scope: 'all', range: '7d' }) - call('claudeUsage', 'getDaily', { scope: 'orca', range: '90d' }) - call('claudeUsage', 'getBreakdown', { scope: 'all', range: 'all', kind: 'model' }) - call('claudeUsage', 'getRecentSessions', { scope: 'orca', range: '30d', limit: 4 }) + await call('claudeUsage', 'getScanState') + await call('codexUsage', 'getScanState') + await call('openCodeUsage', 'getScanState') + await call('museUsage', 'getScanState') + await call('claudeUsage', 'setEnabled', { enabled: true }) + await call('claudeUsage', 'refresh') + await call('claudeUsage', 'refresh', { force: true }) + await call('claudeUsage', 'getSnapshot', { scope: 'orca', range: '30d', limit: 7 }) + await call('claudeUsage', 'getSummary', { scope: 'all', range: '7d' }) + await call('claudeUsage', 'getDaily', { scope: 'orca', range: '90d' }) + await call('claudeUsage', 'getBreakdown', { scope: 'all', range: 'all', kind: 'model' }) + await call('claudeUsage', 'getRecentSessions', { scope: 'orca', range: '30d', limit: 4 }) expect(claudeUsage.getScanState).toHaveBeenCalledWith() expect(codexUsage.getScanState).toHaveBeenCalledWith() @@ -67,4 +68,46 @@ describe('usage provider IPC handlers', () => { expect(claudeUsage.getBreakdown).toHaveBeenCalledWith('all', 'all', 'model') expect(claudeUsage.getRecentSessions).toHaveBeenCalledWith('orca', '30d', 4) }) + + it('answers the synchronous reads only once persisted state has loaded', async () => { + handle.mockClear() + let finishLoad = () => {} + const loaded = new Promise((resolve) => { + finishLoad = resolve + }) + const createUsage = () => ({ + whenLoaded: () => loaded, + getScanState: vi.fn(() => 'scan-state'), + setEnabled: vi.fn(), + refresh: vi.fn(), + getSnapshot: vi.fn(() => 'snapshot'), + getSummary: vi.fn(), + getDaily: vi.fn(), + getBreakdown: vi.fn(), + getRecentSessions: vi.fn() + }) + const claudeUsage = createUsage() + const codexUsage = createUsage() + const openCodeUsage = createUsage() + const museUsage = createUsage() + registerUsageProviderHandlers({ + claudeUsage: claudeUsage as never, + codexUsage: codexUsage as never, + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the mock implements every method the registrar calls on a usage store. + museUsage: museUsage as never, + openCodeUsage: openCodeUsage as never + }) + const handler = (suffix: string): ((event: unknown, args?: unknown) => Promise) => + handle.mock.calls.find(([channel]) => channel === `claudeUsage:${suffix}`)?.[1] + + const scanState = handler('getScanState')({}) + const snapshot = handler('getSnapshot')({}, { scope: 'all', range: 'all' }) + await Promise.resolve() + expect(claudeUsage.getScanState).not.toHaveBeenCalled() + expect(claudeUsage.getSnapshot).not.toHaveBeenCalled() + + finishLoad() + await expect(scanState).resolves.toBe('scan-state') + await expect(snapshot).resolves.toBe('snapshot') + }) }) diff --git a/src/main/ipc/usage-provider-handlers.ts b/src/main/ipc/usage-provider-handlers.ts index 2ede51cf43c..0f48f73b973 100644 --- a/src/main/ipc/usage-provider-handlers.ts +++ b/src/main/ipc/usage-provider-handlers.ts @@ -14,6 +14,7 @@ type UsageProviderStores = { type UsageProviderChannelPrefix = keyof UsageProviderStores type UsageProviderHandlerStore = { + whenLoaded: () => Promise getScanState: () => unknown setEnabled: (enabled: boolean) => unknown refresh: (force?: boolean) => unknown @@ -33,7 +34,11 @@ function registerProviderHandlers( prefix: UsageProviderChannelPrefix, usage: UsageProviderHandlerStore ): void { - ipcMain.handle(`${prefix}:getScanState`, () => usage.getScanState()) + // Why: persisted state can still be loading on the worker; the sync getters must not see defaults. + ipcMain.handle(`${prefix}:getScanState`, async () => { + await usage.whenLoaded() + return usage.getScanState() + }) ipcMain.handle(`${prefix}:setEnabled`, (_event, args: { enabled: boolean }) => usage.setEnabled(args.enabled) ) @@ -42,8 +47,10 @@ function registerProviderHandlers( ) ipcMain.handle( `${prefix}:getSnapshot`, - (_event, args: UsageRangeArgs & { limit?: number }) => - usage.getSnapshot(args.scope, args.range, args.limit) + async (_event, args: UsageRangeArgs & { limit?: number }) => { + await usage.whenLoaded() + return usage.getSnapshot(args.scope, args.range, args.limit) + } ) ipcMain.handle(`${prefix}:getSummary`, (_event, args: UsageRangeArgs) => usage.getSummary(args.scope, args.range) diff --git a/src/main/muse-usage/muse-usage-provider.ts b/src/main/muse-usage/muse-usage-provider.ts index bbcd83dffa0..158eaf1779c 100644 --- a/src/main/muse-usage/muse-usage-provider.ts +++ b/src/main/muse-usage/muse-usage-provider.ts @@ -1,6 +1,6 @@ import type { UsageProvider } from '../usage/usage-provider-contract' import { scanMuseUsageFilesViaWorker } from '../usage/usage-scan-worker-spawn' -import type { MuseUsageDailyAggregate, MuseUsagePersistedFile, MuseUsageSession } from './types' +import type { MuseUsageDailyAggregate, MuseUsageSession } from './types' // v2: event keys carry a per-log ordinal. export const MUSE_USAGE_SCHEMA_VERSION = 2 @@ -10,9 +10,4 @@ export const museUsageProvider = { label: 'Muse Code', schemaVersion: MUSE_USAGE_SCHEMA_VERSION, scan: scanMuseUsageFilesViaWorker -} satisfies UsageProvider< - 'processedFiles', - MuseUsagePersistedFile, - MuseUsageSession, - MuseUsageDailyAggregate -> +} satisfies UsageProvider diff --git a/src/main/muse-usage/persisted-file-normalization.ts b/src/main/muse-usage/persisted-file-normalization.ts new file mode 100644 index 00000000000..75b9fedbc19 --- /dev/null +++ b/src/main/muse-usage/persisted-file-normalization.ts @@ -0,0 +1,16 @@ +import type { MuseUsagePersistedFile } from './types' + +/** Fills fields older Muse caches never wrote. Electron-free: the scan worker runs it too. */ +export function normalizeMuseUsagePersistedFiles( + files: MuseUsagePersistedFile[] | undefined +): MuseUsagePersistedFile[] { + return (files ?? []).map((file) => ({ + ...file, + sessions: file.sessions ?? [], + dailyAggregates: file.dailyAggregates ?? [], + ownedEventKeys: file.ownedEventKeys ?? [], + hasDeferredClaims: file.hasDeferredClaims ?? true, + sessionCwd: file.sessionCwd ?? null, + inheritedCwd: file.inheritedCwd ?? null + })) +} diff --git a/src/main/muse-usage/persisted-state-normalization.ts b/src/main/muse-usage/persisted-state-normalization.ts index 6290d5e0a65..84e2132f105 100644 --- a/src/main/muse-usage/persisted-state-normalization.ts +++ b/src/main/muse-usage/persisted-state-normalization.ts @@ -1,4 +1,5 @@ import { MUSE_USAGE_SCHEMA_VERSION } from './muse-usage-provider' +import { normalizeMuseUsagePersistedFiles } from './persisted-file-normalization' import type { MuseUsagePersistedState } from './types' export function getDefaultMuseUsageState(): MuseUsagePersistedState { @@ -28,15 +29,7 @@ export function normalizeMuseUsageState(state: MuseUsagePersistedState): MuseUsa } return { ...state, - processedFiles: (state.processedFiles ?? []).map((file) => ({ - ...file, - sessions: file.sessions ?? [], - dailyAggregates: file.dailyAggregates ?? [], - ownedEventKeys: file.ownedEventKeys ?? [], - hasDeferredClaims: file.hasDeferredClaims ?? true, - sessionCwd: file.sessionCwd ?? null, - inheritedCwd: file.inheritedCwd ?? null - })), + processedFiles: normalizeMuseUsagePersistedFiles(state.processedFiles), sessions: state.sessions ?? [], dailyAggregates: state.dailyAggregates ?? [] } diff --git a/src/main/muse-usage/store.ts b/src/main/muse-usage/store.ts index b1b22996848..cc8722643ab 100644 --- a/src/main/muse-usage/store.ts +++ b/src/main/muse-usage/store.ts @@ -12,6 +12,7 @@ import type { } from '../../shared/muse-usage-types' import type { Store } from '../persistence' import { UsageProviderStoreLifecycle } from '../usage/usage-provider-store-lifecycle' +import { splitUsageCacheFileViaWorker } from '../usage/usage-scan-worker-spawn' import { filterUsageDaily, filterUsageSessions } from '../usage/usage-scope-filters' import { museUsageProvider } from './muse-usage-provider' import { getDefaultMuseUsageState, normalizeMuseUsageState } from './persisted-state-normalization' @@ -47,7 +48,8 @@ export class MuseUsageStore extends UsageProviderStoreLifecycle< normalizeState: normalizeMuseUsageState, sourceKey: 'processedFiles', dataPresenceKey: 'hasAnyMuseData', - scan: museUsageProvider.scan + scan: museUsageProvider.scan, + splitCacheFile: splitUsageCacheFileViaWorker }) } diff --git a/src/main/opencode-usage/opencode-usage-provider.ts b/src/main/opencode-usage/opencode-usage-provider.ts index 6e815fb1727..77b34ed4720 100644 --- a/src/main/opencode-usage/opencode-usage-provider.ts +++ b/src/main/opencode-usage/opencode-usage-provider.ts @@ -1,10 +1,6 @@ import type { UsageProvider } from '../usage/usage-provider-contract' import { scanOpenCodeUsageDatabasesViaWorker } from '../usage/usage-scan-worker-spawn' -import type { - OpenCodeUsageDailyAggregate, - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession -} from './types' +import type { OpenCodeUsageDailyAggregate, OpenCodeUsageSession } from './types' // Why: v4 reads OpenCode 2's `session_v2` table; v3 caches miss every v2 session. // v5 merges a migrated session's two rows per column instead of picking one, so @@ -17,9 +13,4 @@ export const openCodeUsageProvider = { label: 'OpenCode', schemaVersion: OPENCODE_USAGE_SCHEMA_VERSION, scan: scanOpenCodeUsageDatabasesViaWorker -} satisfies UsageProvider< - 'processedDatabases', - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession, - OpenCodeUsageDailyAggregate -> +} satisfies UsageProvider diff --git a/src/main/opencode-usage/persisted-database-normalization.ts b/src/main/opencode-usage/persisted-database-normalization.ts new file mode 100644 index 00000000000..aae8138cf78 --- /dev/null +++ b/src/main/opencode-usage/persisted-database-normalization.ts @@ -0,0 +1,46 @@ +import type { + OpenCodeUsageDailyAggregate, + OpenCodeUsagePersistedDatabase, + OpenCodeUsagePersistedState +} from './types' + +/** Fills cost fields older OpenCode caches never wrote. Electron-free: the scan worker runs it too. */ +export function normalizeOpenCodeUsagePersistedDatabases( + databases: OpenCodeUsagePersistedDatabase[] | undefined +): OpenCodeUsagePersistedDatabase[] { + return (databases ?? []).map((database) => ({ + ...database, + sessions: (database.sessions ?? []).map(normalizeSessionCost), + dailyAggregates: (database.dailyAggregates ?? []).map(normalizeDailyAggregateCost) + })) +} + +export function normalizeDailyAggregateCost( + entry: OpenCodeUsageDailyAggregate +): OpenCodeUsageDailyAggregate { + return { + ...entry, + estimatedCostUsd: entry.estimatedCostUsd ?? null + } +} + +export function normalizeSessionCost( + session: OpenCodeUsagePersistedState['sessions'][number] +): OpenCodeUsagePersistedState['sessions'][number] { + return { + ...session, + estimatedCostUsd: session.estimatedCostUsd ?? null, + locationBreakdown: (session.locationBreakdown ?? []).map((entry) => ({ + ...entry, + estimatedCostUsd: entry.estimatedCostUsd ?? null + })), + modelBreakdown: (session.modelBreakdown ?? []).map((entry) => ({ + ...entry, + estimatedCostUsd: entry.estimatedCostUsd ?? null + })), + locationModelBreakdown: (session.locationModelBreakdown ?? []).map((entry) => ({ + ...entry, + estimatedCostUsd: entry.estimatedCostUsd ?? null + })) + } +} diff --git a/src/main/opencode-usage/persisted-state-normalization.ts b/src/main/opencode-usage/persisted-state-normalization.ts index b84178bcc30..3382b550e4a 100644 --- a/src/main/opencode-usage/persisted-state-normalization.ts +++ b/src/main/opencode-usage/persisted-state-normalization.ts @@ -1,5 +1,10 @@ -import type { OpenCodeUsageDailyAggregate, OpenCodeUsagePersistedState } from './types' +import type { OpenCodeUsagePersistedState } from './types' import { OPENCODE_USAGE_SCHEMA_VERSION } from './opencode-usage-provider' +import { + normalizeDailyAggregateCost, + normalizeOpenCodeUsagePersistedDatabases, + normalizeSessionCost +} from './persisted-database-normalization' const SCHEMA_VERSION = OPENCODE_USAGE_SCHEMA_VERSION @@ -31,42 +36,8 @@ export function normalizePersistedState( } return { ...state, - processedDatabases: (state.processedDatabases ?? []).map((database) => ({ - ...database, - sessions: (database.sessions ?? []).map(normalizeSessionCost), - dailyAggregates: (database.dailyAggregates ?? []).map(normalizeDailyAggregateCost) - })), + processedDatabases: normalizeOpenCodeUsagePersistedDatabases(state.processedDatabases), sessions: state.sessions.map(normalizeSessionCost), dailyAggregates: state.dailyAggregates.map(normalizeDailyAggregateCost) } } - -function normalizeDailyAggregateCost( - entry: OpenCodeUsageDailyAggregate -): OpenCodeUsageDailyAggregate { - return { - ...entry, - estimatedCostUsd: entry.estimatedCostUsd ?? null - } -} - -function normalizeSessionCost( - session: OpenCodeUsagePersistedState['sessions'][number] -): OpenCodeUsagePersistedState['sessions'][number] { - return { - ...session, - estimatedCostUsd: session.estimatedCostUsd ?? null, - locationBreakdown: (session.locationBreakdown ?? []).map((entry) => ({ - ...entry, - estimatedCostUsd: entry.estimatedCostUsd ?? null - })), - modelBreakdown: (session.modelBreakdown ?? []).map((entry) => ({ - ...entry, - estimatedCostUsd: entry.estimatedCostUsd ?? null - })), - locationModelBreakdown: (session.locationModelBreakdown ?? []).map((entry) => ({ - ...entry, - estimatedCostUsd: entry.estimatedCostUsd ?? null - })) - } -} diff --git a/src/main/opencode-usage/store.test.ts b/src/main/opencode-usage/store.test.ts index 50b530cff2a..57fa7928db9 100644 --- a/src/main/opencode-usage/store.test.ts +++ b/src/main/opencode-usage/store.test.ts @@ -19,7 +19,8 @@ vi.mock('electron', () => ({ })) vi.mock('../usage/usage-scan-worker-spawn', () => ({ - scanOpenCodeUsageDatabasesViaWorker: vi.fn() + scanOpenCodeUsageDatabasesViaWorker: vi.fn(), + splitUsageCacheFileViaWorker: vi.fn() })) import { OpenCodeUsageStore, initOpenCodeUsagePath } from './store' @@ -29,7 +30,6 @@ import { scanOpenCodeUsageDatabasesViaWorker } from '../usage/usage-scan-worker- function createEmptyScanResult() { return { - processedDatabases: [], sessions: [], dailyAggregates: [] } @@ -175,7 +175,7 @@ describe('OpenCodeUsageStore', () => { rmSync(tempUserData, { recursive: true, force: true }) }) - it('adapts OpenCode scans to pretty-printed cache persistence', async () => { + it('hands OpenCode scans the worker-owned source cache and persists a compact report', async () => { const store = createStoreWithState({ scanState: { enabled: true, @@ -188,8 +188,13 @@ describe('OpenCodeUsageStore', () => { await store.refresh(true) const persistedJson = readFileSync(join(tempUserData, 'orca-opencode-usage.json'), 'utf-8') - expect(scanOpenCodeUsageDatabasesViaWorker).toHaveBeenCalledWith([], []) - expect(persistedJson).toContain('\n') + expect(scanOpenCodeUsageDatabasesViaWorker).toHaveBeenCalledWith([], { + path: join(tempUserData, 'orca-opencode-usage-sources.json'), + schemaVersion: OPENCODE_USAGE_SCHEMA_VERSION, + worktreeFingerprint: '[]', + reuse: false + }) + expect(persistedJson).not.toContain('\n') }) it('reports no data for Orca scope when only non-Orca OpenCode usage exists', async () => { diff --git a/src/main/opencode-usage/store.ts b/src/main/opencode-usage/store.ts index 3bdcefeb0a7..7b3b656a88f 100644 --- a/src/main/opencode-usage/store.ts +++ b/src/main/opencode-usage/store.ts @@ -30,6 +30,7 @@ import { buildOpenCodeUsageSummary } from './snapshot-rollups' import { UsageProviderStoreLifecycle } from '../usage/usage-provider-store-lifecycle' +import { splitUsageCacheFileViaWorker } from '../usage/usage-scan-worker-spawn' let _openCodeUsageFile: string | null = null @@ -61,8 +62,8 @@ export class OpenCodeUsageStore extends UsageProviderStoreLifecycle< normalizeState: normalizePersistedState, sourceKey: 'processedDatabases', dataPresenceKey: 'hasAnyOpenCodeData', - jsonIndent: 2, - scan: openCodeUsageProvider.scan + scan: openCodeUsageProvider.scan, + splitCacheFile: splitUsageCacheFileViaWorker }) } diff --git a/src/main/usage/usage-provider-contract.ts b/src/main/usage/usage-provider-contract.ts index bbff062dc30..efaa1d6a6a3 100644 --- a/src/main/usage/usage-provider-contract.ts +++ b/src/main/usage/usage-provider-contract.ts @@ -1,3 +1,5 @@ +import type { UsageSourceCacheRef } from './usage-source-cache-file' + /** * Seam a usage source implements to be scanned by Orca, including plugin-contributed ones. * @@ -23,25 +25,19 @@ export type UsageScanWorktreeRef = { displayName: string } -/** - * `TSourceKey` is the provider's own name for its per-source scan cache (`processedFiles`, - * `processedDatabases`). It stays a type parameter because those names are persisted on disk. - */ -export type UsageScanResult = Record< - TSourceKey, - readonly TSource[] -> & { +/** What a scan reports. The per-source scan cache stays wherever the scan ran, behind `sourceCache`. */ +export type UsageScanResult = { sessions: readonly TSession[] dailyAggregates: readonly TDaily[] } -export type UsageProvider = { +export type UsageProvider = { readonly id: UsageProviderId readonly label: string /** Bumped when the persisted projection changes shape; older caches are discarded. */ readonly schemaVersion: number scan( worktrees: UsageScanWorktreeRef[], - previous: TSource[] - ): Promise> + sourceCache: UsageSourceCacheRef + ): Promise> } diff --git a/src/main/usage/usage-provider-store-lifecycle.test.ts b/src/main/usage/usage-provider-store-lifecycle.test.ts index cef33bf6d89..0ab96ae7900 100644 --- a/src/main/usage/usage-provider-store-lifecycle.test.ts +++ b/src/main/usage/usage-provider-store-lifecycle.test.ts @@ -9,6 +9,12 @@ import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { UsageScanWorktreeRef } from './usage-provider-contract' import { UsageProviderStoreLifecycle } from './usage-provider-store-lifecycle' +import { + splitUsageCacheFile, + type UsageCacheSplitRequest, + type UsageCacheSplitResult, + type UsageSourceCacheRef +} from './usage-source-cache-file' const { writeProbe } = vi.hoisted(() => ({ writeProbe: { @@ -56,11 +62,12 @@ type TestState = { dailyAggregates: TestDailyAggregate[] scanState: TestScanState } -type TestScanResult = Pick +type TestScanResult = Pick type TestScan = ( worktrees: UsageScanWorktreeRef[], - previous: TestSource[] + sourceCache: UsageSourceCacheRef ) => Promise +type TestSplit = (request: UsageCacheSplitRequest) => Promise const NOW = Date.parse('2026-04-10T16:00:00.000Z') const EMPTY_WORKTREE_FINGERPRINT = '[]' @@ -87,7 +94,7 @@ function makeState( } function emptyScanResult(): TestScanResult { - return { processedSources: [], sessions: [], dailyAggregates: [] } + return { sessions: [], dailyAggregates: [] } } function createDeferred(): { promise: Promise; resolve: (value: T) => void } { @@ -103,7 +110,7 @@ class TestUsageStore extends UsageProviderStoreLifecycle< TestState, 'hasAnyTestData' > { - constructor(cacheFile: string, scan: TestScan) { + constructor(cacheFile: string, scan: TestScan, splitCacheFile: TestSplit) { super( { getRepos: () => [], @@ -127,7 +134,8 @@ class TestUsageStore extends UsageProviderStoreLifecycle< normalizeState: (state) => state, sourceKey: 'processedSources', dataPresenceKey: 'hasAnyTestData', - scan + scan, + splitCacheFile } ) } @@ -145,11 +153,12 @@ describe('UsageProviderStoreLifecycle', () => { let tempDirectory: string let stores: TestUsageStore[] let scan: ReturnType> + let split: ReturnType> function createStore( cacheFile = join(tempDirectory, `usage-${stores.length}.json`) ): TestUsageStore { - const store = new TestUsageStore(cacheFile, scan) + const store = new TestUsageStore(cacheFile, scan, split) stores.push(store) return store } @@ -162,6 +171,8 @@ describe('UsageProviderStoreLifecycle', () => { writeProbe.blocked = false writeProbe.waiters = [] scan = vi.fn().mockResolvedValue(emptyScanResult()) + // The real split, in-process: the worker only adds the thread around it. + split = vi.fn(splitUsageCacheFile) vi.spyOn(Date, 'now').mockReturnValue(NOW) }) @@ -261,27 +272,152 @@ describe('UsageProviderStoreLifecycle', () => { it('invalidates prior sources on fingerprint changes and reuses them for forced scans', async () => { const store = createStore() - const refreshedSources = [{ id: 'refreshed' }] - scan.mockResolvedValueOnce({ - processedSources: refreshedSources, - sessions: [], - dailyAggregates: [] - }) + const sourceCache = { + path: join(tempDirectory, 'usage-0-sources.json'), + schemaVersion: 1, + worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT + } store.replaceState( makeState({ worktreeFingerprint: 'outdated', - processedSources: [{ id: 'stale' }], scanState: { enabled: true, lastScanCompletedAt: NOW - 1 } }) ) await store.refresh() - expect(scan).toHaveBeenLastCalledWith([], []) + expect(scan).toHaveBeenLastCalledWith([], { ...sourceCache, reuse: false }) scan.mockClear() await store.refresh(true) - expect(scan).toHaveBeenCalledWith([], refreshedSources) + expect(scan).toHaveBeenCalledWith([], { ...sourceCache, reuse: true }) + }) + + it('keeps per-source records out of the report main writes', async () => { + const cacheFile = join(tempDirectory, 'provider.json') + const store = createStore(cacheFile) + await store.setEnabled(true) + scan.mockResolvedValueOnce({ sessions: [{ id: 'session' }], dailyAggregates: [] }) + + await store.refresh(true) + await store.flush() + + const persisted = readFileSync(cacheFile, 'utf-8') + expect(JSON.parse(persisted)).not.toHaveProperty('processedSources') + expect(JSON.parse(persisted).sessions).toEqual([{ id: 'session' }]) + // Compact: the report is rewritten after every scan. + expect(persisted).not.toContain('\n') + }) + + it('preserves a small legacy cache for the next warm scan', async () => { + const cacheFile = join(tempDirectory, 'provider.json') + writeFileSync( + cacheFile, + JSON.stringify( + makeState({ + worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT, + processedSources: [{ id: 'inline' }], + sessions: [{ id: 'session' }] + }) + ) + ) + + const store = createStore(cacheFile) + + await store.whenLoaded() + expect(split).toHaveBeenCalledWith({ cacheFile, sourceKey: 'processedSources' }) + expect(store.getState().sessions).toEqual([{ id: 'session' }]) + expect(store.getState().processedSources).toEqual([]) + expect( + JSON.parse(readFileSync(join(tempDirectory, 'provider-sources.json'), 'utf-8')) + ).toMatchObject({ + sources: [{ id: 'inline' }] + }) + }) + + it('splits a large legacy cache on the worker before any reader or writer sees it', async () => { + const cacheFile = join(tempDirectory, 'provider.json') + // Past the main-thread parse ceiling, the shape a pre-sidecar Claude history has. + const processedSources = Array.from({ length: 9_000 }, (_, index) => ({ + id: `source-${index}-${'x'.repeat(1_000)}` + })) + writeFileSync( + cacheFile, + JSON.stringify( + makeState({ + worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT, + processedSources, + sessions: [{ id: 'session' }], + scanState: { enabled: true, lastScanCompletedAt: NOW - 1 } + }), + null, + 2 + ) + ) + + const store = createStore(cacheFile) + // A write racing the load must not clobber the history being loaded. + const disabled = store.setEnabled(false) + + expect(split).toHaveBeenCalledWith({ cacheFile, sourceKey: 'processedSources' }) + await store.whenLoaded() + await disabled + await store.flush() + expect(store.getState().sessions).toEqual([{ id: 'session' }]) + expect(store.getState().processedSources).toEqual([]) + const report = JSON.parse(readFileSync(cacheFile, 'utf-8')) + expect(report).not.toHaveProperty('processedSources') + expect(report).toMatchObject({ sessions: [{ id: 'session' }], scanState: { enabled: false } }) + const sidecar = JSON.parse(readFileSync(join(tempDirectory, 'provider-sources.json'), 'utf-8')) + expect(sidecar).toEqual({ + schemaVersion: 1, + worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT, + sources: processedSources + }) + }) + + it('reads a large cache on the main thread when the worker cannot split it', async () => { + const cacheFile = join(tempDirectory, 'provider.json') + writeFileSync( + cacheFile, + JSON.stringify( + makeState({ + sessions: [{ id: 'session' }], + processedSources: [{ id: 'x'.repeat(9 * 1024 * 1024) }] + }) + ) + ) + split.mockRejectedValueOnce(new Error('worker unavailable')) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + + const store = createStore(cacheFile) + await store.whenLoaded() + + expect(store.getState().sessions).toEqual([{ id: 'session' }]) + expect(store.getState().processedSources).toEqual([]) + }) + + it('flush waits for a pending migration and its report rewrite', async () => { + const cacheFile = join(tempDirectory, 'provider.json') + writeFileSync( + cacheFile, + JSON.stringify(makeState({ processedSources: [{ id: 'inline' }] })) + + ' '.repeat(9 * 1024 * 1024) + ) + const pendingSplit = createDeferred() + split.mockReturnValueOnce(pendingSplit.promise) + const store = createStore(cacheFile) + let flushed = false + const flushing = store.flush().then(() => { + flushed = true + }) + + await new Promise((resolve) => setImmediate(resolve)) + expect(flushed).toBe(false) + pendingSplit.resolve({ reportText: JSON.stringify(makeState()), migrated: true }) + await flushing + + expect(JSON.parse(readFileSync(cacheFile, 'utf-8'))).not.toHaveProperty('processedSources') }) it('shares one in-flight scan and exposes its live state', async () => { @@ -304,7 +440,6 @@ describe('UsageProviderStoreLifecycle', () => { expect(writeProbe.opens).toBe(0) pendingScan.resolve({ - processedSources: [{ id: 'source' }], sessions: [{ id: 'session' }], dailyAggregates: [] }) @@ -316,7 +451,7 @@ describe('UsageProviderStoreLifecycle', () => { hasAnyTestData: true }) expect(writeProbe.opens).toBe(1) - expect(store.getState().processedSources).toEqual([{ id: 'source' }]) + expect(store.getState().sessions).toEqual([{ id: 'session' }]) }) it('vetoes a superseded generation before rename', async () => { @@ -352,7 +487,6 @@ describe('UsageProviderStoreLifecycle', () => { const store = createStore() const previousState = makeState({ worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT, - processedSources: [{ id: 'source' }], sessions: [{ id: 'session' }], dailyAggregates: [{ day: '2026-04-09' }], scanState: { enabled: true, lastScanCompletedAt: NOW - 1_000 } @@ -368,7 +502,6 @@ describe('UsageProviderStoreLifecycle', () => { expect(store.getState()).toMatchObject({ worktreeFingerprint: EMPTY_WORKTREE_FINGERPRINT, - processedSources: previousState.processedSources, sessions: previousState.sessions, dailyAggregates: previousState.dailyAggregates }) diff --git a/src/main/usage/usage-provider-store-lifecycle.ts b/src/main/usage/usage-provider-store-lifecycle.ts index 550f79e3ce5..fd0492d9b5b 100644 --- a/src/main/usage/usage-provider-store-lifecycle.ts +++ b/src/main/usage/usage-provider-store-lifecycle.ts @@ -1,4 +1,5 @@ -import { existsSync, readFileSync } from 'node:fs' +import { readFileSync, statSync } from 'node:fs' +import { readFile } from 'node:fs/promises' import { AgentTokenUsageReporter } from './agent-token-usage-reporter' import type { AgentTokenSession } from './agent-token-usage' import type { AgentTokenUsage } from '../../shared/telemetry-agent-token-usage-schema' @@ -10,8 +11,16 @@ import { UsageCacheSnapshotWriter } from '../usage-cache-snapshot-writer' import { loadKnownUsageWorktreesByRepo } from '../usage-worktree-metadata' import type { UsageScanWorktreeRef } from './usage-provider-contract' import { createWorktreeRefs, getUsageWorktreeFingerprint } from './usage-worktree-refs' +import { + usageSourceCachePath, + type UsageCacheSplitRequest, + type UsageCacheSplitResult, + type UsageSourceCacheRef +} from './usage-source-cache-file' const STALE_MS = 5 * 60_000 +// Keep large cache parsing on the worker; small reports retain synchronous reads. +const MAIN_THREAD_PARSE_MAX_BYTES = 8 * 1024 * 1024 type UsageProviderScanState = { enabled: boolean @@ -21,17 +30,13 @@ type UsageProviderScanState = { } type UsageProviderStoreState = { + schemaVersion: number worktreeFingerprint: string | null sessions: unknown[] dailyAggregates: unknown[] scanState: UsageProviderScanState } & Record -type UsageProviderScanProjection< - SourceKey extends string, - State extends UsageProviderStoreState -> = Pick - type UsageProviderStoreLifecycleConfig< SourceKey extends string, State extends UsageProviderStoreState, @@ -43,15 +48,15 @@ type UsageProviderStoreLifecycleConfig< normalizeState: (state: State) => State sourceKey: SourceKey dataPresenceKey: DataPresenceKey - jsonIndent?: number tokenUsage?: { provider: AgentTokenUsage['provider'] selectSessions: (state: State) => AgentTokenSession[] } scan: ( worktrees: UsageScanWorktreeRef[], - previous: State[SourceKey] - ) => Promise> + sourceCache: UsageSourceCacheRef + ) => Promise> + splitCacheFile: (request: UsageCacheSplitRequest) => Promise } type PublicUsageProviderScanState = UsageProviderScanState & { @@ -63,7 +68,10 @@ export abstract class UsageProviderStoreLifecycle< State extends UsageProviderStoreState, DataPresenceKey extends string > { + // Per-source records belong to the worker and are excluded from report state. protected state: State + private readonly loaded: Promise + private readonly schemaVersion: number private scanPromise: Promise | null = null private tokenReporter: AgentTokenUsageReporter | null = null private analyticsSessionIds: AnalyticsSessionIdStore | null = null @@ -74,7 +82,15 @@ export abstract class UsageProviderStoreLifecycle< private readonly config: UsageProviderStoreLifecycleConfig ) { this.writer = new UsageCacheSnapshotWriter(config.logTag, config.resolveCacheFile) - this.state = this.load() + const defaults = config.createDefaultState() + this.schemaVersion = defaults.schemaVersion + this.state = defaults + this.loaded = this.load() + } + + /** Resolves once persisted state is in memory; synchronous readers must wait for it first. */ + whenLoaded(): Promise { + return this.loaded } getScanState(): PublicUsageProviderScanState { @@ -99,17 +115,20 @@ export abstract class UsageProviderStoreLifecycle< /** Await queued cache writes so quit does not drop the final snapshot. */ async flush(): Promise { + await this.loaded await this.tokenReporter?.flush() await Promise.all([this.writer.flush(), this.analyticsSessionIds?.flush()]) } async setEnabled(enabled: boolean): Promise> { + await this.loaded this.state.scanState.enabled = enabled await this.writeToDisk() return this.getScanState() } async refresh(force = false): Promise> { + await this.loaded if (!this.state.scanState.enabled) { return this.getScanState() } @@ -125,28 +144,69 @@ export abstract class UsageProviderStoreLifecycle< } protected writeToDisk(): Promise { - return this.writer.write(() => JSON.stringify(this.state, null, this.config.jsonIndent)) + return this.writer.write(() => { + const { [this.config.sourceKey]: _sources, ...report } = this.state + return JSON.stringify(report) + }) } - private load(): State { - const defaults = this.config.createDefaultState() + private load(): Promise { + const cacheFile = this.config.resolveCacheFile() try { - const cacheFile = this.config.resolveCacheFile() - if (!existsSync(cacheFile)) { - return defaults + if (statSync(cacheFile).size > MAIN_THREAD_PARSE_MAX_BYTES) { + return this.loadOnWorker(cacheFile) + } + const parsed: State = JSON.parse(readFileSync(cacheFile, 'utf-8')) + this.state = this.normalizeReport(parsed) + if ( + Array.isArray(parsed[this.config.sourceKey]) && + parsed[this.config.sourceKey].length > 0 + ) { + return this.loadOnWorker(cacheFile) + } + } catch (error) { + if (!isMissingFileError(error)) { + console.error( + `${this.config.logTag} Failed to load persisted state, starting fresh:`, + error + ) + } + } + return Promise.resolve() + } + + private async loadOnWorker(cacheFile: string): Promise { + try { + const { reportText, migrated } = await this.config + .splitCacheFile({ cacheFile, sourceKey: this.config.sourceKey }) + .catch(async (error: unknown) => { + // Preserve usage history when the worker cannot start. + console.warn(`${this.config.logTag} Reading the usage cache on the main thread:`, error) + return { reportText: await readFile(cacheFile, 'utf-8'), migrated: false } + }) + if (reportText === null) { + return + } + this.state = this.normalizeReport(JSON.parse(reportText)) + if (migrated) { + // Future launches can read the compact report synchronously. + await this.writeToDisk().catch(() => {}) } - const parsed = JSON.parse(readFileSync(cacheFile, 'utf-8')) as State - return this.config.normalizeState({ - ...defaults, - ...parsed, - scanState: { ...defaults.scanState, ...parsed.scanState } - }) } catch (error) { console.error(`${this.config.logTag} Failed to load persisted state, starting fresh:`, error) - return defaults } } + private normalizeReport(parsed: State): State { + const defaults = this.config.createDefaultState() + return this.config.normalizeState({ + ...defaults, + ...parsed, + [this.config.sourceKey]: [], + scanState: { ...defaults.scanState, ...parsed.scanState } + }) + } + private async runScan(): Promise { if (this.scanPromise) { await this.scanPromise @@ -162,13 +222,12 @@ export abstract class UsageProviderStoreLifecycle< const repos = this.store.getRepos() const worktreesByRepo = loadKnownUsageWorktreesByRepo(this.store, repos) const worktreeFingerprint = getUsageWorktreeFingerprint(worktreesByRepo) - const result = await this.config.scan( - createWorktreeRefs(repos, worktreesByRepo), - this.state.worktreeFingerprint === worktreeFingerprint - ? this.state[this.config.sourceKey] - : this.config.createDefaultState()[this.config.sourceKey] - ) - this.state[this.config.sourceKey] = result[this.config.sourceKey] + const result = await this.config.scan(createWorktreeRefs(repos, worktreesByRepo), { + path: usageSourceCachePath(this.config.resolveCacheFile()), + schemaVersion: this.schemaVersion, + worktreeFingerprint, + reuse: this.state.worktreeFingerprint === worktreeFingerprint + }) this.state.sessions = result.sessions this.state.dailyAggregates = result.dailyAggregates this.state.worktreeFingerprint = worktreeFingerprint @@ -216,3 +275,7 @@ export abstract class UsageProviderStoreLifecycle< return getUsageWorktreeFingerprint(loadKnownUsageWorktreesByRepo(this.store, repos)) } } + +function isMissingFileError(error: unknown): boolean { + return error instanceof Error && 'code' in error && error.code === 'ENOENT' +} diff --git a/src/main/usage/usage-scan-worker-cache-cleanup.test.ts b/src/main/usage/usage-scan-worker-cache-cleanup.test.ts new file mode 100644 index 00000000000..aabf7d13b35 --- /dev/null +++ b/src/main/usage/usage-scan-worker-cache-cleanup.test.ts @@ -0,0 +1,102 @@ +import { once } from 'node:events' +import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import { Worker } from 'node:worker_threads' +import { build } from 'esbuild' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { durableWriteTempPath } from '../durable-file-write' +import { createUsageScanWorkerTransport } from './usage-scan-worker-cache-cleanup' +import type { UsageScanWorkerRequest } from './usage-scan-worker-protocol' +import { usageSourceCachePath } from './usage-source-cache-file' + +let directory: string +let cacheModule: string + +beforeAll(async () => { + directory = mkdtempSync(join(tmpdir(), 'orca-usage-worker-cleanup-')) + cacheModule = join(directory, 'source-cache.cjs') + await build({ + entryPoints: [resolve(__dirname, 'usage-source-cache-file.ts')], + outfile: cacheModule, + bundle: true, + platform: 'node', + format: 'cjs', + target: 'node22' + }) +}) + +afterAll(() => rmSync(directory, { recursive: true, force: true })) + +// Pause an actual durable write after opening its temp; terminating the worker skips its finally. +const BLOCKED_WRITE_WORKER = ` +const { parentPort, workerData } = require('node:worker_threads'); +const fs = require('node:fs/promises'); +const originalOpen = fs.open; +fs.open = async (...args) => { + const handle = await originalOpen(...args); + if (args[1] === 'w') { + await handle.writeFile('partial-cache'); + parentPort.postMessage(args[0]); + await new Promise(() => {}); + } + return handle; +}; +const cache = require(workerData.cacheModule); +parentPort.on('message', (request) => { + const ref = request.operation === 'scan' ? request.sourceCache : { + path: cache.usageSourceCachePath(request.cacheFile), + schemaVersion: 1, worktreeFingerprint: '[]', reuse: false + }; + cache.writeUsageSourceCache(ref, [{ id: 'source' }]); +}); +` + +describe('usage worker cache cleanup', () => { + it.each(['scan', 'splitCacheFile'] as const)( + 'reclaims only the exited %s worker temp while replacement writes survive', + async (operation) => { + const cacheFile = join(directory, `${operation}.json`) + const sourcePath = usageSourceCachePath(cacheFile) + const worker = new Worker(BLOCKED_WRITE_WORKER, { + eval: true, + workerData: { cacheModule } + }) + const transport = createUsageScanWorkerTransport(() => worker) + try { + const started = once(worker, 'message') + const request: UsageScanWorkerRequest = + operation === 'scan' + ? { + id: 1, + operation, + providerId: 'codex', + worktrees: [], + sourceCache: { + path: sourcePath, + schemaVersion: 1, + worktreeFingerprint: '[]', + reuse: false + } + } + : { id: 1, operation, cacheFile, sourceKey: 'processedFiles' } + transport.postMessage(request) + const [tempPath] = await started + if (typeof tempPath !== 'string') { + throw new Error('Worker did not report its temporary path') + } + const replacement = durableWriteTempPath(sourcePath, String(worker.threadId + 1)) + writeFileSync(replacement, 'replacement-cache') + expect(existsSync(tempPath)).toBe(true) + + await transport.terminate() + + expect(existsSync(tempPath)).toBe(false) + expect(existsSync(replacement)).toBe(true) + } finally { + await worker.terminate() + } + }, + 15_000 + ) +}) diff --git a/src/main/usage/usage-scan-worker-cache-cleanup.ts b/src/main/usage/usage-scan-worker-cache-cleanup.ts new file mode 100644 index 00000000000..9476d2b7a1a --- /dev/null +++ b/src/main/usage/usage-scan-worker-cache-cleanup.ts @@ -0,0 +1,39 @@ +import type { WorkerRequestTransport, WorkerThreadFactory } from '../lazy-worker-thread-host' +import { removeStaleDurableWriteTempFiles } from '../durable-file-write' +import type { UsageScanWorkerRequest } from './usage-scan-worker-protocol' +import { usageSourceCachePath } from './usage-source-cache-file' + +export function createUsageScanWorkerTransport( + factory: WorkerThreadFactory +): WorkerRequestTransport { + const worker = factory() + if (!('threadId' in worker) || typeof worker.threadId !== 'number' || worker.threadId < 1) { + return worker + } + const owner = String(worker.threadId) + const cachePaths = new Set() + return { + on: (...args) => worker.on(...args), + off: (...args) => worker.off(...args), + removeAllListeners: () => worker.removeAllListeners(), + unref: () => worker.unref(), + postMessage: (request: UsageScanWorkerRequest, transferList) => { + cachePaths.add( + request.operation === 'scan' + ? request.sourceCache.path + : usageSourceCachePath(request.cacheFile) + ) + worker.postMessage(request, transferList) + }, + terminate: async () => { + const code = await worker.terminate() + // Only this exited worker's files are safe to reclaim while its replacement is running. + await Promise.all( + [...cachePaths].map((path) => + removeStaleDurableWriteTempFiles(path, { retiredOwner: owner }) + ) + ) + return code + } + } +} diff --git a/src/main/usage/usage-scan-worker-client.test.ts b/src/main/usage/usage-scan-worker-client.test.ts index c8838920335..c010d1afaa0 100644 --- a/src/main/usage/usage-scan-worker-client.test.ts +++ b/src/main/usage/usage-scan-worker-client.test.ts @@ -5,10 +5,8 @@ import { UsageScanWorkerClient, scanCodexUsageOnWorker } from './usage-scan-worker-client' -import type { - UsageScanWorkerRequest, - UsageScanWorkerRequestBody -} from './usage-scan-worker-protocol' +import type { UsageScanWorkerRequest, UsageScanWorkerScanBody } from './usage-scan-worker-protocol' +import type { UsageSourceCacheRef } from './usage-source-cache-file' // A worker_threads stand-in the tests drive directly: it records posted requests // and lets a test emit message/error/exit without a built worker bundle. @@ -55,14 +53,29 @@ class FakeWorker { } function createClient(factory: () => FakeWorker): UsageScanWorkerClient { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeWorker implements the on/off/postMessage/terminate surface LazyWorkerThreadHost uses, and nothing here touches the rest of Worker. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: FakeWorker implements the transport lifecycle; absent threadId skips optional owner cleanup. return new UsageScanWorkerClient({ workerFactory: factory as never, log: () => {} }) } -const CODEX_BODY: UsageScanWorkerRequestBody = { +const SOURCE_CACHE: UsageSourceCacheRef = { + path: '/tmp/orca-codex-usage-sources.json', + schemaVersion: 6, + worktreeFingerprint: '[]', + reuse: true +} + +const CODEX_BODY: UsageScanWorkerScanBody = { + operation: 'scan', providerId: 'codex', worktrees: [], - previous: [] + sourceCache: SOURCE_CACHE +} + +const CODEX_VALUE = { + operation: 'scan', + providerId: 'codex', + sessions: [{ sessionId: 'session-1' }], + dailyAggregates: [] } describe('UsageScanWorkerClient', () => { @@ -70,21 +83,43 @@ describe('UsageScanWorkerClient', () => { const worker = new FakeWorker() const client = createClient(() => worker) - const pending = scanCodexUsageOnWorker((body) => client.scan(body), [], []) + const pending = scanCodexUsageOnWorker((body) => client.scan(body), [], SOURCE_CACHE) await vi.waitFor(() => expect(worker.postedRequests).toHaveLength(1)) - expect(worker.postedRequests[0]?.providerId).toBe('codex') + // The request names where the worker keeps the per-source cache instead of carrying it. + expect(worker.postedRequests[0]).toMatchObject({ + operation: 'scan', + providerId: 'codex', + sourceCache: SOURCE_CACHE + }) + worker.emit('message', { id: worker.lastId(), ok: true, value: CODEX_VALUE }) + + await expect(pending).resolves.toEqual({ + sessions: [{ sessionId: 'session-1' }], + dailyAggregates: [] + }) + }) + + it('routes a cache split to the worker and hands back the report text', async () => { + const worker = new FakeWorker() + const client = createClient(() => worker) + + const pending = client.splitCacheFile({ + cacheFile: '/tmp/usage.json', + sourceKey: 'processedFiles' + }) + await vi.waitFor(() => expect(worker.postedRequests).toHaveLength(1)) + expect(worker.postedRequests[0]).toMatchObject({ + operation: 'splitCacheFile', + cacheFile: '/tmp/usage.json', + sourceKey: 'processedFiles' + }) worker.emit('message', { id: worker.lastId(), ok: true, - value: { - providerId: 'codex', - source: [{ path: 'a.jsonl' }], - sessions: [], - dailyAggregates: [] - } + value: { operation: 'splitCacheFile', reportText: '{}', migrated: true } }) - await expect(pending).resolves.toMatchObject({ source: [{ path: 'a.jsonl' }] }) + await expect(pending).resolves.toEqual({ reportText: '{}', migrated: true }) }) it('fails closed instead of scanning on the calling thread when spawn fails', async () => { @@ -126,18 +161,9 @@ describe('UsageScanWorkerClient', () => { worker.emit('message', { id: worker.lastId(), filesScanned: window * 100 }) } await vi.advanceTimersByTimeAsync(USAGE_SCAN_NO_PROGRESS_TIMEOUT_MS - 1) - worker.emit('message', { - id: worker.lastId(), - ok: true, - value: { - providerId: 'codex', - source: [{ path: 'a.jsonl' }], - sessions: [], - dailyAggregates: [] - } - }) + worker.emit('message', { id: worker.lastId(), ok: true, value: CODEX_VALUE }) - await expect(pending).resolves.toMatchObject({ source: [{ path: 'a.jsonl' }] }) + await expect(pending).resolves.toMatchObject({ sessions: [{ sessionId: 'session-1' }] }) } finally { vi.useRealTimers() } @@ -201,12 +227,12 @@ describe('UsageScanWorkerClient', () => { const worker = new FakeWorker() const client = createClient(() => worker) - const pending = scanCodexUsageOnWorker((body) => client.scan(body), [], []) + const pending = scanCodexUsageOnWorker((body) => client.scan(body), [], SOURCE_CACHE) await vi.waitFor(() => expect(worker.postedRequests).toHaveLength(1)) worker.emit('message', { id: worker.lastId(), ok: true, - value: { providerId: 'claude', source: [], sessions: [], dailyAggregates: [] } + value: { operation: 'scan', providerId: 'claude', sessions: [], dailyAggregates: [] } }) await expect(pending).rejects.toThrow(/answered for claude/) diff --git a/src/main/usage/usage-scan-worker-client.ts b/src/main/usage/usage-scan-worker-client.ts index 785b82157fd..24a7ab5feb9 100644 --- a/src/main/usage/usage-scan-worker-client.ts +++ b/src/main/usage/usage-scan-worker-client.ts @@ -1,34 +1,25 @@ import { WorkerThreadRequestQueue } from '../worker-thread-request-queue' import type { WorkerThreadFactory } from '../lazy-worker-thread-host' -import type { - ClaudeUsageDailyAggregate, - ClaudeUsagePersistedFile, - ClaudeUsageSession -} from '../claude-usage/types' -import type { - CodexUsageDailyAggregate, - CodexUsagePersistedFile, - CodexUsageSession -} from '../codex-usage/types' -import type { - OpenCodeUsageDailyAggregate, - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession -} from '../opencode-usage/types' -import type { - MuseUsageDailyAggregate, - MuseUsagePersistedFile, - MuseUsageSession -} from '../muse-usage/types' +import type { ClaudeUsageDailyAggregate, ClaudeUsageSession } from '../claude-usage/types' +import type { CodexUsageDailyAggregate, CodexUsageSession } from '../codex-usage/types' +import type { OpenCodeUsageDailyAggregate, OpenCodeUsageSession } from '../opencode-usage/types' +import type { MuseUsageDailyAggregate, MuseUsageSession } from '../muse-usage/types' import type { UsageScanWorktreeRef } from './usage-provider-contract' import type { UsageScanWorkerProviderId, UsageScanWorkerRequest, UsageScanWorkerRequestBody, UsageScanWorkerResponse, + UsageScanWorkerScanBody, UsageScanWorkerValue } from './usage-scan-worker-protocol' +import type { + UsageCacheSplitRequest, + UsageCacheSplitResult, + UsageSourceCacheRef +} from './usage-source-cache-file' import { isUsageScanWorkerProgress } from './usage-scan-worker-protocol' +import { createUsageScanWorkerTransport } from './usage-scan-worker-cache-cleanup' // Why (#20940): this module owns the request half of the shared usage scan // worker — FIFO one-at-a-time dispatch, a no-progress deadline, respawn-on-fault @@ -54,8 +45,7 @@ export const MAX_CONSECUTIVE_DEATHS = 3 /** Thrown when no worker could be started at all, as distinct from a fault. */ export class UsageScanWorkerUnavailableError extends Error {} -type ProviderScanResult = { - source: TSource[] +type ProviderScanResult = { sessions: TSession[] dailyAggregates: TDaily[] } @@ -72,7 +62,7 @@ export class UsageScanWorkerClient { constructor(options: { workerFactory: WorkerThreadFactory; log?: (message: string) => void }) { const log = options.log ?? ((message: string) => console.warn(message)) this.queue = new WorkerThreadRequestQueue({ - factory: options.workerFactory, + factory: () => createUsageScanWorkerTransport(options.workerFactory), idleTeardownMs: IDLE_TEARDOWN_MS, maxConsecutiveDeaths: MAX_CONSECUTIVE_DEATHS, createUnavailableError: (message) => new UsageScanWorkerUnavailableError(message), @@ -90,10 +80,23 @@ export class UsageScanWorkerClient { /** * Run one provider's scan on the worker. - * @param body - Provider id, worktree refs, and that provider's previous cache. + * @param body - Provider id, worktree refs, and where that provider's per-source cache lives. * @returns The worker's value for that provider. */ - async scan(body: UsageScanWorkerRequestBody): Promise { + scan(body: UsageScanWorkerScanBody): Promise { + return this.dispatch(body) + } + + /** Split a cache that still carries its per-source records, so main parses only the report. */ + async splitCacheFile(request: UsageCacheSplitRequest): Promise { + const value = await this.dispatch({ operation: 'splitCacheFile', ...request }) + if (value.operation !== 'splitCacheFile') { + throw new Error(`Usage scan worker answered ${value.operation}, expected splitCacheFile`) + } + return { reportText: value.reportText, migrated: value.migrated } + } + + private async dispatch(body: UsageScanWorkerRequestBody): Promise { const response = await this.queue.dispatch( (id) => ({ ...body, id }), USAGE_SCAN_NO_PROGRESS_TIMEOUT_MS @@ -110,7 +113,8 @@ function message(error: unknown): string { } /** A response for the wrong provider means the worker and client disagree on the protocol. */ -function wrongProvider(expected: UsageScanWorkerProviderId, actual: string): Error { +function wrongProvider(expected: UsageScanWorkerProviderId, value: UsageScanWorkerValue): Error { + const actual = value.operation === 'scan' ? value.providerId : value.operation return new Error(`Usage scan worker answered for ${actual}, expected ${expected}`) } @@ -118,84 +122,74 @@ function wrongProvider(expected: UsageScanWorkerProviderId, actual: string): Err * Scan Claude usage transcripts on the shared worker. * @param scan - Dispatch function, injected so tests need no real thread. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns Processed files plus the session and daily projections. + * @param sourceCache - Where the worker reads and writes that provider's per-source cache. + * @returns The session and daily projections. */ export async function scanClaudeUsageOnWorker( - scan: (body: UsageScanWorkerRequestBody) => Promise, + scan: (body: UsageScanWorkerScanBody) => Promise, worktrees: UsageScanWorktreeRef[], - previous: ClaudeUsagePersistedFile[] -): Promise< - ProviderScanResult -> { - const value = await scan({ providerId: 'claude', worktrees, previous }) - if (value.providerId !== 'claude') { - throw wrongProvider('claude', value.providerId) + sourceCache: UsageSourceCacheRef +): Promise> { + const value = await scan({ operation: 'scan', providerId: 'claude', worktrees, sourceCache }) + if (value.operation !== 'scan' || value.providerId !== 'claude') { + throw wrongProvider('claude', value) } - return value + return { sessions: value.sessions, dailyAggregates: value.dailyAggregates } } /** * Scan Codex rollouts on the shared worker. * @param scan - Dispatch function, injected so tests need no real thread. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns Processed files plus the session and daily projections. + * @param sourceCache - Where the worker reads and writes that provider's per-source cache. + * @returns The session and daily projections. */ export async function scanCodexUsageOnWorker( - scan: (body: UsageScanWorkerRequestBody) => Promise, + scan: (body: UsageScanWorkerScanBody) => Promise, worktrees: UsageScanWorktreeRef[], - previous: CodexUsagePersistedFile[] -): Promise< - ProviderScanResult -> { - const value = await scan({ providerId: 'codex', worktrees, previous }) - if (value.providerId !== 'codex') { - throw wrongProvider('codex', value.providerId) + sourceCache: UsageSourceCacheRef +): Promise> { + const value = await scan({ operation: 'scan', providerId: 'codex', worktrees, sourceCache }) + if (value.operation !== 'scan' || value.providerId !== 'codex') { + throw wrongProvider('codex', value) } - return value + return { sessions: value.sessions, dailyAggregates: value.dailyAggregates } } /** * Scan OpenCode usage databases on the shared worker. * @param scan - Dispatch function, injected so tests need no real thread. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-database cache. - * @returns Processed databases plus the session and daily projections. + * @param sourceCache - Where the worker reads and writes that provider's per-source cache. + * @returns The session and daily projections. */ export async function scanOpenCodeUsageOnWorker( - scan: (body: UsageScanWorkerRequestBody) => Promise, + scan: (body: UsageScanWorkerScanBody) => Promise, worktrees: UsageScanWorktreeRef[], - previous: OpenCodeUsagePersistedDatabase[] -): Promise< - ProviderScanResult< - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession, - OpenCodeUsageDailyAggregate - > -> { - const value = await scan({ providerId: 'opencode', worktrees, previous }) - if (value.providerId !== 'opencode') { - throw wrongProvider('opencode', value.providerId) + sourceCache: UsageSourceCacheRef +): Promise> { + const value = await scan({ operation: 'scan', providerId: 'opencode', worktrees, sourceCache }) + if (value.operation !== 'scan' || value.providerId !== 'opencode') { + throw wrongProvider('opencode', value) } - return value + return { sessions: value.sessions, dailyAggregates: value.dailyAggregates } } /** * Scan Muse Code session logs on the shared worker. * @param scan - Dispatch function, injected so tests need no real thread. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns Processed files plus the session and daily projections. + * @param sourceCache - Where the worker reads and writes that provider's per-source cache. + * @returns The session and daily projections. */ export async function scanMuseUsageOnWorker( - scan: (body: UsageScanWorkerRequestBody) => Promise, + scan: (body: UsageScanWorkerScanBody) => Promise, worktrees: UsageScanWorktreeRef[], - previous: MuseUsagePersistedFile[] -): Promise> { - const value = await scan({ providerId: 'muse', worktrees, previous }) - if (value.providerId !== 'muse') { - throw wrongProvider('muse', value.providerId) + sourceCache: UsageSourceCacheRef +): Promise> { + const value = await scan({ operation: 'scan', providerId: 'muse', worktrees, sourceCache }) + if (value.operation !== 'scan' || value.providerId !== 'muse') { + throw wrongProvider('muse', value) } - return value + return { sessions: value.sessions, dailyAggregates: value.dailyAggregates } } diff --git a/src/main/usage/usage-scan-worker-entry.ts b/src/main/usage/usage-scan-worker-entry.ts index 9c4c4b1880d..eefa7a2a468 100644 --- a/src/main/usage/usage-scan-worker-entry.ts +++ b/src/main/usage/usage-scan-worker-entry.ts @@ -3,12 +3,25 @@ import { scanClaudeUsageFiles } from '../claude-usage/scanner' import { scanCodexUsageFiles } from '../codex-usage/scanner' import { scanOpenCodeUsageDatabases } from '../opencode-usage/scanner' import { scanMuseUsageFiles } from '../muse-usage/scanner' +import { normalizeMuseUsagePersistedFiles } from '../muse-usage/persisted-file-normalization' +import { normalizeOpenCodeUsagePersistedDatabases } from '../opencode-usage/persisted-database-normalization' +import type { ClaudeUsagePersistedFile } from '../claude-usage/types' +import type { CodexUsagePersistedFile } from '../codex-usage/types' +import type { OpenCodeUsagePersistedDatabase } from '../opencode-usage/types' +import type { MuseUsagePersistedFile } from '../muse-usage/types' import type { UsageScanWorkerProgress, UsageScanWorkerRequest, UsageScanWorkerResponse, + UsageScanWorkerScanBody, UsageScanWorkerValue } from './usage-scan-worker-protocol' +import { + readUsageSourceCache, + splitUsageCacheFile, + writeUsageSourceCache, + type UsageSourceCacheRef +} from './usage-source-cache-file' // Why (#20940): the Claude/Codex/OpenCode/Muse usage scans parse whole history // corpora and read SQLite synchronously. Running them on this worker thread @@ -55,32 +68,48 @@ function createProgressReporter(id: number): (count: number) => void { } } -async function runScan( +async function runRequest( request: UsageScanWorkerRequest, onFilesScanned: (count: number) => void ): Promise { - // Switched, not table-driven: each branch narrows `previous` to that + if (request.operation === 'splitCacheFile') { + return { operation: 'splitCacheFile', ...(await splitUsageCacheFile(request)) } + } + return runScan(request, onFilesScanned) +} + +async function runScan( + request: UsageScanWorkerScanBody, + onFilesScanned: (count: number) => void +): Promise { + // Switched, not table-driven: each branch narrows the cache to that // provider's own record type, so nothing here needs a type assertion. switch (request.providerId) { case 'claude': { const result = await scanClaudeUsageFiles( request.worktrees, - request.previous, + await readUsageSourceCache(request.sourceCache), onFilesScanned, request.profileDirs ) + await persistSourceCache(request.sourceCache, result.processedFiles) return { + operation: 'scan', providerId: 'claude', - source: result.processedFiles, sessions: result.sessions, dailyAggregates: result.dailyAggregates } } case 'codex': { - const result = await scanCodexUsageFiles(request.worktrees, request.previous, onFilesScanned) + const result = await scanCodexUsageFiles( + request.worktrees, + await readUsageSourceCache(request.sourceCache), + onFilesScanned + ) + await persistSourceCache(request.sourceCache, result.processedFiles) return { + operation: 'scan', providerId: 'codex', - source: result.processedFiles, sessions: result.sessions, dailyAggregates: result.dailyAggregates } @@ -88,21 +117,31 @@ async function runScan( case 'opencode': { const result = await scanOpenCodeUsageDatabases( request.worktrees, - request.previous, + normalizeOpenCodeUsagePersistedDatabases( + await readUsageSourceCache(request.sourceCache) + ), onFilesScanned ) + await persistSourceCache(request.sourceCache, result.processedDatabases) return { + operation: 'scan', providerId: 'opencode', - source: result.processedDatabases, sessions: result.sessions, dailyAggregates: result.dailyAggregates } } case 'muse': { - const result = await scanMuseUsageFiles(request.worktrees, request.previous, onFilesScanned) + const result = await scanMuseUsageFiles( + request.worktrees, + normalizeMuseUsagePersistedFiles( + await readUsageSourceCache(request.sourceCache) + ), + onFilesScanned + ) + await persistSourceCache(request.sourceCache, result.processedFiles) return { + operation: 'scan', providerId: 'muse', - source: result.processedFiles, sessions: result.sessions, dailyAggregates: result.dailyAggregates } @@ -110,12 +149,25 @@ async function runScan( } } +// Why: like the main-thread report, a cache that cannot be written must not +// turn a successful scan into a failed one; the next scan just starts colder. +async function persistSourceCache( + sourceCache: UsageSourceCacheRef, + sources: readonly unknown[] +): Promise { + try { + await writeUsageSourceCache(sourceCache, sources) + } catch (error) { + console.warn('[usage-scan] Could not persist the per-source usage cache:', error) + } +} + async function handleRequest(request: UsageScanWorkerRequest): Promise { try { return { id: request.id, ok: true, - value: await runScan(request, createProgressReporter(request.id)) + value: await runRequest(request, createProgressReporter(request.id)) } } catch (err) { return { id: request.id, ok: false, error: err instanceof Error ? err.message : String(err) } diff --git a/src/main/usage/usage-scan-worker-event-loop.test.ts b/src/main/usage/usage-scan-worker-event-loop.test.ts index 3f39a95ebbf..800bdcf2c1f 100644 --- a/src/main/usage/usage-scan-worker-event-loop.test.ts +++ b/src/main/usage/usage-scan-worker-event-loop.test.ts @@ -8,6 +8,7 @@ import { Worker } from 'node:worker_threads' import { scanCodexUsageFiles } from '../codex-usage/scanner' import { UsageScanWorkerClient, scanCodexUsageOnWorker } from './usage-scan-worker-client' import type { UsageScanWorktreeRef } from './usage-provider-contract' +import { readUsageSourceCache, type UsageSourceCacheRef } from './usage-source-cache-file' // Why this test exists: "the scan no longer blocks the main process" is not a // stopwatch claim. It measures the *calling* thread's event-loop active time — @@ -196,16 +197,25 @@ describe('usage scan worker event-loop occupancy', () => { expect(caller.value.dailyAggregates[0]?.eventCount).toBe(EXPECTED_EVENTS) const client = createWorkerClient() + const sourceCache: UsageSourceCacheRef = { + path: join(corpusRoot, 'orca-codex-usage-sources.json'), + schemaVersion: 1, + worktreeFingerprint: '[]', + reuse: true + } const worker = await measureCallerOccupancy(() => - withCorpusEnv(() => scanCodexUsageOnWorker((body) => client.scan(body), WORKTREES, [])) + withCorpusEnv(() => + scanCodexUsageOnWorker((body) => client.scan(body), WORKTREES, sourceCache) + ) ) - expect(worker.value.source).toHaveLength(FILE_COUNT) + // The per-source records stay on the worker's side of the boundary. + expect(await readUsageSourceCache(sourceCache)).toHaveLength(FILE_COUNT) expect(worker.value.sessions).toHaveLength(FILE_COUNT) expect(worker.value.dailyAggregates).toHaveLength(1) expect(worker.value.dailyAggregates[0]?.eventCount).toBe(EXPECTED_EVENTS) - // The caller still pays to post the request and structured-clone a - // 600-file result back, so this is a fifth, not a rout. Measured margin is + // The caller still pays to post the request and structured-clone 600 + // sessions back, so this is a fifth, not a rout. Measured margin is // ~50x idle and ~90x under CPU contention. expect( worker.occupancy.activeMs, diff --git a/src/main/usage/usage-scan-worker-protocol.ts b/src/main/usage/usage-scan-worker-protocol.ts index 0d657cd535b..3c49410b345 100644 --- a/src/main/usage/usage-scan-worker-protocol.ts +++ b/src/main/usage/usage-scan-worker-protocol.ts @@ -1,24 +1,13 @@ -import type { - ClaudeUsageDailyAggregate, - ClaudeUsagePersistedFile, - ClaudeUsageSession -} from '../claude-usage/types' -import type { - CodexUsageDailyAggregate, - CodexUsagePersistedFile, - CodexUsageSession -} from '../codex-usage/types' -import type { - OpenCodeUsageDailyAggregate, - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession -} from '../opencode-usage/types' -import type { - MuseUsageDailyAggregate, - MuseUsagePersistedFile, - MuseUsageSession -} from '../muse-usage/types' +import type { ClaudeUsageDailyAggregate, ClaudeUsageSession } from '../claude-usage/types' +import type { CodexUsageDailyAggregate, CodexUsageSession } from '../codex-usage/types' +import type { OpenCodeUsageDailyAggregate, OpenCodeUsageSession } from '../opencode-usage/types' +import type { MuseUsageDailyAggregate, MuseUsageSession } from '../muse-usage/types' import type { UsageScanWorktreeRef } from './usage-provider-contract' +import type { + UsageCacheSplitRequest, + UsageCacheSplitResult, + UsageSourceCacheRef +} from './usage-source-cache-file' // Why (#20940): the first-party usage scans walk whole rollout/transcript // corpora and read SQLite synchronously, all on the Electron main process. They @@ -33,55 +22,74 @@ import type { UsageScanWorktreeRef } from './usage-provider-contract' */ export type UsageScanWorkerProviderId = 'claude' | 'codex' | 'opencode' | 'muse' -/** Request body per provider; `previous` is that provider's own per-source cache. */ -export type UsageScanWorkerRequestBody = +/** + * Scan body per provider. The worker reads that provider's per-source cache from + * `sourceCache` and writes the new one back, so those records never cross to main. + */ +export type UsageScanWorkerScanBody = | { + operation: 'scan' providerId: 'claude' profileDirs?: string[] worktrees: UsageScanWorktreeRef[] - previous: ClaudeUsagePersistedFile[] + sourceCache: UsageSourceCacheRef } - | { providerId: 'codex'; worktrees: UsageScanWorktreeRef[]; previous: CodexUsagePersistedFile[] } | { + operation: 'scan' + providerId: 'codex' + worktrees: UsageScanWorktreeRef[] + sourceCache: UsageSourceCacheRef + } + | { + operation: 'scan' providerId: 'opencode' worktrees: UsageScanWorktreeRef[] - previous: OpenCodeUsagePersistedDatabase[] + sourceCache: UsageSourceCacheRef } - | { providerId: 'muse'; worktrees: UsageScanWorktreeRef[]; previous: MuseUsagePersistedFile[] } + | { + operation: 'scan' + providerId: 'muse' + worktrees: UsageScanWorktreeRef[] + sourceCache: UsageSourceCacheRef + } + +export type UsageScanWorkerRequestBody = + | UsageScanWorkerScanBody + | ({ operation: 'splitCacheFile' } & UsageCacheSplitRequest) export type UsageScanWorkerRequest = UsageScanWorkerRequestBody & { id: number } -/** - * Scan result per provider. `source` is the provider's per-source cache under a - * uniform name: the persisted key (`processedFiles` / `processedDatabases`) is a - * disk-format concern and each route renames it back on the main thread. - */ -export type UsageScanWorkerValue = +/** Scan result per provider: only the projections main reports from. */ +export type UsageScanWorkerScanValue = | { + operation: 'scan' providerId: 'claude' - source: ClaudeUsagePersistedFile[] sessions: ClaudeUsageSession[] dailyAggregates: ClaudeUsageDailyAggregate[] } | { + operation: 'scan' providerId: 'codex' - source: CodexUsagePersistedFile[] sessions: CodexUsageSession[] dailyAggregates: CodexUsageDailyAggregate[] } | { + operation: 'scan' providerId: 'opencode' - source: OpenCodeUsagePersistedDatabase[] sessions: OpenCodeUsageSession[] dailyAggregates: OpenCodeUsageDailyAggregate[] } | { + operation: 'scan' providerId: 'muse' - source: MuseUsagePersistedFile[] sessions: MuseUsageSession[] dailyAggregates: MuseUsageDailyAggregate[] } +export type UsageScanWorkerValue = + | UsageScanWorkerScanValue + | ({ operation: 'splitCacheFile' } & UsageCacheSplitResult) + export type UsageScanWorkerResponse = | { id: number; ok: true; value: UsageScanWorkerValue } | { id: number; ok: false; error: string } diff --git a/src/main/usage/usage-scan-worker-spawn.ts b/src/main/usage/usage-scan-worker-spawn.ts index b38b3cc44a9..998b0ca98d2 100644 --- a/src/main/usage/usage-scan-worker-spawn.ts +++ b/src/main/usage/usage-scan-worker-spawn.ts @@ -2,27 +2,16 @@ import { claudeProfileTranscriptDirs } from '../claude-usage/transcript-file-dis import { existsSync } from 'node:fs' import { Worker } from 'node:worker_threads' import { currentWorkerEntryLayout, resolveWorkerThreadEntryPath } from '../worker-thread-entry-path' -import type { - ClaudeUsageDailyAggregate, - ClaudeUsagePersistedFile, - ClaudeUsageSession -} from '../claude-usage/types' -import type { - CodexUsageDailyAggregate, - CodexUsagePersistedFile, - CodexUsageSession -} from '../codex-usage/types' -import type { - OpenCodeUsageDailyAggregate, - OpenCodeUsagePersistedDatabase, - OpenCodeUsageSession -} from '../opencode-usage/types' -import type { - MuseUsageDailyAggregate, - MuseUsagePersistedFile, - MuseUsageSession -} from '../muse-usage/types' +import type { ClaudeUsageDailyAggregate, ClaudeUsageSession } from '../claude-usage/types' +import type { CodexUsageDailyAggregate, CodexUsageSession } from '../codex-usage/types' +import type { OpenCodeUsageDailyAggregate, OpenCodeUsageSession } from '../opencode-usage/types' +import type { MuseUsageDailyAggregate, MuseUsageSession } from '../muse-usage/types' import type { UsageScanWorktreeRef } from './usage-provider-contract' +import type { + UsageCacheSplitRequest, + UsageCacheSplitResult, + UsageSourceCacheRef +} from './usage-source-cache-file' import { scanClaudeUsageOnWorker, scanCodexUsageOnWorker, @@ -60,18 +49,17 @@ function getSharedClient(): UsageScanWorkerClient { /** * Scan Claude usage transcripts through the shared worker client. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns The same projection `scanClaudeUsageFiles` returns, computed off the main thread. + * @param sourceCache - Where the worker keeps that provider's per-source cache. + * @returns The session and daily projections, computed off the main thread. */ -export async function scanClaudeUsageFilesViaWorker( +export function scanClaudeUsageFilesViaWorker( worktrees: UsageScanWorktreeRef[], - previous: ClaudeUsagePersistedFile[] = [] + sourceCache: UsageSourceCacheRef ): Promise<{ - processedFiles: ClaudeUsagePersistedFile[] sessions: ClaudeUsageSession[] dailyAggregates: ClaudeUsageDailyAggregate[] }> { - const value = await scanClaudeUsageOnWorker( + return scanClaudeUsageOnWorker( (body) => getSharedClient().scan( body.providerId === 'claude' @@ -79,89 +67,65 @@ export async function scanClaudeUsageFilesViaWorker( : body ), worktrees, - previous + sourceCache ) - return { - processedFiles: value.source, - sessions: value.sessions, - dailyAggregates: value.dailyAggregates - } } /** * Scan Codex rollouts through the shared worker client. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns The same projection `scanCodexUsageFiles` returns, computed off the main thread. + * @param sourceCache - Where the worker keeps that provider's per-source cache. + * @returns The session and daily projections, computed off the main thread. */ -export async function scanCodexUsageFilesViaWorker( +export function scanCodexUsageFilesViaWorker( worktrees: UsageScanWorktreeRef[], - previous: CodexUsagePersistedFile[] = [] + sourceCache: UsageSourceCacheRef ): Promise<{ - processedFiles: CodexUsagePersistedFile[] sessions: CodexUsageSession[] dailyAggregates: CodexUsageDailyAggregate[] }> { - const value = await scanCodexUsageOnWorker( - (body) => getSharedClient().scan(body), - worktrees, - previous - ) - return { - processedFiles: value.source, - sessions: value.sessions, - dailyAggregates: value.dailyAggregates - } + return scanCodexUsageOnWorker((body) => getSharedClient().scan(body), worktrees, sourceCache) } /** * Scan OpenCode usage databases through the shared worker client. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-database cache. - * @returns The same projection `scanOpenCodeUsageDatabases` returns, computed off the main thread. + * @param sourceCache - Where the worker keeps that provider's per-source cache. + * @returns The session and daily projections, computed off the main thread. */ -export async function scanOpenCodeUsageDatabasesViaWorker( +export function scanOpenCodeUsageDatabasesViaWorker( worktrees: UsageScanWorktreeRef[], - previous: OpenCodeUsagePersistedDatabase[] = [] + sourceCache: UsageSourceCacheRef ): Promise<{ - processedDatabases: OpenCodeUsagePersistedDatabase[] sessions: OpenCodeUsageSession[] dailyAggregates: OpenCodeUsageDailyAggregate[] }> { - const value = await scanOpenCodeUsageOnWorker( - (body) => getSharedClient().scan(body), - worktrees, - previous - ) - return { - processedDatabases: value.source, - sessions: value.sessions, - dailyAggregates: value.dailyAggregates - } + return scanOpenCodeUsageOnWorker((body) => getSharedClient().scan(body), worktrees, sourceCache) } /** * Scan Muse Code session logs through the shared worker client. * @param worktrees - Worktree refs used to attribute usage. - * @param previous - Last scan's per-file cache. - * @returns The same projection `scanMuseUsageFiles` returns, computed off the main thread. + * @param sourceCache - Where the worker keeps that provider's per-source cache. + * @returns The session and daily projections, computed off the main thread. */ -export async function scanMuseUsageFilesViaWorker( +export function scanMuseUsageFilesViaWorker( worktrees: UsageScanWorktreeRef[], - previous: MuseUsagePersistedFile[] = [] + sourceCache: UsageSourceCacheRef ): Promise<{ - processedFiles: MuseUsagePersistedFile[] sessions: MuseUsageSession[] dailyAggregates: MuseUsageDailyAggregate[] }> { - const value = await scanMuseUsageOnWorker( - (body) => getSharedClient().scan(body), - worktrees, - previous - ) - return { - processedFiles: value.source, - sessions: value.sessions, - dailyAggregates: value.dailyAggregates - } + return scanMuseUsageOnWorker((body) => getSharedClient().scan(body), worktrees, sourceCache) +} + +/** + * Split a usage cache that still carries its per-source records, through the shared worker client. + * @param request - The cache file and the key its per-source records sit under. + * @returns The report alone, as JSON text, and whether a split happened. + */ +export function splitUsageCacheFileViaWorker( + request: UsageCacheSplitRequest +): Promise { + return getSharedClient().splitCacheFile(request) } diff --git a/src/main/usage/usage-source-cache-file.test.ts b/src/main/usage/usage-source-cache-file.test.ts new file mode 100644 index 00000000000..ab4acc5484e --- /dev/null +++ b/src/main/usage/usage-source-cache-file.test.ts @@ -0,0 +1,113 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + readUsageSourceCache, + splitUsageCacheFile, + usageSourceCachePath, + writeUsageSourceCache, + type UsageSourceCacheRef +} from './usage-source-cache-file' + +describe('usage source cache file', () => { + let directory: string + let ref: UsageSourceCacheRef + + beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), 'orca-usage-source-cache-')) + ref = { + path: join(directory, 'usage-sources.json'), + schemaVersion: 6, + worktreeFingerprint: '[]', + reuse: true + } + }) + + afterEach(() => { + rmSync(directory, { recursive: true, force: true }) + }) + + it('names the sidecar after the report it belongs to', () => { + expect(usageSourceCachePath(join(directory, 'orca-claude-usage.json'))).toBe( + join(directory, 'orca-claude-usage-sources.json') + ) + }) + + it('round-trips the records a scan wrote', async () => { + await writeUsageSourceCache(ref, [{ path: 'a.jsonl' }]) + + await expect(readUsageSourceCache(ref)).resolves.toEqual([{ path: 'a.jsonl' }]) + }) + + it('rejects attribution from a sidecar committed ahead of its report', async () => { + await writeUsageSourceCache({ ...ref, worktreeFingerprint: 'new-worktrees' }, [ + { path: 'a.jsonl', repoId: 'new-repo' } + ]) + + await expect( + readUsageSourceCache({ ...ref, worktreeFingerprint: 'old-worktrees' }) + ).resolves.toEqual([]) + await expect( + readUsageSourceCache({ ...ref, worktreeFingerprint: 'new-worktrees' }) + ).resolves.toEqual([{ path: 'a.jsonl', repoId: 'new-repo' }]) + }) + + it('starts cold when told not to reuse, on another schema, or on a torn file', async () => { + await writeUsageSourceCache(ref, [{ path: 'a.jsonl' }]) + + await expect(readUsageSourceCache({ ...ref, reuse: false })).resolves.toEqual([]) + await expect(readUsageSourceCache({ ...ref, schemaVersion: 7 })).resolves.toEqual([]) + writeFileSync(ref.path, '{"schemaVersion":6,"sources":[') + await expect(readUsageSourceCache(ref)).resolves.toEqual([]) + await expect( + readUsageSourceCache({ ...ref, path: join(directory, 'none.json') }) + ).resolves.toEqual([]) + }) + + it('moves inline records of a legacy cache into the sidecar and returns only the report', async () => { + const cacheFile = join(directory, 'orca-claude-usage.json') + const legacy = { + schemaVersion: 6, + worktreeFingerprint: '[]', + processedFiles: [{ path: 'a.jsonl' }], + sessions: [{ sessionId: 's' }] + } + writeFileSync(cacheFile, JSON.stringify(legacy, null, 2)) + + const result = await splitUsageCacheFile({ cacheFile, sourceKey: 'processedFiles' }) + + expect(result.migrated).toBe(true) + expect(JSON.parse(result.reportText ?? '')).toEqual({ + schemaVersion: 6, + worktreeFingerprint: '[]', + sessions: [{ sessionId: 's' }] + }) + expect(JSON.parse(readFileSync(usageSourceCachePath(cacheFile), 'utf-8'))).toEqual({ + schemaVersion: 6, + worktreeFingerprint: '[]', + sources: [{ path: 'a.jsonl' }] + }) + }) + + it('returns a report that is already split unchanged and leaves the sidecar alone', async () => { + const cacheFile = join(directory, 'orca-claude-usage.json') + const text = JSON.stringify({ schemaVersion: 6, sessions: [] }) + writeFileSync(cacheFile, text) + + await expect(splitUsageCacheFile({ cacheFile, sourceKey: 'processedFiles' })).resolves.toEqual({ + reportText: text, + migrated: false + }) + expect(existsSync(usageSourceCachePath(cacheFile))).toBe(false) + }) + + it('reports a missing cache as no report rather than an error', async () => { + await expect( + splitUsageCacheFile({ + cacheFile: join(directory, 'missing.json'), + sourceKey: 'processedFiles' + }) + ).resolves.toEqual({ reportText: null, migrated: false }) + }) +}) diff --git a/src/main/usage/usage-source-cache-file.ts b/src/main/usage/usage-source-cache-file.ts new file mode 100644 index 00000000000..0e0cffb2f78 --- /dev/null +++ b/src/main/usage/usage-source-cache-file.ts @@ -0,0 +1,96 @@ +import { mkdir, readFile } from 'node:fs/promises' +import { dirname, join, parse } from 'node:path' +import { threadId } from 'node:worker_threads' +import { + durableWriteTempPath, + removeStaleDurableWriteTempFiles, + writeFileDurable +} from '../durable-file-write' + +// Why: per-source records are most of a usage cache (43 of 56 MB on one real Claude history) and +// only a scan reads them, so they live in this worker-owned sidecar and never touch main. + +/** Where one provider's per-source scan cache lives, and whether this scan may start from it. */ +export type UsageSourceCacheRef = { + path: string + schemaVersion: number + worktreeFingerprint: string | null + reuse: boolean +} + +export type UsageCacheSplitRequest = { cacheFile: string; sourceKey: string } +export type UsageCacheSplitResult = { reportText: string | null; migrated: boolean } + +export function usageSourceCachePath(cacheFile: string): string { + const { dir, name } = parse(cacheFile) + return join(dir, `${name}-sources.json`) +} + +/** The last scan's records; none when unreadable or from another schema, so the scan starts cold. */ +export async function readUsageSourceCache(ref: UsageSourceCacheRef): Promise { + if (!ref.reuse) { + return [] + } + try { + const parsed = JSON.parse(await readFile(ref.path, 'utf-8')) + // The sidecar can commit before its report, so validate attribution independently. + return parsed?.schemaVersion === ref.schemaVersion && + parsed.worktreeFingerprint === ref.worktreeFingerprint && + Array.isArray(parsed.sources) + ? parsed.sources + : [] + } catch { + return [] + } +} + +export async function writeUsageSourceCache( + ref: UsageSourceCacheRef, + sources: readonly unknown[] +): Promise { + await mkdir(dirname(ref.path), { recursive: true }).catch(() => {}) + // Why: a worker terminated mid-write orphans a multi-MB temp file; reclaim earlier launches' ones. + await removeStaleDurableWriteTempFiles(ref.path) + await writeFileDurable( + durableWriteTempPath(ref.path, threadId > 0 ? String(threadId) : undefined), + ref.path, + JSON.stringify({ + schemaVersion: ref.schemaVersion, + worktreeFingerprint: ref.worktreeFingerprint, + sources + }) + ) +} + +/** + * Caches written before the sidecar existed carry the per-source records inline. Move them into + * the sidecar and hand back only the report, so main parses a few MB instead of the whole file. + */ +export async function splitUsageCacheFile( + request: UsageCacheSplitRequest +): Promise { + let text: string + try { + text = await readFile(request.cacheFile, 'utf-8') + } catch (error) { + if (error instanceof Error && 'code' in error && error.code === 'ENOENT') { + return { reportText: null, migrated: false } + } + throw error + } + const parsed = JSON.parse(text) + if (!parsed || typeof parsed !== 'object' || !Array.isArray(parsed[request.sourceKey])) { + return { reportText: text, migrated: false } + } + const { [request.sourceKey]: sources, ...report } = parsed + await writeUsageSourceCache( + { + path: usageSourceCachePath(request.cacheFile), + schemaVersion: report.schemaVersion, + worktreeFingerprint: report.worktreeFingerprint ?? null, + reuse: true + }, + sources + ) + return { reportText: JSON.stringify(report), migrated: true } +}