mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 00:03:08 +00:00
Fix cross-tool usage double-counting for fork/resume-copied history
- Widen Claude dedupe keys with message-id and uuid fallbacks so forks missing requestId still dedupe correctly, and drop Codex's sessionId from event keys since fork/resume rewrites session_meta.id while copying identical token_count records. - Track hasDeferredClaims per cached file/database across Claude, Codex, and OpenCode scanners so that when an owning file is deleted, only files that deferred a claim need reparsing to reclaim those turns instead of rescanning the entire corpus. - Let OpenCode's live opencode.db reclaim sessions from a stale backup claim once it reappears, avoiding a frozen stale snapshot. - Bump schema versions to invalidate caches built with the old, narrower ownership keys (#8006, #8013 follow-up).
This commit is contained in:
@@ -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<typeof Os>('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<typeof Os>('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<typeof Os>('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<typeof Os>('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')
|
||||
|
||||
@@ -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<string, ClaudeUsagePersistedFile>()
|
||||
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<string, string>()
|
||||
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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<string>()
|
||||
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<string, CodexUsagePersistedFile>()
|
||||
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<string, string>()
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<ScanResult>()
|
||||
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: [],
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -852,6 +852,7 @@ export async function parseOpenCodeUsageDatabase(
|
||||
db.pragma('query_only = ON')
|
||||
const events: OpenCodeUsageAttributedEvent[] = []
|
||||
const claimedBySessionId = new Map<string, boolean>()
|
||||
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<string, OpenCodeUsagePersistedDatabase>()
|
||||
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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
Reference in New Issue
Block a user