mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
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 (`<cache>-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 <noreply@anthropic.com> * fix(usage): preserve cache attribution and legacy warm scans * fix(usage): reclaim cache temps after worker retirement --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
co-authored by
Claude Opus 5.5
Neil
parent
618dc62889
commit
09aff1b94d
@@ -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: []
|
||||
}
|
||||
|
||||
@@ -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<AutomationRunUsage> {
|
||||
await this.whenLoaded()
|
||||
return resolveAutomationRunUsage(input, {
|
||||
getState: () => this.state,
|
||||
refresh: (force) => this.refresh(force),
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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<CodexUsageSession, CodexUsageDailyAggregate>
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -9,7 +9,6 @@ import type { CodexUsagePersistedState } from './types'
|
||||
|
||||
export function createEmptyScanResult() {
|
||||
return {
|
||||
processedFiles: [],
|
||||
sessions: [],
|
||||
dailyAggregates: []
|
||||
}
|
||||
|
||||
@@ -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<AutomationRunUsage> {
|
||||
await this.whenLoaded()
|
||||
return resolveCodexAutomationRunUsage(input, {
|
||||
getState: () => this.state,
|
||||
refresh: (force) => this.refresh(force),
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<void> {
|
||||
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) {
|
||||
|
||||
@@ -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<void>((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<unknown>) =>
|
||||
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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -14,6 +14,7 @@ type UsageProviderStores = {
|
||||
type UsageProviderChannelPrefix = keyof UsageProviderStores
|
||||
|
||||
type UsageProviderHandlerStore<Scope, Range, BreakdownKind> = {
|
||||
whenLoaded: () => Promise<void>
|
||||
getScanState: () => unknown
|
||||
setEnabled: (enabled: boolean) => unknown
|
||||
refresh: (force?: boolean) => unknown
|
||||
@@ -33,7 +34,11 @@ function registerProviderHandlers<Scope, Range, BreakdownKind>(
|
||||
prefix: UsageProviderChannelPrefix,
|
||||
usage: UsageProviderHandlerStore<Scope, Range, BreakdownKind>
|
||||
): 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<Scope, Range, BreakdownKind>(
|
||||
)
|
||||
ipcMain.handle(
|
||||
`${prefix}:getSnapshot`,
|
||||
(_event, args: UsageRangeArgs<Scope, Range> & { limit?: number }) =>
|
||||
usage.getSnapshot(args.scope, args.range, args.limit)
|
||||
async (_event, args: UsageRangeArgs<Scope, Range> & { limit?: number }) => {
|
||||
await usage.whenLoaded()
|
||||
return usage.getSnapshot(args.scope, args.range, args.limit)
|
||||
}
|
||||
)
|
||||
ipcMain.handle(`${prefix}:getSummary`, (_event, args: UsageRangeArgs<Scope, Range>) =>
|
||||
usage.getSummary(args.scope, args.range)
|
||||
|
||||
@@ -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<MuseUsageSession, MuseUsageDailyAggregate>
|
||||
|
||||
@@ -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
|
||||
}))
|
||||
}
|
||||
@@ -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 ?? []
|
||||
}
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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<OpenCodeUsageSession, OpenCodeUsageDailyAggregate>
|
||||
|
||||
@@ -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
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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<TSourceKey extends string, TSource, TSession, TDaily> = Record<
|
||||
TSourceKey,
|
||||
readonly TSource[]
|
||||
> & {
|
||||
/** What a scan reports. The per-source scan cache stays wherever the scan ran, behind `sourceCache`. */
|
||||
export type UsageScanResult<TSession, TDaily> = {
|
||||
sessions: readonly TSession[]
|
||||
dailyAggregates: readonly TDaily[]
|
||||
}
|
||||
|
||||
export type UsageProvider<TSourceKey extends string, TSource, TSession, TDaily> = {
|
||||
export type UsageProvider<TSession, TDaily> = {
|
||||
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<UsageScanResult<TSourceKey, TSource, TSession, TDaily>>
|
||||
sourceCache: UsageSourceCacheRef
|
||||
): Promise<UsageScanResult<TSession, TDaily>>
|
||||
}
|
||||
|
||||
@@ -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<TestState, 'processedSources' | 'sessions' | 'dailyAggregates'>
|
||||
type TestScanResult = Pick<TestState, 'sessions' | 'dailyAggregates'>
|
||||
type TestScan = (
|
||||
worktrees: UsageScanWorktreeRef[],
|
||||
previous: TestSource[]
|
||||
sourceCache: UsageSourceCacheRef
|
||||
) => Promise<TestScanResult>
|
||||
type TestSplit = (request: UsageCacheSplitRequest) => Promise<UsageCacheSplitResult>
|
||||
|
||||
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<T>(): { promise: Promise<T>; 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<typeof vi.fn<TestScan>>
|
||||
let split: ReturnType<typeof vi.fn<TestSplit>>
|
||||
|
||||
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<TestScan>().mockResolvedValue(emptyScanResult())
|
||||
// The real split, in-process: the worker only adds the thread around it.
|
||||
split = vi.fn<TestSplit>(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<UsageCacheSplitResult>()
|
||||
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
|
||||
})
|
||||
|
||||
@@ -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<SourceKey extends string> = {
|
||||
schemaVersion: number
|
||||
worktreeFingerprint: string | null
|
||||
sessions: unknown[]
|
||||
dailyAggregates: unknown[]
|
||||
scanState: UsageProviderScanState
|
||||
} & Record<SourceKey, unknown[]>
|
||||
|
||||
type UsageProviderScanProjection<
|
||||
SourceKey extends string,
|
||||
State extends UsageProviderStoreState<SourceKey>
|
||||
> = Pick<State, SourceKey | 'sessions' | 'dailyAggregates'>
|
||||
|
||||
type UsageProviderStoreLifecycleConfig<
|
||||
SourceKey extends string,
|
||||
State extends UsageProviderStoreState<SourceKey>,
|
||||
@@ -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<UsageProviderScanProjection<SourceKey, State>>
|
||||
sourceCache: UsageSourceCacheRef
|
||||
) => Promise<Pick<State, 'sessions' | 'dailyAggregates'>>
|
||||
splitCacheFile: (request: UsageCacheSplitRequest) => Promise<UsageCacheSplitResult>
|
||||
}
|
||||
|
||||
type PublicUsageProviderScanState<DataPresenceKey extends string> = UsageProviderScanState & {
|
||||
@@ -63,7 +68,10 @@ export abstract class UsageProviderStoreLifecycle<
|
||||
State extends UsageProviderStoreState<SourceKey>,
|
||||
DataPresenceKey extends string
|
||||
> {
|
||||
// Per-source records belong to the worker and are excluded from report state.
|
||||
protected state: State
|
||||
private readonly loaded: Promise<void>
|
||||
private readonly schemaVersion: number
|
||||
private scanPromise: Promise<void> | null = null
|
||||
private tokenReporter: AgentTokenUsageReporter | null = null
|
||||
private analyticsSessionIds: AnalyticsSessionIdStore | null = null
|
||||
@@ -74,7 +82,15 @@ export abstract class UsageProviderStoreLifecycle<
|
||||
private readonly config: UsageProviderStoreLifecycleConfig<SourceKey, State, DataPresenceKey>
|
||||
) {
|
||||
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<void> {
|
||||
return this.loaded
|
||||
}
|
||||
|
||||
getScanState(): PublicUsageProviderScanState<DataPresenceKey> {
|
||||
@@ -99,17 +115,20 @@ export abstract class UsageProviderStoreLifecycle<
|
||||
|
||||
/** Await queued cache writes so quit does not drop the final snapshot. */
|
||||
async flush(): Promise<void> {
|
||||
await this.loaded
|
||||
await this.tokenReporter?.flush()
|
||||
await Promise.all([this.writer.flush(), this.analyticsSessionIds?.flush()])
|
||||
}
|
||||
|
||||
async setEnabled(enabled: boolean): Promise<PublicUsageProviderScanState<DataPresenceKey>> {
|
||||
await this.loaded
|
||||
this.state.scanState.enabled = enabled
|
||||
await this.writeToDisk()
|
||||
return this.getScanState()
|
||||
}
|
||||
|
||||
async refresh(force = false): Promise<PublicUsageProviderScanState<DataPresenceKey>> {
|
||||
await this.loaded
|
||||
if (!this.state.scanState.enabled) {
|
||||
return this.getScanState()
|
||||
}
|
||||
@@ -125,28 +144,69 @@ export abstract class UsageProviderStoreLifecycle<
|
||||
}
|
||||
|
||||
protected writeToDisk(): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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'
|
||||
}
|
||||
|
||||
@@ -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
|
||||
)
|
||||
})
|
||||
@@ -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<string>()
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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/)
|
||||
|
||||
@@ -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<TSource, TSession, TDaily> = {
|
||||
source: TSource[]
|
||||
type ProviderScanResult<TSession, TDaily> = {
|
||||
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<UsageScanWorkerValue> {
|
||||
scan(body: UsageScanWorkerScanBody): Promise<UsageScanWorkerValue> {
|
||||
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<UsageCacheSplitResult> {
|
||||
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<UsageScanWorkerValue> {
|
||||
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<UsageScanWorkerValue>,
|
||||
scan: (body: UsageScanWorkerScanBody) => Promise<UsageScanWorkerValue>,
|
||||
worktrees: UsageScanWorktreeRef[],
|
||||
previous: ClaudeUsagePersistedFile[]
|
||||
): Promise<
|
||||
ProviderScanResult<ClaudeUsagePersistedFile, ClaudeUsageSession, ClaudeUsageDailyAggregate>
|
||||
> {
|
||||
const value = await scan({ providerId: 'claude', worktrees, previous })
|
||||
if (value.providerId !== 'claude') {
|
||||
throw wrongProvider('claude', value.providerId)
|
||||
sourceCache: UsageSourceCacheRef
|
||||
): Promise<ProviderScanResult<ClaudeUsageSession, ClaudeUsageDailyAggregate>> {
|
||||
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<UsageScanWorkerValue>,
|
||||
scan: (body: UsageScanWorkerScanBody) => Promise<UsageScanWorkerValue>,
|
||||
worktrees: UsageScanWorktreeRef[],
|
||||
previous: CodexUsagePersistedFile[]
|
||||
): Promise<
|
||||
ProviderScanResult<CodexUsagePersistedFile, CodexUsageSession, CodexUsageDailyAggregate>
|
||||
> {
|
||||
const value = await scan({ providerId: 'codex', worktrees, previous })
|
||||
if (value.providerId !== 'codex') {
|
||||
throw wrongProvider('codex', value.providerId)
|
||||
sourceCache: UsageSourceCacheRef
|
||||
): Promise<ProviderScanResult<CodexUsageSession, CodexUsageDailyAggregate>> {
|
||||
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<UsageScanWorkerValue>,
|
||||
scan: (body: UsageScanWorkerScanBody) => Promise<UsageScanWorkerValue>,
|
||||
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<ProviderScanResult<OpenCodeUsageSession, OpenCodeUsageDailyAggregate>> {
|
||||
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<UsageScanWorkerValue>,
|
||||
scan: (body: UsageScanWorkerScanBody) => Promise<UsageScanWorkerValue>,
|
||||
worktrees: UsageScanWorktreeRef[],
|
||||
previous: MuseUsagePersistedFile[]
|
||||
): Promise<ProviderScanResult<MuseUsagePersistedFile, MuseUsageSession, MuseUsageDailyAggregate>> {
|
||||
const value = await scan({ providerId: 'muse', worktrees, previous })
|
||||
if (value.providerId !== 'muse') {
|
||||
throw wrongProvider('muse', value.providerId)
|
||||
sourceCache: UsageSourceCacheRef
|
||||
): Promise<ProviderScanResult<MuseUsageSession, MuseUsageDailyAggregate>> {
|
||||
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 }
|
||||
}
|
||||
|
||||
@@ -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<UsageScanWorkerValue> {
|
||||
// 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<UsageScanWorkerValue> {
|
||||
// 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<ClaudeUsagePersistedFile>(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<CodexUsagePersistedFile>(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<OpenCodeUsagePersistedDatabase>(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<MuseUsagePersistedFile>(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<void> {
|
||||
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<UsageScanWorkerResponse> {
|
||||
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) }
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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<UsageCacheSplitResult> {
|
||||
return getSharedClient().splitCacheFile(request)
|
||||
}
|
||||
|
||||
@@ -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 })
|
||||
})
|
||||
})
|
||||
@@ -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<T>(ref: UsageSourceCacheRef): Promise<T[]> {
|
||||
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<void> {
|
||||
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<UsageCacheSplitResult> {
|
||||
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 }
|
||||
}
|
||||
Reference in New Issue
Block a user