diff --git a/src/main/claude-usage/scanner-scan.test.ts b/src/main/claude-usage/scanner-scan.test.ts index fe84d3da50a..83fd6356d46 100644 --- a/src/main/claude-usage/scanner-scan.test.ts +++ b/src/main/claude-usage/scanner-scan.test.ts @@ -165,6 +165,82 @@ describe('scanClaudeUsageFiles', () => { expect(second.dailyAggregates[0]?.inputTokens).toBe(999) }) + it('dedupes fork copies when only message.id is present (no requestId)', async () => { + const root = await makeClaudeProjectsRoot() + const projectDir = join(root, '.claude', 'projects', 'project-a') + const originalFile = join(projectDir, 'aaaa-original.jsonl') + const forkFile = join(projectDir, 'bbbb-fork.jsonl') + + const turn = (sessionId: string, messageId: string, inputTokens: number): string => + JSON.stringify({ + type: 'assistant', + sessionId, + timestamp: '2026-04-09T10:00:00.000Z', + cwd: '/workspace/repo-a', + message: { + id: messageId, + model: 'claude-sonnet-4-6', + usage: { input_tokens: inputTokens, output_tokens: 10 } + } + }) + + await writeFile(originalFile, turn('session-1', 'msg_1', 100)) + await writeFile( + forkFile, + [turn('session-2', 'msg_1', 100), turn('session-2', 'msg_2', 50)].join('\n') + ) + + vi.resetModules() + vi.doMock('os', async () => ({ + ...(await vi.importActual('os')), + homedir: () => root + })) + const { scanClaudeUsageFiles } = await import('./scanner') + + const result = await scanClaudeUsageFiles([]) + expect(result.dailyAggregates.reduce((sum, row) => sum + row.inputTokens, 0)).toBe(150) + expect(result.processedFiles[0]?.ownedDedupeKeys).toEqual(['msg:msg_1']) + expect(result.processedFiles[1]?.ownedDedupeKeys).toEqual(['msg:msg_2']) + }) + + it('dedupes fork copies via uuid when message ids are absent', async () => { + const root = await makeClaudeProjectsRoot() + const projectDir = join(root, '.claude', 'projects', 'project-a') + const originalFile = join(projectDir, 'aaaa-original.jsonl') + const forkFile = join(projectDir, 'bbbb-fork.jsonl') + + const turn = (sessionId: string, uuid: string, inputTokens: number): string => + JSON.stringify({ + type: 'assistant', + sessionId, + uuid, + timestamp: '2026-04-09T10:00:00.000Z', + cwd: '/workspace/repo-a', + message: { + model: 'claude-sonnet-4-6', + usage: { input_tokens: inputTokens, output_tokens: 10 } + } + }) + + await writeFile(originalFile, turn('session-1', 'uuid-1', 100)) + await writeFile( + forkFile, + [turn('session-2', 'uuid-1', 100), turn('session-2', 'uuid-2', 40)].join('\n') + ) + + vi.resetModules() + vi.doMock('os', async () => ({ + ...(await vi.importActual('os')), + homedir: () => root + })) + const { scanClaudeUsageFiles } = await import('./scanner') + + const result = await scanClaudeUsageFiles([]) + expect(result.dailyAggregates.reduce((sum, row) => sum + row.inputTokens, 0)).toBe(140) + expect(result.processedFiles[0]?.ownedDedupeKeys).toEqual(['uuid:uuid-1']) + expect(result.processedFiles[1]?.ownedDedupeKeys).toEqual(['uuid:uuid-2']) + }) + it('counts turns copied into forked session files exactly once', async () => { const root = await makeClaudeProjectsRoot() const projectDir = join(root, '.claude', 'projects', 'project-a') @@ -222,7 +298,9 @@ describe('scanClaudeUsageFiles', () => { expect(totalInput).toBe(700) expect(result.processedFiles[0]?.ownedDedupeKeys).toEqual(['msg_1:req_1', 'msg_2:req_2']) + expect(result.processedFiles[0]?.hasDeferredClaims).toBe(false) expect(result.processedFiles[1]?.ownedDedupeKeys).toEqual(['msg_3:req_3']) + expect(result.processedFiles[1]?.hasDeferredClaims).toBe(true) }) it('keeps fork dedupe stable when the original file projection is reused from cache', async () => { @@ -287,6 +365,119 @@ describe('scanClaudeUsageFiles', () => { ) }) + it('reclaims fork-copied turns when the owning original file is deleted', async () => { + const root = await makeClaudeProjectsRoot() + const projectDir = join(root, '.claude', 'projects', 'project-a') + const originalFile = join(projectDir, 'aaaa-original.jsonl') + const forkFile = join(projectDir, 'bbbb-fork.jsonl') + + const turn = ( + sessionId: string, + messageId: string, + requestId: string, + inputTokens: number + ): string => + JSON.stringify({ + type: 'assistant', + sessionId, + timestamp: '2026-04-09T10:00:00.000Z', + requestId, + cwd: '/workspace/repo-a', + message: { + id: messageId, + model: 'claude-sonnet-4-6', + usage: { input_tokens: inputTokens, output_tokens: 10 } + } + }) + + await writeFile(originalFile, turn('session-1', 'msg_1', 'req_1', 100)) + await writeFile( + forkFile, + [turn('session-2', 'msg_1', 'req_1', 100), turn('session-2', 'msg_9', 'req_9', 50)].join('\n') + ) + + vi.resetModules() + vi.doMock('os', async () => ({ + ...(await vi.importActual('os')), + homedir: () => root + })) + const { scanClaudeUsageFiles } = await import('./scanner') + + const first = await scanClaudeUsageFiles([]) + expect(first.dailyAggregates.reduce((sum, aggregate) => sum + aggregate.inputTokens, 0)).toBe( + 150 + ) + expect(first.processedFiles[0]?.ownedDedupeKeys).toEqual(['msg_1:req_1']) + expect(first.processedFiles[1]?.ownedDedupeKeys).toEqual(['msg_9:req_9']) + + // Deleting the owner must reparse deferred forks so they can re-claim the + // copied turn instead of permanently under-counting. + await rm(originalFile) + + const second = await scanClaudeUsageFiles([], first.processedFiles) + expect(second.dailyAggregates.reduce((sum, aggregate) => sum + aggregate.inputTokens, 0)).toBe( + 150 + ) + expect(second.processedFiles).toHaveLength(1) + expect(second.processedFiles[0]?.ownedDedupeKeys).toEqual(['msg_1:req_1', 'msg_9:req_9']) + }) + + it('keeps unrelated cached transcripts when a different owner file is deleted', async () => { + const root = await makeClaudeProjectsRoot() + const projectDir = join(root, '.claude', 'projects', 'project-a') + const originalFile = join(projectDir, 'aaaa-original.jsonl') + const forkFile = join(projectDir, 'bbbb-fork.jsonl') + const unrelatedFile = join(projectDir, 'cccc-unrelated.jsonl') + + const turn = ( + sessionId: string, + messageId: string, + requestId: string, + inputTokens: number + ): string => + JSON.stringify({ + type: 'assistant', + sessionId, + timestamp: '2026-04-09T10:00:00.000Z', + requestId, + cwd: '/workspace/repo-a', + message: { + id: messageId, + model: 'claude-sonnet-4-6', + usage: { input_tokens: inputTokens, output_tokens: 10 } + } + }) + + await writeFile(originalFile, turn('session-1', 'msg_1', 'req_1', 100)) + await writeFile( + forkFile, + [turn('session-2', 'msg_1', 'req_1', 100), turn('session-2', 'msg_9', 'req_9', 50)].join('\n') + ) + await writeFile(unrelatedFile, turn('session-3', 'msg_u', 'req_u', 25)) + + vi.resetModules() + vi.doMock('os', async () => ({ + ...(await vi.importActual('os')), + homedir: () => root + })) + const { scanClaudeUsageFiles } = await import('./scanner') + + const first = await scanClaudeUsageFiles([]) + const unrelatedBefore = first.processedFiles.find((file) => file.path === unrelatedFile) + expect(unrelatedBefore?.hasDeferredClaims).toBe(false) + expect(unrelatedBefore?.ownedDedupeKeys).toEqual(['msg_u:req_u']) + + await rm(originalFile) + + const second = await scanClaudeUsageFiles([], first.processedFiles) + const unrelatedAfter = second.processedFiles.find((file) => file.path === unrelatedFile) + // Unrelated files must keep their cached projection identity (stat-only path). + expect(unrelatedAfter).toBe(unrelatedBefore) + expect(second.dailyAggregates.reduce((sum, aggregate) => sum + aggregate.inputTokens, 0)).toBe( + 175 + ) + }) + it('canonicalizes repeated cwd paths once per scan file', async () => { const root = await makeClaudeProjectsRoot() const projectDir = join(root, '.claude', 'projects', 'project-a') diff --git a/src/main/claude-usage/scanner.ts b/src/main/claude-usage/scanner.ts index 1269536c86e..db1c1f21f7c 100644 --- a/src/main/claude-usage/scanner.ts +++ b/src/main/claude-usage/scanner.ts @@ -29,6 +29,8 @@ type ClaudeUsageSourceRecord = { cwd?: string gitBranch?: string requestId?: string + /** Stable row id when present; preserved across fork-copied history. */ + uuid?: string isSidechain?: boolean agentId?: string message?: { @@ -274,8 +276,10 @@ function parseClaudeUsageSourceRecord( model: parsed.message?.model ?? null, cwd: parsed.cwd ?? null, gitBranch: parsed.gitBranch ?? null, - dedupeKey: - parsed.message?.id && parsed.requestId ? `${parsed.message.id}:${parsed.requestId}` : null, + // Why: forks rewrite sessionId but keep message/request ids (and usually + // uuid). Prefer the strongest stable identity available so ownership still + // works when requestId is missing on older or partial rows. + dedupeKey: buildClaudeUsageDedupeKey(parsed), inputTokens, outputTokens, cacheReadTokens, @@ -283,6 +287,22 @@ function parseClaudeUsageSourceRecord( } } +function buildClaudeUsageDedupeKey(parsed: ClaudeUsageSourceRecord): string | null { + const messageId = parsed.message?.id?.trim() + const requestId = parsed.requestId?.trim() + if (messageId && requestId) { + return `${messageId}:${requestId}` + } + if (messageId) { + return `msg:${messageId}` + } + const uuid = parsed.uuid?.trim() + if (uuid) { + return `uuid:${uuid}` + } + return null +} + export function parseClaudeUsageRecord(line: string): ClaudeUsageParsedTurn | null { const parsed = parseClaudeUsageSourceRecord(line) return parsed ? stripClaudeSourceMetadata(parsed) : null @@ -610,6 +630,18 @@ export async function scanClaudeUsageFiles( const previousByPath = new Map(previousProcessedFiles.map((file) => [file.path, file])) const worktreeLookup = await buildWorktreeLookup(worktrees) + const currentPaths = new Set(files) + // Why: when a file that owned dedupe keys is deleted, remaining forks still + // contain those turns but their caches record them as unowned. Only files + // that previously deferred claims can reclaim, so invalidate those — not the + // entire transcript corpus (histories can be gigabytes). + const lostOwnerPath = previousProcessedFiles.some( + (file) => + !currentPaths.has(file.path) && + Array.isArray(file.ownedDedupeKeys) && + file.ownedDedupeKeys.length > 0 + ) + const reusedByPath = new Map() const pathsToParse: string[] = [] for (let index = 0; index < files.length; index += FILE_SCAN_BATCH_SIZE) { @@ -620,13 +652,17 @@ export async function scanClaudeUsageFiles( const previous = previousByPath.get(filePath) // Why: Claude histories can be gigabytes. Unchanged files should pay // only stat cost on refresh while preserving exactly the old projection. + // When an owner disappears, only deferred-claim files need reparse. + const mustReclaimDeferred = lostOwnerPath && previous?.hasDeferredClaims !== false const canReuse = + !mustReclaimDeferred && previous && previous.mtimeMs === fileInfo.mtimeMs && previous.size === fileInfo.size && Array.isArray(previous.sessions) && Array.isArray(previous.dailyAggregates) && - Array.isArray(previous.ownedDedupeKeys) + Array.isArray(previous.ownedDedupeKeys) && + typeof previous.hasDeferredClaims === 'boolean' return canReuse ? previous : null }) ) @@ -650,7 +686,10 @@ export async function scanClaudeUsageFiles( const turnOwnerByDedupeKey = new Map() for (const [filePath, previous] of reusedByPath) { for (const dedupeKey of previous.ownedDedupeKeys) { - turnOwnerByDedupeKey.set(dedupeKey, filePath) + // First cached claim wins so conflicting projections stay deterministic. + if (!turnOwnerByDedupeKey.has(dedupeKey)) { + turnOwnerByDedupeKey.set(dedupeKey, filePath) + } } } @@ -666,10 +705,12 @@ export async function scanClaudeUsageFiles( // rescans assign duplicated turns to the same file deterministically. const ownedTurns: ClaudeUsageParsedTurn[] = [] const ownedDedupeKeys: string[] = [] + let hasDeferredClaims = false for (const turn of turns) { if (turn.dedupeKey) { const owner = turnOwnerByDedupeKey.get(turn.dedupeKey) if (owner !== undefined && owner !== filePath) { + hasDeferredClaims = true continue } turnOwnerByDedupeKey.set(turn.dedupeKey, filePath) @@ -681,7 +722,8 @@ export async function scanClaudeUsageFiles( parsedByPath.set(filePath, { ...processedFile, ...aggregateClaudeUsage(attributed), - ownedDedupeKeys + ownedDedupeKeys, + hasDeferredClaims }) } if (index + batch.length < pathsToParse.length) { diff --git a/src/main/claude-usage/store.ts b/src/main/claude-usage/store.ts index dfdd86ee36e..a020186ee90 100644 --- a/src/main/claude-usage/store.ts +++ b/src/main/claude-usage/store.ts @@ -19,9 +19,10 @@ import { loadKnownUsageWorktreesByRepo, type UsageWorktreeRef } from '../usage-w import type { ClaudeUsagePersistedState } from './types' import { createWorktreeRefs, getSessionProjectLabel, scanClaudeUsageFiles } from './scanner' -// Why: v4 adds cross-file turn ownership (fork-copied history dedupe). Older -// caches were built without it and can carry multiply-counted turns (#8006). -const SCHEMA_VERSION = 4 +// Why: v5 widens Claude ownership keys (message-id / uuid fallbacks). Older +// caches either lack ownership or used narrower keys and can under/over-count +// after fork reclaim (#8006). +const SCHEMA_VERSION = 5 const STALE_MS = 5 * 60_000 const AUTOMATION_ATTRIBUTION_WINDOW_MS = 5 * 60_000 diff --git a/src/main/claude-usage/types.ts b/src/main/claude-usage/types.ts index fd66f24ada4..ec497e5e098 100644 --- a/src/main/claude-usage/types.ts +++ b/src/main/claude-usage/types.ts @@ -70,6 +70,10 @@ export type ClaudeUsagePersistedFile = ClaudeUsageProcessedFile & { * sessions copy earlier turns into new files; ownership keeps each turn * counted by exactly one cached file across incremental scans. */ ownedDedupeKeys: string[] + /** True when this file saw turns already claimed by another file. When that + * owner disappears, only deferred files need reparse to reclaim — not the + * entire transcript corpus. */ + hasDeferredClaims: boolean } export type ClaudeUsageParsedTurn = { diff --git a/src/main/codex-usage/scanner-paths.test.ts b/src/main/codex-usage/scanner-paths.test.ts index ed8dd52e089..ecdc03e139f 100644 --- a/src/main/codex-usage/scanner-paths.test.ts +++ b/src/main/codex-usage/scanner-paths.test.ts @@ -1,5 +1,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { linkSync, lstatSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { + linkSync, + lstatSync, + mkdirSync, + mkdtempSync, + rmSync, + unlinkSync, + writeFileSync +} from 'node:fs' import { tmpdir } from 'node:os' import type * as NodeOs from 'node:os' import { join } from 'node:path' @@ -328,6 +336,41 @@ describe('listCodexSessionFiles', () => { expect(result.sessions[0]?.totalTokens).toBe(22) }) + it('dedupes fork copies even when session_meta id is rewritten', async () => { + const sessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + mkdirSync(sessionsDir, { recursive: true }) + const originalPath = join(sessionsDir, 'aaaa-original.jsonl') + const forkPath = join(sessionsDir, 'bbbb-fork.jsonl') + const originalPrefix = [ + `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'session-original', cwd: join(fakeHomeDir, 'repo') } + })}\n`, + usageRecord('2026-05-26T12:00:00.000Z', 10), + usageRecord('2026-05-26T12:01:00.000Z', 5, 15) + ].join('') + // Same token_count records, different session_meta id (real resume/fork shape). + const forkBody = [ + `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'session-fork', cwd: join(fakeHomeDir, 'repo') } + })}\n`, + usageRecord('2026-05-26T12:00:00.000Z', 10), + usageRecord('2026-05-26T12:01:00.000Z', 5, 15), + usageRecord('2026-05-26T12:02:00.000Z', 7, 22) + ].join('') + writeFileSync(originalPath, originalPrefix, 'utf-8') + writeFileSync(forkPath, forkBody) + + const result = await scanCodexUsageFiles([], []) + expect( + result.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(22) + expect(result.processedFiles[0]?.ownedEventKeys).toHaveLength(2) + expect(result.processedFiles[1]?.ownedEventKeys).toHaveLength(1) + expect(result.processedFiles[1]?.hasDeferredClaims).toBe(true) + }) + it('does not re-count the cumulative session for copied total-only records', async () => { const sessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') mkdirSync(sessionsDir, { recursive: true }) @@ -396,6 +439,85 @@ describe('listCodexSessionFiles', () => { ).toBe(22) }) + it('reclaims copied token events when the owning original file is deleted', async () => { + const sessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + mkdirSync(sessionsDir, { recursive: true }) + const originalPath = join(sessionsDir, 'aaaa-original.jsonl') + const forkPath = join(sessionsDir, 'bbbb-fork.jsonl') + const copiedPrefix = [ + `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'session-1', cwd: join(fakeHomeDir, 'repo') } + })}\n`, + usageRecord('2026-05-26T12:00:00.000Z', 10), + usageRecord('2026-05-26T12:01:00.000Z', 5, 15) + ].join('') + writeFileSync(originalPath, copiedPrefix, 'utf-8') + writeFileSync(forkPath, `${copiedPrefix}${usageRecord('2026-05-26T12:02:00.000Z', 7, 22)}`) + + const first = await scanCodexUsageFiles([], []) + expect( + first.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(22) + expect(first.processedFiles[0]?.ownedEventKeys).toHaveLength(2) + expect(first.processedFiles[0]?.hasDeferredClaims).toBe(false) + expect(first.processedFiles[1]?.ownedEventKeys).toHaveLength(1) + expect(first.processedFiles[1]?.hasDeferredClaims).toBe(true) + + // Deleting the owner must reparse deferred forks so they can re-claim the + // copied prefix instead of permanently under-counting. + unlinkSync(originalPath) + + const second = await scanCodexUsageFiles([], first.processedFiles) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(22) + expect(second.processedFiles).toHaveLength(1) + expect(second.processedFiles[0]?.ownedEventKeys).toHaveLength(3) + }) + + it('keeps unrelated cached rollouts when a different owner file is deleted', async () => { + const sessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + mkdirSync(sessionsDir, { recursive: true }) + const originalPath = join(sessionsDir, 'aaaa-original.jsonl') + const forkPath = join(sessionsDir, 'bbbb-fork.jsonl') + const unrelatedPath = join(sessionsDir, 'cccc-unrelated.jsonl') + const copiedPrefix = [ + `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'session-1', cwd: join(fakeHomeDir, 'repo') } + })}\n`, + usageRecord('2026-05-26T12:00:00.000Z', 10), + usageRecord('2026-05-26T12:01:00.000Z', 5, 15) + ].join('') + writeFileSync(originalPath, copiedPrefix, 'utf-8') + writeFileSync(forkPath, `${copiedPrefix}${usageRecord('2026-05-26T12:02:00.000Z', 7, 22)}`) + writeFileSync( + unrelatedPath, + [ + `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'session-9', cwd: join(fakeHomeDir, 'repo') } + })}\n`, + usageRecord('2026-05-26T13:00:00.000Z', 3) + ].join('') + ) + + const first = await scanCodexUsageFiles([], []) + const unrelatedBefore = first.processedFiles.find((file) => file.path === unrelatedPath) + expect(unrelatedBefore?.hasDeferredClaims).toBe(false) + expect(unrelatedBefore?.ownedEventKeys).toHaveLength(1) + + unlinkSync(originalPath) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const unrelatedAfter = second.processedFiles.find((file) => file.path === unrelatedPath) + expect(unrelatedAfter).toBe(unrelatedBefore) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(25) + }) + it('treats a leading total-only source suffix record as baseline', async () => { const runtimeSessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') const runtimeBridgeMarkerDir = join( diff --git a/src/main/codex-usage/scanner.ts b/src/main/codex-usage/scanner.ts index 6027ce34d0d..97e8f6f7b09 100644 --- a/src/main/codex-usage/scanner.ts +++ b/src/main/codex-usage/scanner.ts @@ -389,14 +389,14 @@ function resolveCodexUsageDelta( } function buildCodexUsageEventKey( - sessionId: string, timestamp: string, totalUsage: CodexUsageRawUsage | null, lastUsage: CodexUsageRawUsage | null ): string { // Why: fork/resume copies token_count records byte-for-byte into a new - // rollout file. Keying on the raw record (not the derived delta) makes the - // copy and the original identical regardless of surrounding parse context. + // rollout file, but session_meta.id is often rewritten to the new session. + // Key only on the raw record fields (timestamp + usage tuples) so the copy + // matches the original regardless of surrounding parse context / session id. const tupleOf = (usage: CodexUsageRawUsage | null): string => usage ? [ @@ -407,7 +407,7 @@ function buildCodexUsageEventKey( usage.totalTokens ].join(',') : '' - return [sessionId, timestamp, tupleOf(totalUsage), tupleOf(lastUsage)].join('|') + return [timestamp, tupleOf(totalUsage), tupleOf(lastUsage)].join('|') } function extractString(value: unknown): string | null { @@ -978,7 +978,7 @@ export function parseCodexUsageRecord( return { sessionId: context.sessionId, timestamp: parsed.timestamp, - eventKey: buildCodexUsageEventKey(context.sessionId, parsed.timestamp, totalUsage, lastUsage), + eventKey: buildCodexUsageEventKey(parsed.timestamp, totalUsage, lastUsage), cwd: context.currentCwd ?? context.sessionCwd, model: resolvedModel, hasInferredPricing, @@ -1016,6 +1016,7 @@ export async function parseCodexUsageFile( } const ownedEventKeys = new Set() + let hasDeferredClaims = false for await (const line of lines) { const parsed = parseCodexUsageRecord(line, context) if (!parsed) { @@ -1025,6 +1026,7 @@ export async function parseCodexUsageFile( // Events another file already owns are dropped here, but the record still // advanced context.previousTotals above, so later deltas stay correct. if (options.claimEventKey && !options.claimEventKey(parsed.eventKey)) { + hasDeferredClaims = true continue } ownedEventKeys.add(parsed.eventKey) @@ -1037,7 +1039,8 @@ export async function parseCodexUsageFile( return { ...processedFile, ...aggregateCodexUsage(events), - ownedEventKeys: [...ownedEventKeys] + ownedEventKeys: [...ownedEventKeys], + hasDeferredClaims } } @@ -1054,18 +1057,34 @@ export async function scanCodexUsageFiles( const worktreesWithCanonicalPaths = await buildWorktreesWithCanonicalPaths(worktrees) const legacySourceSkipBytesByPath = getLegacySourceSkipBytesByPath(files) + const currentPaths = new Set(files) + // Why: when a rollout that owned event keys is deleted, remaining forks still + // contain those records but their caches record them as unowned. Only files + // that previously deferred claims can reclaim, so invalidate those — not the + // entire rollout corpus. + const lostOwnerPath = previousProcessedFiles.some( + (file) => + !currentPaths.has(file.path) && + Array.isArray(file.ownedEventKeys) && + file.ownedEventKeys.length > 0 + ) + const reusedByPath = new Map() const pathsToParse: string[] = [] for (const [index, filePath] of files.entries()) { const legacySourceSkipBytes = legacySourceSkipBytesByPath.get(filePath) ?? 0 const fileInfo = await getProcessedFileInfo(filePath) const previous = previousByPath.get(filePath) + // When an owner disappears, only deferred-claim files need reparse. + const mustReclaimDeferred = lostOwnerPath && previous?.hasDeferredClaims !== false const canReuse = + !mustReclaimDeferred && legacySourceSkipBytes === 0 && previous && previous.mtimeMs === fileInfo.mtimeMs && previous.size === fileInfo.size && - Array.isArray(previous.ownedEventKeys) + Array.isArray(previous.ownedEventKeys) && + typeof previous.hasDeferredClaims === 'boolean' if (canReuse) { reusedByPath.set(filePath, previous) } else { @@ -1085,7 +1104,10 @@ export async function scanCodexUsageFiles( const eventOwnerByKey = new Map() for (const [filePath, previous] of reusedByPath) { for (const eventKey of previous.ownedEventKeys) { - eventOwnerByKey.set(eventKey, filePath) + // First cached claim wins so conflicting projections stay deterministic. + if (!eventOwnerByKey.has(eventKey)) { + eventOwnerByKey.set(eventKey, filePath) + } } } diff --git a/src/main/codex-usage/store.test.ts b/src/main/codex-usage/store.test.ts index 326ea745730..055af8f3a23 100644 --- a/src/main/codex-usage/store.test.ts +++ b/src/main/codex-usage/store.test.ts @@ -111,7 +111,7 @@ describe('CodexUsageStore', () => { it('persists a successful refresh with one compact disk write', async () => { const store = createStoreWithState({ - schemaVersion: 4, + schemaVersion: 5, scanState: { enabled: true, lastScanStartedAt: null, @@ -138,7 +138,7 @@ describe('CodexUsageStore', () => { const pendingScan = createDeferred() vi.mocked(scanCodexUsageFiles).mockReturnValueOnce(pendingScan.promise) const store = createStoreWithState({ - schemaVersion: 4, + schemaVersion: 5, scanState: { enabled: true, lastScanStartedAt: null, @@ -768,7 +768,7 @@ describe('CodexUsageStore', () => { } as unknown as CodexUsagePersistedState) expect(normalized).toEqual({ - schemaVersion: 4, + schemaVersion: 5, worktreeFingerprint: null, processedFiles: [], sessions: [], diff --git a/src/main/codex-usage/store.ts b/src/main/codex-usage/store.ts index fa294f82f4f..ca42ba34061 100644 --- a/src/main/codex-usage/store.ts +++ b/src/main/codex-usage/store.ts @@ -19,9 +19,10 @@ import { loadKnownUsageWorktreesByRepo, type UsageWorktreeRef } from '../usage-w import type { CodexUsagePersistedState } from './types' import { createWorktreeRefs, scanCodexUsageFiles } from './scanner' -// Why: v4 adds cross-file event ownership (fork/resume-copied rollout dedupe). -// Older caches were built without it and can carry multiply-counted events (#8006). -const SCHEMA_VERSION = 4 +// 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- +// scoped keys and can double-count after fork/resume (#8006). +const SCHEMA_VERSION = 5 const STALE_MS = 5 * 60_000 const AUTOMATION_ATTRIBUTION_WINDOW_MS = 5 * 60_000 diff --git a/src/main/codex-usage/types.ts b/src/main/codex-usage/types.ts index 9a269afb0be..d7e6f058b45 100644 --- a/src/main/codex-usage/types.ts +++ b/src/main/codex-usage/types.ts @@ -90,6 +90,10 @@ export type CodexUsagePersistedFile = CodexUsageProcessedFile & { * token_count records into new files; ownership keeps each record counted * by exactly one cached file across incremental scans. */ ownedEventKeys: string[] + /** True when this file saw events already claimed by another file. When that + * owner disappears, only deferred files need reparse to reclaim — not the + * entire rollout corpus. */ + hasDeferredClaims: boolean } export type CodexUsagePersistedState = { @@ -109,8 +113,9 @@ export type CodexUsagePersistedState = { export type CodexUsageParsedEvent = { sessionId: string timestamp: string - /** Raw-record identity (session, timestamp, token tuples) used to dedupe - * the same token_count record copied across fork/resume rollout files. */ + /** Raw-record identity (timestamp + token tuples) used to dedupe the same + * token_count record copied across fork/resume rollout files. Session id is + * omitted because forks rewrite session_meta while copying the record. */ eventKey: string model: string | null cwd: string | null diff --git a/src/main/opencode-usage/scanner.test.ts b/src/main/opencode-usage/scanner.test.ts index 8ef08ab5449..fbdc56bb5d5 100644 --- a/src/main/opencode-usage/scanner.test.ts +++ b/src/main/opencode-usage/scanner.test.ts @@ -1,4 +1,4 @@ -import { mkdirSync, mkdtempSync, rmSync } from 'node:fs' +import { mkdirSync, mkdtempSync, rmSync, unlinkSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' @@ -475,4 +475,59 @@ describe('scanOpenCodeUsageDatabases', () => { expect(sessionOne?.totalInputTokens).toBe(1000) expect(sessionOne?.eventCount).toBe(1) }) + + it('lets the live database reclaim sessions after a sticky backup-only claim', async () => { + // First scan only has the backup (e.g. live db temporarily missing), so it + // owns session-1 at the stale snapshot. When opencode.db reappears with + // higher totals it must reclaim the session instead of staying frozen. + writeSessionTotalsDb('opencode-backup.db', [['session-1', 400]]) + + const first = await scanOpenCodeUsageDatabases([], []) + expect( + first.dailyAggregates.reduce((total, aggregate) => total + aggregate.inputTokens, 0) + ).toBe(400) + expect(first.processedDatabases[0]?.ownedSessionIds).toEqual(['session-1']) + + writeSessionTotalsDb('opencode.db', [ + ['session-1', 1000], + ['session-2', 200] + ]) + + const second = await scanOpenCodeUsageDatabases([], first.processedDatabases) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.inputTokens, 0) + ).toBe(1200) + const sessionOne = second.sessions.find((session) => session.sessionId === 'session-1') + expect(sessionOne?.totalInputTokens).toBe(1000) + expect(sessionOne?.eventCount).toBe(1) + }) + + it('reclaims sessions when the owning live database is deleted', async () => { + const canonicalPath = writeSessionTotalsDb('opencode.db', [ + ['session-1', 1000], + ['session-2', 200] + ]) + writeSessionTotalsDb('opencode-backup.db', [ + ['session-1', 400], + ['session-9', 50] + ]) + + const first = await scanOpenCodeUsageDatabases([], []) + expect( + first.dailyAggregates.reduce((total, aggregate) => total + aggregate.inputTokens, 0) + ).toBe(1250) + + unlinkSync(canonicalPath) + + const second = await scanOpenCodeUsageDatabases([], first.processedDatabases) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.inputTokens, 0) + ).toBe(450) + expect(second.sessions.map((session) => session.sessionId).sort()).toEqual([ + 'session-1', + 'session-9' + ]) + const sessionOne = second.sessions.find((session) => session.sessionId === 'session-1') + expect(sessionOne?.totalInputTokens).toBe(400) + }) }) diff --git a/src/main/opencode-usage/scanner.ts b/src/main/opencode-usage/scanner.ts index 3150df8c1dc..4e2a3963597 100644 --- a/src/main/opencode-usage/scanner.ts +++ b/src/main/opencode-usage/scanner.ts @@ -852,6 +852,7 @@ export async function parseOpenCodeUsageDatabase( db.pragma('query_only = ON') const events: OpenCodeUsageAttributedEvent[] = [] const claimedBySessionId = new Map() + let hasDeferredClaims = false for (const row of selectUsageRows(db)) { const parsed = parseOpenCodeUsageRow(row) if (!parsed) { @@ -865,6 +866,7 @@ export async function parseOpenCodeUsageDatabase( claimedBySessionId.set(parsed.sessionId, owned) } if (!owned) { + hasDeferredClaims = true continue } const attributed = await attributeOpenCodeUsageEvent(parsed, worktrees) @@ -877,7 +879,8 @@ export async function parseOpenCodeUsageDatabase( ...aggregateOpenCodeUsage(events), ownedSessionIds: [...claimedBySessionId.entries()] .filter(([, owned]) => owned) - .map(([sessionId]) => sessionId) + .map(([sessionId]) => sessionId), + hasDeferredClaims } } finally { db.close() @@ -898,16 +901,31 @@ export async function scanOpenCodeUsageDatabases( ) const worktreesWithCanonicalPaths = await buildWorktreesWithCanonicalPaths(worktrees) + const currentPaths = new Set(dbPaths) + // Why: when a database that owned sessions is deleted, remaining siblings + // still contain those sessions but their caches record them as unowned. + // Only databases that previously deferred claims can reclaim. + const lostOwnerPath = previousProcessedDatabases.some( + (database) => + !currentPaths.has(database.path) && + Array.isArray(database.ownedSessionIds) && + database.ownedSessionIds.length > 0 + ) + const reusedByPath = new Map() const pathsToParse: string[] = [] for (const dbPath of dbPaths) { const databaseInfo = await getProcessedDatabaseInfo(dbPath) const previous = previousByPath.get(dbPath) + // When an owner disappears, only deferred-claim databases need reparse. + const mustReclaimDeferred = lostOwnerPath && previous?.hasDeferredClaims !== false const canReuse = + !mustReclaimDeferred && previous && previous.mtimeMs === databaseInfo.mtimeMs && previous.size === databaseInfo.size && - Array.isArray(previous.ownedSessionIds) + Array.isArray(previous.ownedSessionIds) && + typeof previous.hasDeferredClaims === 'boolean' if (canReuse) { reusedByPath.set(dbPath, previous) } else { @@ -915,6 +933,25 @@ export async function scanOpenCodeUsageDatabases( } } + // Why: a sticky backup claim from a scan where opencode.db was missing would + // otherwise freeze a still-growing session at the backup snapshot when the + // live db reappears. Reparse lower-priority siblings whenever a higher- + // priority path is being re-evaluated so the live db can reclaim sessions + // without double-counting cached backup aggregates. + const demotedReusePaths: string[] = [] + for (const dbPath of reusedByPath.keys()) { + const higherPriorityParsing = pathsToParse.some( + (candidate) => compareOpenCodeClaimPriority(candidate, dbPath) < 0 + ) + if (higherPriorityParsing) { + demotedReusePaths.push(dbPath) + } + } + for (const dbPath of demotedReusePaths) { + reusedByPath.delete(dbPath) + pathsToParse.push(dbPath) + } + // Why: `opencode-*.db` siblings are typically stale copies of `opencode.db` // (backups), so mergeSessions would double every duplicated session (#8006). // Each session is counted from exactly one database. The canonical live db diff --git a/src/main/opencode-usage/store.test.ts b/src/main/opencode-usage/store.test.ts index 969fdfdd6f3..aea35628f60 100644 --- a/src/main/opencode-usage/store.test.ts +++ b/src/main/opencode-usage/store.test.ts @@ -284,7 +284,8 @@ describe('OpenCodeUsageStore', () => { size: 2, sessions: [makeSession()], dailyAggregates: [makeDaily()], - ownedSessionIds: ['session-1'] + ownedSessionIds: ['session-1'], + hasDeferredClaims: false } ], sessions: [makeSession()], @@ -302,7 +303,8 @@ describe('OpenCodeUsageStore', () => { size: 2, sessions: [], dailyAggregates: [], - ownedSessionIds: [] + ownedSessionIds: [], + hasDeferredClaims: false } ] }).processedDatabases diff --git a/src/main/opencode-usage/types.ts b/src/main/opencode-usage/types.ts index 61fbc43b2c4..cf8d2c6a8d0 100644 --- a/src/main/opencode-usage/types.ts +++ b/src/main/opencode-usage/types.ts @@ -90,6 +90,9 @@ export type OpenCodeUsagePersistedDatabase = OpenCodeUsageProcessedDatabase & { * duplicate sessions; ownership keeps each session counted by exactly one * cached database across incremental scans. */ ownedSessionIds: string[] + /** True when this database saw sessions already claimed by another database. + * When that owner disappears, only deferred databases need reparse. */ + hasDeferredClaims: boolean } export type OpenCodeUsagePersistedState = {