diff --git a/src/main/claude-usage/claude-usage-automation-attribution.ts b/src/main/claude-usage/claude-usage-automation-attribution.ts index 0b4a8a9844f..cdbfb5fb916 100644 --- a/src/main/claude-usage/claude-usage-automation-attribution.ts +++ b/src/main/claude-usage/claude-usage-automation-attribution.ts @@ -1,6 +1,7 @@ import type { AutomationRunUsage } from '../../shared/automations-types' import type { ClaudeUsagePersistedState } from './types' import { estimateCostUsd } from './claude-model-pricing' +import { shouldForceAutomationUsageScan } from '../usage/automation-usage-scan-forcing' const AUTOMATION_ATTRIBUTION_WINDOW_MS = 5 * 60_000 @@ -14,16 +15,7 @@ export type AutomationUsageLookupInput = { type ClaudeUsageStateAccess = { getState: () => ClaudeUsagePersistedState refresh: (force: boolean) => Promise<{ lastScanError: string | null }> -} - -function shouldForceAutomationUsageScan( - state: ClaudeUsagePersistedState, - completedAt: number -): boolean { - const { lastScanCompletedAt, lastScanError } = state.scanState - // Why: attribution needs a scan after the run finishes, but repeated - // lookups after that point should not rescan all Claude transcript history. - return Boolean(lastScanError) || lastScanCompletedAt === null || lastScanCompletedAt < completedAt + isScanning: () => boolean } export async function resolveAutomationRunUsage( @@ -61,7 +53,11 @@ export async function resolveAutomationRunUsage( } const scanState = await access.refresh( - shouldForceAutomationUsageScan(access.getState(), input.completedAt) + shouldForceAutomationUsageScan( + access.getState().scanState, + input.completedAt, + access.isScanning() + ) ) if (scanState.lastScanError) { return unavailable('scan_failed', scanState.lastScanError) diff --git a/src/main/claude-usage/store.test.ts b/src/main/claude-usage/store.test.ts index f71d0269050..594be466e50 100644 --- a/src/main/claude-usage/store.test.ts +++ b/src/main/claude-usage/store.test.ts @@ -51,6 +51,42 @@ function createStoreWithState(state: Partial): Claude return store } +function createWorktreeUsageSession(worktreeId: string) { + const tokens = { + turnCount: 1, + inputTokens: 1000, + outputTokens: 500, + cacheReadTokens: 200, + cacheWriteTokens: 100, + cacheWrite1hTokens: 0 + } + return { + sessionId: 'session-1', + firstTimestamp: '2026-04-09T15:00:00.000Z', + lastTimestamp: '2026-04-09T15:05:00.000Z', + model: 'claude-sonnet-4-6', + lastCwd: '/workspace/repo-a', + lastGitBranch: 'feature/a', + primaryWorktreeId: worktreeId, + primaryRepoId: 'repo-1', + totalInputTokens: 1000, + totalOutputTokens: 500, + totalCacheReadTokens: 200, + totalCacheWriteTokens: 100, + totalCacheWrite1hTokens: 0, + ...tokens, + locationBreakdown: [ + { + locationKey: `worktree:${worktreeId}`, + projectLabel: 'Repo A', + repoId: 'repo-1', + worktreeId, + ...tokens + } + ] + } +} + describe('ClaudeUsageStore', () => { let tempUserData: string @@ -582,38 +618,7 @@ describe('ClaudeUsageStore', () => { lastScanCompletedAt: 2, lastScanError: null }, - sessions: [ - { - sessionId: 'session-1', - firstTimestamp: '2026-04-09T15:00:00.000Z', - lastTimestamp: '2026-04-09T15:05:00.000Z', - model: 'claude-sonnet-4-6', - lastCwd: '/workspace/repo-a', - lastGitBranch: 'feature/a', - primaryWorktreeId: worktreeId, - primaryRepoId: 'repo-1', - turnCount: 1, - totalInputTokens: 1000, - totalOutputTokens: 500, - totalCacheReadTokens: 200, - totalCacheWriteTokens: 100, - totalCacheWrite1hTokens: 0, - locationBreakdown: [ - { - locationKey: `worktree:${worktreeId}`, - projectLabel: 'Repo A', - repoId: 'repo-1', - worktreeId, - turnCount: 1, - inputTokens: 1000, - outputTokens: 500, - cacheReadTokens: 200, - cacheWriteTokens: 100, - cacheWrite1hTokens: 0 - } - ] - } - ] + sessions: [createWorktreeUsageSession(worktreeId)] }) const refreshMock = vi.fn().mockResolvedValue({ enabled: true, @@ -648,6 +653,48 @@ describe('ClaudeUsageStore', () => { expect(refreshMock).toHaveBeenCalledWith(false) }) + it('forces one scan per run and stops re-forcing after a failed attempt', async () => { + const completedAt = Date.parse('2026-04-09T15:06:00.000Z') + const scanError = 'EMFILE: too many open files' + const failedScanState = (lastScanStartedAt: number) => ({ + enabled: true, + lastScanStartedAt, + lastScanCompletedAt: completedAt - 60_000, + lastScanError: scanError + }) + const scanStateResult = { + enabled: true, + isScanning: false, + lastScanStartedAt: completedAt - 60_000, + lastScanCompletedAt: completedAt - 60_000, + lastScanError: scanError, + hasAnyClaudeData: false + } + const request = { + worktreeId: 'repo-1::/workspace/repo-a', + terminalSessionId: 'tab-1', + startedAt: completedAt - 120_000, + completedAt + } + + const beforeAttempt = createStoreWithState({ + scanState: failedScanState(completedAt - 60_000) + }) + const beforeRefresh = vi.spyOn(beforeAttempt, 'refresh').mockResolvedValue(scanStateResult) + await beforeAttempt.getAutomationRunUsage(request) + + expect(beforeRefresh).toHaveBeenCalledWith(true) + + // That forced scan failed: it recorded an attempt but no completion. Later + // lookups must not keep forcing a full rescan of all Claude history. + const afterAttempt = createStoreWithState({ scanState: failedScanState(completedAt + 1000) }) + const afterRefresh = vi.spyOn(afterAttempt, 'refresh').mockResolvedValue(scanStateResult) + const usage = await afterAttempt.getAutomationRunUsage(request) + + expect(afterRefresh).toHaveBeenCalledWith(false) + expect(usage.unavailableReason).toBe('scan_failed') + }) + it('adapts Claude scans to pretty-printed cache persistence', async () => { const store = createStoreWithState({ schemaVersion: 5, @@ -664,4 +711,61 @@ describe('ClaudeUsageStore', () => { expect(scanClaudeUsageFiles).toHaveBeenCalledWith([], []) expect(readFileSync(join(tempUserData, 'orca-claude-usage.json'), 'utf-8')).toContain('\n') }) + + it('joins a scan that is already in flight when the run finished before it started', async () => { + const worktreeId = 'repo-1::/workspace/repo-a' + const store = createStoreWithState({ + scanState: { + enabled: true, + lastScanStartedAt: null, + lastScanCompletedAt: null, + lastScanError: null + } + }) + // Prime the worktree fingerprint so an unforced refresh can return early. + vi.mocked(scanClaudeUsageFiles).mockResolvedValue({ + processedFiles: [], + sessions: [], + dailyAggregates: [] + }) + await store.refresh(true) + + const completedAt = Date.now() + 10_000 + vi.setSystemTime(new Date(completedAt + 1_000)) + + let startScan = () => {} + let finishScan = () => {} + const scanStarted = new Promise((resolve) => { + startScan = resolve + }) + const scanFinished = new Promise((resolve) => { + finishScan = resolve + }) + vi.mocked(scanClaudeUsageFiles).mockImplementationOnce(async () => { + startScan() + await scanFinished + return { + processedFiles: [], + sessions: [createWorktreeUsageSession(worktreeId)], + dailyAggregates: [] + } + }) + + const inFlight = store.refresh(true) + await scanStarted + + const usage = store.getAutomationRunUsage({ + worktreeId, + terminalSessionId: 'session-1', + startedAt: completedAt - 60_000, + completedAt + }) + finishScan() + await inFlight + + // The in-flight scan's start time is not a finished attempt, so the lookup + // forces and rides that scan instead of reading a pre-run cache. + expect((await usage).status).toBe('known') + expect((await usage).providerSessionId).toBe('session-1') + }) }) diff --git a/src/main/claude-usage/store.ts b/src/main/claude-usage/store.ts index 921c09c8732..085b4453147 100644 --- a/src/main/claude-usage/store.ts +++ b/src/main/claude-usage/store.ts @@ -139,7 +139,8 @@ export class ClaudeUsageStore extends UsageProviderStoreLifecycle< async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise { return resolveAutomationRunUsage(input, { getState: () => this.state, - refresh: (force) => this.refresh(force) + refresh: (force) => this.refresh(force), + isScanning: () => this.getScanState().isScanning }) } } diff --git a/src/main/codex-usage/codex-automation-run-attribution.ts b/src/main/codex-usage/codex-automation-run-attribution.ts index 10b366d6756..0d5cb8e9a6f 100644 --- a/src/main/codex-usage/codex-automation-run-attribution.ts +++ b/src/main/codex-usage/codex-automation-run-attribution.ts @@ -1,6 +1,7 @@ import type { AutomationRunUsage } from '../../shared/automations-types' import type { CodexUsagePersistedState } from './types' import { estimateCostUsd } from './codex-usage-cost-estimate' +import { shouldForceAutomationUsageScan } from '../usage/automation-usage-scan-forcing' const AUTOMATION_ATTRIBUTION_WINDOW_MS = 5 * 60_000 @@ -15,16 +16,7 @@ type CodexAutomationAttributionDeps = { /** Callback, not a snapshot: refresh mutates persisted state in place. */ getState: () => CodexUsagePersistedState refresh: (force: boolean) => Promise<{ lastScanError: string | null }> -} - -function shouldForceAutomationUsageScan( - scanState: CodexUsagePersistedState['scanState'], - completedAt: number -): boolean { - const { lastScanCompletedAt, lastScanError } = scanState - // Why: attribution needs a scan after the run finishes, but repeated - // lookups after that point should not rescan all Codex session history. - return Boolean(lastScanError) || lastScanCompletedAt === null || lastScanCompletedAt < completedAt + isScanning: () => boolean } export async function resolveCodexAutomationRunUsage( @@ -62,7 +54,7 @@ export async function resolveCodexAutomationRunUsage( } const scanState = await deps.refresh( - shouldForceAutomationUsageScan(deps.getState().scanState, input.completedAt) + shouldForceAutomationUsageScan(deps.getState().scanState, input.completedAt, deps.isScanning()) ) if (scanState.lastScanError) { return unavailable('scan_failed', scanState.lastScanError) diff --git a/src/main/codex-usage/codex-rollout-file-parse.ts b/src/main/codex-usage/codex-rollout-file-parse.ts new file mode 100644 index 00000000000..01755ee69aa --- /dev/null +++ b/src/main/codex-usage/codex-rollout-file-parse.ts @@ -0,0 +1,191 @@ +import { basename } from 'node:path' +import { stat } from 'node:fs/promises' +import { readJsonlLinesFromOffset } from '../usage/jsonl-line-offsets' +import { attributeCodexUsageEvent } from './codex-usage-event-attribution' +import type { UsageWorktreeResolver } from '../usage/usage-worktree-resolver' +import { parseCodexUsageRecord, type CodexUsageParseContext } from './codex-usage-record-parser' +import { codexUsageAggregation } from './codex-usage-aggregation' +import { + buildCodexRolloutResumeState, + resolveCodexRolloutResume +} from './codex-rollout-resume-state' +import type { + CodexUsageAttributedEvent, + CodexUsageDailyAggregate, + CodexUsageParseResumeState, + CodexUsagePersistedFile, + CodexUsageProcessedFile, + CodexUsageSession +} from './types' + +const { finalizeSessions, mergeSessions, mergeDailyAggregates, sortDailyAggregates } = + codexUsageAggregation + +export type CodexRolloutParseOptions = { + /** Suffix-only parse for a diverged legacy copied-session bridge. */ + legacySourceSkipBytes?: number + claimEventKey?: (eventKey: string) => boolean + /** Resume point verified by the caller, with the cached projection to extend. */ + resume?: { state: CodexUsageParseResumeState; previous: CodexUsagePersistedFile } +} + +export async function getProcessedFileInfo(filePath: string): Promise { + const fileStat = await stat(filePath) + return { + path: filePath, + mtimeMs: fileStat.mtimeMs, + size: fileStat.size + } +} + +function mergeRolloutProjections( + previous: CodexUsagePersistedFile, + appended: { sessions: CodexUsageSession[]; dailyAggregates: CodexUsageDailyAggregate[] } +): { sessions: CodexUsageSession[]; dailyAggregates: CodexUsageDailyAggregate[] } { + const sessionsById = new Map() + mergeSessions(sessionsById, previous.sessions) + mergeSessions(sessionsById, appended.sessions) + const dailyByKey = new Map() + mergeDailyAggregates(dailyByKey, previous.dailyAggregates) + mergeDailyAggregates(dailyByKey, appended.dailyAggregates) + return { + sessions: finalizeSessions(sessionsById), + dailyAggregates: sortDailyAggregates(dailyByKey) + } +} + +function createParseContext( + filePath: string, + options: CodexRolloutParseOptions +): CodexUsageParseContext { + const resume = options.resume?.state + if (resume) { + return { + sessionId: resume.sessionId, + sessionCwd: resume.sessionCwd, + currentCwd: resume.currentCwd, + currentModel: resume.currentModel, + previousTotals: resume.previousTotals, + totalOnlyBaselinePending: false + } + } + return { + sessionId: basename(filePath, '.jsonl'), + sessionCwd: null, + currentCwd: null, + currentModel: null, + previousTotals: null, + // Why: suffix-only legacy copy parsing lacks the copied prefix context. A + // leading total-only snapshot is a baseline, not the suffix's billable delta. + totalOnlyBaselinePending: (options.legacySourceSkipBytes ?? 0) > 0 + } +} + +export async function parseCodexUsageFile( + filePath: string, + resolveWorktree: UsageWorktreeResolver, + options: CodexRolloutParseOptions = {} +): Promise { + // Why: the caller verified this resume point while walking the directory, and + // every file discovered or parsed since then has run in between. Re-verify + // here, against the file about to be read, or a rollout replaced in that gap + // gets the cached session id, cwd, model and running totals stitched onto an + // unrelated file's records — and `processedFile` below re-stats to the new + // size, so the reuse path then freezes the corrupted projection. + if ( + options.resume && + (await resolveCodexRolloutResume(filePath, options.resume.previous)) === null + ) { + return parseCodexUsageFile(filePath, resolveWorktree, { ...options, resume: undefined }) + } + + const processedFile = await getProcessedFileInfo(filePath) + const legacySourceSkipBytes = options.legacySourceSkipBytes ?? 0 + const startOffset = options.resume?.state.parsedBytes ?? legacySourceSkipBytes + const context = createParseContext(filePath, options) + + const events: CodexUsageAttributedEvent[] = [] + const ownedEventKeys = new Set() + let hasDeferredClaims = false + let parsedBytes = startOffset + // Points at the context as of `parsedBytes`, which excludes a partial tail. + let resumeContext = context + let partialTailProducedEvent = false + + for await (const { line, endOffset, terminated } of readJsonlLinesFromOffset( + filePath, + startOffset + )) { + if (!terminated) { + // Only the final fragment can be unterminated, and the next scan re-reads + // it, so its context edits must not leak into the persisted resume point. + resumeContext = { ...context } + } + const parsed = parseCodexUsageRecord(line, context) + if (terminated) { + parsedBytes = endOffset + } else if (parsed) { + partialTailProducedEvent = true + } + if (!parsed) { + continue + } + // Why: fork/resume rollouts start with a copied prefix of the parent file. + // 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) + const attributed = await attributeCodexUsageEvent(parsed, resolveWorktree) + if (attributed) { + events.push(attributed) + } + } + + // A counted-but-unterminated tail would be counted again on resume, and a + // legacy suffix offset is recomputed per scan, so neither may be resumed. A + // prefix under the resumable floor is turned away by the builder itself. + const resumeStateSuppressed = partialTailProducedEvent || legacySourceSkipBytes > 0 + const parseResumeState = resumeStateSuppressed + ? null + : await buildCodexRolloutResumeState( + filePath, + parsedBytes, + resumeContext, + // Already verified against the file at the top of this scan. + options.resume?.state.headDigest ?? null + ) + + // Why: a resume point only exists past the resumable floor and `parsedBytes` + // only grows, so the builder's other null — a prefix too short to be worth + // resuming — is unreachable here and this null means a short read: the file + // shrank past the prefix this parse merged history for, after the + // re-verification above and during the read. `processedFile` already re-stat'd + // to the smaller size, so persisting that pair would let the next scan reuse a + // pre-truncation total forever. An unterminated tail proves the file still + // runs past the resume offset, so it cannot be this case. + if (options.resume && !resumeStateSuppressed && parseResumeState === null) { + return parseCodexUsageFile(filePath, resolveWorktree, { ...options, resume: undefined }) + } + + const appended = codexUsageAggregation.aggregate(events) + const previous = options.resume?.previous + if (!previous) { + return { + ...processedFile, + ...appended, + ownedEventKeys: [...ownedEventKeys], + hasDeferredClaims, + parseResumeState + } + } + return { + ...processedFile, + ...mergeRolloutProjections(previous, appended), + ownedEventKeys: [...new Set([...previous.ownedEventKeys, ...ownedEventKeys])], + hasDeferredClaims: previous.hasDeferredClaims || hasDeferredClaims, + parseResumeState + } +} diff --git a/src/main/codex-usage/codex-rollout-resume-state.test.ts b/src/main/codex-usage/codex-rollout-resume-state.test.ts new file mode 100644 index 00000000000..bacec7f50a9 --- /dev/null +++ b/src/main/codex-usage/codex-rollout-resume-state.test.ts @@ -0,0 +1,146 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { mkdtempSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + buildCodexRolloutResumeState, + MIN_RESUMABLE_PREFIX_BYTES, + resolveCodexRolloutResume +} from './codex-rollout-resume-state' +import type { CodexUsageParseContext } from './codex-usage-record-parser' +import type { CodexUsagePersistedFile } from './types' + +/** Mirrors HEAD_WINDOW_BYTES / BOUNDARY_WINDOW_BYTES in the module under test. */ +const WINDOW_BYTES = 4096 + +const context: CodexUsageParseContext = { + sessionId: 'session-resume-state', + sessionCwd: null, + currentCwd: null, + currentModel: null, + previousTotals: null +} + +let workDir: string + +function writeRollout(name: string, bytes: number): string { + const filePath = join(workDir, name) + writeFileSync(filePath, 'x'.repeat(bytes), 'utf-8') + return filePath +} + +function persistedFile( + filePath: string, + parseResumeState: CodexUsagePersistedFile['parseResumeState'] +): CodexUsagePersistedFile { + const fileStat = statSync(filePath) + return { + path: filePath, + mtimeMs: fileStat.mtimeMs, + size: fileStat.size, + sessions: [], + dailyAggregates: [], + ownedEventKeys: [], + hasDeferredClaims: false, + parseResumeState + } +} + +beforeEach(() => { + workDir = mkdtempSync(join(tmpdir(), 'orca-codex-resume-state-')) +}) + +afterEach(() => { + rmSync(workDir, { recursive: true, force: true }) +}) + +describe('buildCodexRolloutResumeState', () => { + // Without the short-read guard the digest is recorded over whatever bytes did + // arrive. A later scan hashes the same too-short range, gets the same digest, + // and accepts a resume point past EOF — so the caller resumes over history + // that is no longer in the file. + it('refuses to record an offset the file no longer reaches', async () => { + const filePath = writeRollout('short-prefix.jsonl', 2000) + + await expect( + buildCodexRolloutResumeState(filePath, 5 * WINDOW_BYTES, context) + ).resolves.toBeNull() + }) + + // Same guard, disjoint-window layout: the head window is fully readable and + // only the boundary window falls past the end of the file. + it('refuses when only the boundary window falls past the end of the file', async () => { + const filePath = writeRollout('short-boundary.jsonl', 4 * WINDOW_BYTES) + + await expect( + buildCodexRolloutResumeState(filePath, 5 * WINDOW_BYTES, context) + ).resolves.toBeNull() + }) + + // Verifying a short prefix costs more than re-reading it, so no resume point + // is recorded there. This is also what keeps every digest read past the point + // where the two windows could overlap. + it('refuses a prefix at or under the resumable floor', async () => { + const filePath = writeRollout('at-floor.jsonl', 4 * WINDOW_BYTES) + + expect(MIN_RESUMABLE_PREFIX_BYTES).toBe(3 * WINDOW_BYTES) + await expect( + buildCodexRolloutResumeState(filePath, MIN_RESUMABLE_PREFIX_BYTES, context) + ).resolves.toBeNull() + await expect( + buildCodexRolloutResumeState(filePath, MIN_RESUMABLE_PREFIX_BYTES + 1, context) + ).resolves.not.toBeNull() + }) + + it('records an offset the file still reaches', async () => { + const filePath = writeRollout('full-prefix.jsonl', 4 * WINDOW_BYTES) + + const state = await buildCodexRolloutResumeState(filePath, 4 * WINDOW_BYTES, context) + + expect(state?.parsedBytes).toBe(4 * WINDOW_BYTES) + expect(state?.headDigest).toMatch(new RegExp(`^${WINDOW_BYTES}:`)) + }) +}) + +describe('resolveCodexRolloutResume', () => { + it('rejects a recorded offset once the file has been truncated past it', async () => { + const filePath = writeRollout('truncated.jsonl', 4 * WINDOW_BYTES) + const state = await buildCodexRolloutResumeState(filePath, 4 * WINDOW_BYTES, context) + expect(state).not.toBeNull() + const previous = persistedFile(filePath, state) + + writeRollout('truncated.jsonl', WINDOW_BYTES) + + await expect(resolveCodexRolloutResume(filePath, previous)).resolves.toBeNull() + }) + + // A persisted offset under the floor would put the boundary window at a + // negative start, so this is input validation on the cache file, not a + // restatement of the builder's guard. + it('rejects a persisted offset under the floor', async () => { + const filePath = writeRollout('under-floor.jsonl', 4 * WINDOW_BYTES) + const previous = persistedFile(filePath, { + parsedBytes: 100, + boundaryDigest: `${WINDOW_BYTES}:unused`, + headDigest: `${WINDOW_BYTES}:unused`, + physicalFileId: null, + sessionId: 'session-under-floor', + sessionCwd: null, + currentCwd: null, + currentModel: null, + previousTotals: null + }) + + await expect(resolveCodexRolloutResume(filePath, previous)).resolves.toBeNull() + }) + + it('accepts a recorded offset when the file only grew', async () => { + const filePath = writeRollout('grown.jsonl', 4 * WINDOW_BYTES) + const state = await buildCodexRolloutResumeState(filePath, 4 * WINDOW_BYTES, context) + const previous = persistedFile(filePath, state) + + writeRollout('grown.jsonl', 5 * WINDOW_BYTES) + + await expect(resolveCodexRolloutResume(filePath, previous)).resolves.toEqual(state) + }) +}) diff --git a/src/main/codex-usage/codex-rollout-resume-state.ts b/src/main/codex-usage/codex-rollout-resume-state.ts new file mode 100644 index 00000000000..368d50bddf9 --- /dev/null +++ b/src/main/codex-usage/codex-rollout-resume-state.ts @@ -0,0 +1,183 @@ +/** + * Decides whether a grown rollout can be parsed from where the last scan + * stopped. A wrong answer here silently corrupts usage totals, so every check + * fails closed: anything unproven falls back to a full reparse. + */ +import { createReadStream } from 'node:fs' +import { stat } from 'node:fs/promises' +import { createHash } from 'node:crypto' +import type { CodexUsageParseContext } from './codex-usage-record-parser' +import type { CodexUsageParseResumeState, CodexUsagePersistedFile } from './types' + +/** Bytes hashed immediately before the resume offset. Large enough to span a + * whole token_count record, small enough that verifying it is free next to + * re-reading a multi-megabyte rollout. */ +const BOUNDARY_WINDOW_BYTES = 4096 + +/** Bytes hashed at the very start of the parsed prefix. The boundary window can + * only speak for the bytes next to the resume offset, so without this a + * same-length rewrite that leaves the tail intact resumes over changed + * history. Every realistic rotation or rewrite of a rollout replaces the + * leading session_meta line, which lands in this window. */ +const HEAD_WINDOW_BYTES = 4096 + +/** + * Shortest prefix worth resuming over. A resumed scan verifies the prefix + * twice — once when the scanner plans the resume, once against the file it is + * about to read — and then records the moved boundary, so it pays five windows + * whatever the file size. A cold reparse pays the prefix itself plus the two + * windows it records. Below this length reading the whole file is cheaper, and + * for a prefix short enough that the two windows overlap it is far cheaper, + * because each verification then rehashes the entire prefix. + * + * Measured on the byte oracle: a 12,191 B prefix costs 21,234 B resumed against + * 21,137 B cold; at 13,699 B it is 21,234 B against 22,645 B. + */ +export const MIN_RESUMABLE_PREFIX_BYTES = 3 * BOUNDARY_WINDOW_BYTES + +/** Below the floor the two windows could also overlap, so every offset that + * reaches the digest reads is guaranteed to give them a disjoint layout. */ +export function isResumablePrefixLength(parsedBytes: number): boolean { + return parsedBytes > MIN_RESUMABLE_PREFIX_BYTES +} + +async function readWindowDigest( + filePath: string, + start: number, + endExclusive: number +): Promise { + const expectedBytes = endExclusive - start + const hash = createHash('sha256') + let readBytes = 0 + const stream = createReadStream(filePath, { start, end: endExclusive - 1 }) + for await (const chunk of stream) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk) + hash.update(buffer) + readBytes += buffer.length + } + // A short read means the file no longer reaches the offset we recorded. + return readBytes === expectedBytes ? `${expectedBytes}:${hash.digest('hex')}` : null +} + +type PrefixDigests = { headDigest: string; boundaryDigest: string } + +/** + * Both window digests for a prefix of `parsedBytes`, or null if the file no + * longer reaches that offset. The window layout is a pure function of + * `parsedBytes`, so a later verification hashes exactly the same ranges. Only + * called for a prefix past `MIN_RESUMABLE_PREFIX_BYTES`, so the two windows are + * always disjoint and always their full size. + * + * `carriedHeadDigest` lets a caller that already verified the head window this + * scan skip re-reading those bytes. Deferring to it is self-correcting: if the + * head did change after that read, the next scan compares against the carried + * value and falls back to a full reparse. + */ +async function readPrefixDigests( + filePath: string, + parsedBytes: number, + carriedHeadDigest: string | null = null +): Promise { + const headDigest = carriedHeadDigest?.startsWith(`${HEAD_WINDOW_BYTES}:`) + ? carriedHeadDigest + : await readWindowDigest(filePath, 0, HEAD_WINDOW_BYTES) + if (headDigest === null) { + return null + } + const boundaryDigest = await readWindowDigest( + filePath, + parsedBytes - BOUNDARY_WINDOW_BYTES, + parsedBytes + ) + return boundaryDigest === null ? null : { headDigest, boundaryDigest } +} + +async function readPhysicalFileId(filePath: string): Promise { + try { + const fileStat = await stat(filePath) + return fileStat.ino === 0 ? null : `${fileStat.dev}:${fileStat.ino}` + } catch { + return null + } +} + +function isUsableResumeState( + resume: CodexUsageParseResumeState | null | undefined +): resume is CodexUsageParseResumeState { + return ( + resume != null && + Number.isInteger(resume.parsedBytes) && + resume.parsedBytes >= 0 && + typeof resume.boundaryDigest === 'string' && + // State persisted before the head window existed cannot be verified. + typeof resume.headDigest === 'string' && + // Caches written before the floor existed can hold a shorter prefix. + isResumablePrefixLength(resume.parsedBytes) && + typeof resume.sessionId === 'string' + ) +} + +export async function buildCodexRolloutResumeState( + filePath: string, + parsedBytes: number, + context: CodexUsageParseContext, + verifiedHeadDigest: string | null = null +): Promise { + if (!isResumablePrefixLength(parsedBytes)) { + return null + } + const digests = await readPrefixDigests(filePath, parsedBytes, verifiedHeadDigest) + if (digests === null) { + return null + } + return { + parsedBytes, + boundaryDigest: digests.boundaryDigest, + headDigest: digests.headDigest, + physicalFileId: await readPhysicalFileId(filePath), + sessionId: context.sessionId, + sessionCwd: context.sessionCwd, + currentCwd: context.currentCwd, + currentModel: context.currentModel, + previousTotals: context.previousTotals + } +} + +/** + * Returns the resume point only when the recorded prefix still looks like the + * file's prefix. Deliberately does not consult any timestamp: filesystems vary + * in clock granularity, so a rewrite can land under the mtime, ctime or + * birthtime the cache already holds. `physicalFileId` is a cheap extra catch + * rather than a rotation check — ext4 and overlayfs hand a recreated path the + * inode the old file freed, so it only fires on filesystems that allocate a + * fresh one. Truncation needs no separate check: a file that no longer reaches + * the offset cannot produce the recorded digests. + * + * Bounded by design, so for a prefix larger than the two windows together this + * cannot prove every byte between them is intact. + */ +export async function resolveCodexRolloutResume( + filePath: string, + previous: CodexUsagePersistedFile | undefined +): Promise { + const resume = previous?.parseResumeState + if (!isUsableResumeState(resume)) { + return null + } + const physicalFileId = await readPhysicalFileId(filePath) + if ( + resume.physicalFileId !== null && + physicalFileId !== null && + resume.physicalFileId !== physicalFileId + ) { + return null + } + const digests = await readPrefixDigests(filePath, resume.parsedBytes) + if (digests === null) { + return null + } + return digests.boundaryDigest === resume.boundaryDigest && + digests.headDigest === resume.headDigest + ? resume + : null +} diff --git a/src/main/codex-usage/codex-usage-aggregation.ts b/src/main/codex-usage/codex-usage-aggregation.ts new file mode 100644 index 00000000000..fd65d6ffc2f --- /dev/null +++ b/src/main/codex-usage/codex-usage-aggregation.ts @@ -0,0 +1,23 @@ +import { createUsageEventAggregation } from '../usage/usage-event-aggregation' +import type { CodexUsageAttributedEvent } from './types' + +type CodexUsageMetric = { hasInferredPricing: boolean } + +export const codexUsageAggregation = createUsageEventAggregation< + CodexUsageAttributedEvent, + CodexUsageMetric +>({ + metric: { + empty: () => ({ hasInferredPricing: false }), + fromEvent: (event) => ({ hasInferredPricing: event.hasInferredPricing }), + fold: (target, source) => { + target.hasInferredPricing ||= source.hasInferredPricing + } + }, + cloneSessionForMerge: (session) => ({ + ...session, + locationBreakdown: session.locationBreakdown.map((entry) => ({ ...entry })), + modelBreakdown: session.modelBreakdown.map((entry) => ({ ...entry })), + locationModelBreakdown: session.locationModelBreakdown.map((entry) => ({ ...entry })) + }) +}) diff --git a/src/main/codex-usage/scanner-incremental-append.test.ts b/src/main/codex-usage/scanner-incremental-append.test.ts new file mode 100644 index 00000000000..f4abec1f1c4 --- /dev/null +++ b/src/main/codex-usage/scanner-incremental-append.test.ts @@ -0,0 +1,854 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { tmpdir } from 'node:os' +import type * as NodeOs from 'node:os' +import type * as NodeFs from 'node:fs' +import { join } from 'node:path' + +const { getPathMock, homedirMock, streamReads, onStreamOpen } = vi.hoisted(() => { + const streamReads: { path: string; bytes: number; start: number; bounded: boolean }[] = [] + // Seam for mutating the tree mid-scan, between two files' parse reads. + const onStreamOpen: { current: ((path: string, bounded: boolean) => void) | null } = { + current: null + } + return { + getPathMock: vi.fn<(name: string) => string>(), + homedirMock: vi.fn<() => string>(), + streamReads, + onStreamOpen + } +}) + +vi.mock('electron', () => ({ + app: { + getPath: getPathMock + } +})) + +vi.mock('node:os', async () => { + const actual = await vi.importActual('node:os') + return { + ...actual, + homedir: homedirMock + } +}) + +// The perf oracle: every byte the scanner streams out of a session file. +vi.mock('node:fs', async () => { + const actual = await vi.importActual('node:fs') + return { + ...actual, + createReadStream: ( + path: Parameters[0], + options?: Parameters[1] + ) => { + const filePath = String(path) + const range = typeof options === 'object' && options !== null ? options : {} + const start = range.start ?? 0 + const size = actual.statSync(filePath).size + const stop = range.end === undefined ? size : Math.min(size, range.end + 1) + streamReads.push({ + path: filePath, + bytes: Math.max(0, stop - start), + start, + bounded: range.end !== undefined + }) + onStreamOpen.current?.(filePath, range.end !== undefined) + return actual.createReadStream(path, options) + } + } +}) + +import { mkdirSync, mkdtempSync, rmSync, statSync, writeFileSync, appendFileSync } from 'node:fs' +import { scanCodexUsageFiles } from './scanner' +import type { CodexUsagePersistedFile } from './types' + +/** Mirrors BOUNDARY_WINDOW_BYTES in codex-rollout-resume-state.ts. */ +const BOUNDARY_WINDOW_BYTES = 4096 + +/** Enough records (~377 B each) to put a prefix past MIN_RESUMABLE_PREFIX_BYTES. + * A shorter rollout is always reparsed whole, so a test meaning to exercise the + * resume path has to clear the floor or it silently stops testing anything. */ +const RESUMABLE_RECORDS = 40 + +const originalCodexHome = process.env.CODEX_HOME +let fakeHomeDir: string +let userDataDir: string +let sessionsDir: string +let previousUserDataPath: string | undefined + +function usageRecord(timestamp: string, inputTokens: number, totalInputTokens: number): string { + return `${JSON.stringify({ + timestamp, + type: 'event_msg', + payload: { + type: 'token_count', + info: { + model: 'gpt-5-codex', + last_token_usage: { + input_tokens: inputTokens, + cached_input_tokens: 0, + output_tokens: 0, + reasoning_output_tokens: 0, + total_tokens: inputTokens + }, + total_token_usage: { + input_tokens: totalInputTokens, + cached_input_tokens: 0, + output_tokens: 0, + reasoning_output_tokens: 0, + total_tokens: totalInputTokens + } + } + } + })}\n` +} + +function totalOnlyUsageRecord(timestamp: string, totalInputTokens: number): string { + return `${JSON.stringify({ + timestamp, + type: 'event_msg', + payload: { + type: 'token_count', + info: { + model: 'gpt-5-codex', + total_token_usage: { + input_tokens: totalInputTokens, + cached_input_tokens: 0, + output_tokens: 0, + reasoning_output_tokens: 0, + total_tokens: totalInputTokens + } + } + } + })}\n` +} + +function sessionMeta(id: string): string { + return `${JSON.stringify({ + type: 'session_meta', + payload: { id, cwd: join(fakeHomeDir, 'repo') } + })}\n` +} + +/** Records numbered [from, to), each worth one token, cumulative totals. */ +function usageRecordRange(from: number, to: number): string { + let out = '' + for (let index = from; index < to; index++) { + const minute = String(index % 60).padStart(2, '0') + // Midday UTC keeps the derived local day stable across test-runner zones. + const hour = String(12 + (Math.floor(index / 60) % 4)).padStart(2, '0') + out += usageRecord(`2026-05-26T${hour}:${minute}:00.000Z`, 1, index + 1) + } + return out +} + +function bytesReadFor(filePath: string): number { + return streamReads + .filter((entry) => entry.path === filePath) + .reduce((total, entry) => total + entry.bytes, 0) +} + +/** Offsets at which this file's parse reads opened, in order. Bounded reads are + * digest windows; an unbounded one at 0 is a full reparse and at the recorded + * offset is a resume, so this says exactly which path a scan took. */ +function parseReadOffsets(filePath: string): number[] { + return streamReads + .filter((entry) => entry.path === filePath && !entry.bounded) + .map((entry) => entry.start) +} + +function reparsedFromStart(filePath: string): boolean { + return parseReadOffsets(filePath).includes(0) +} + +function recordedResumeOffset( + files: { path: string; parseResumeState?: { parsedBytes: number } | null }[], + filePath: string +): number { + return files.find((file) => file.path === filePath)?.parseResumeState?.parsedBytes ?? 0 +} + +/** Which session each record was attributed to. Totals can be identical across + * a misattribution, so this is the oracle for anything context-related. */ +function eventCountsBySession(sessions: { sessionId: string; eventCount: number }[]): unknown[] { + return sessions.map((session) => [session.sessionId, session.eventCount]).sort() +} + +function totalTokens(aggregates: { totalTokens: number }[]): number { + return aggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) +} + +beforeEach(() => { + delete process.env.CODEX_HOME + fakeHomeDir = mkdtempSync(join(tmpdir(), 'orca-codex-incremental-home-')) + userDataDir = mkdtempSync(join(tmpdir(), 'orca-codex-incremental-user-data-')) + previousUserDataPath = process.env.ORCA_USER_DATA_PATH + process.env.ORCA_USER_DATA_PATH = userDataDir + homedirMock.mockReturnValue(fakeHomeDir) + getPathMock.mockImplementation((name: string) => { + if (name === 'userData') { + return userDataDir + } + throw new Error(`unexpected app.getPath(${name})`) + }) + sessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + mkdirSync(sessionsDir, { recursive: true }) + streamReads.length = 0 + onStreamOpen.current = null +}) + +afterEach(() => { + rmSync(fakeHomeDir, { recursive: true, force: true }) + rmSync(userDataDir, { recursive: true, force: true }) + if (originalCodexHome === undefined) { + delete process.env.CODEX_HOME + } else { + process.env.CODEX_HOME = originalCodexHome + } + if (previousUserDataPath === undefined) { + delete process.env.ORCA_USER_DATA_PATH + } else { + process.env.ORCA_USER_DATA_PATH = previousUserDataPath + } + vi.clearAllMocks() +}) + +describe('scanCodexUsageFiles incremental append', () => { + it('re-reads only the appended bytes when a rollout grows', async () => { + const rolloutPath = join(sessionsDir, 'rollout-grow.jsonl') + writeFileSync(rolloutPath, `${sessionMeta('session-grow')}${usageRecordRange(0, 200)}`, 'utf-8') + const sizeBeforeAppend = statSync(rolloutPath).size + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(200) + expect(bytesReadFor(rolloutPath)).toBeGreaterThanOrEqual(sizeBeforeAppend) + + streamReads.length = 0 + appendFileSync(rolloutPath, usageRecordRange(200, 202), 'utf-8') + const appendedBytes = statSync(rolloutPath).size - sizeBeforeAppend + + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(totalTokens(second.dailyAggregates)).toBe(202) + expect(second.sessions[0]?.eventCount).toBe(202) + + // The defect: the scanner restarts at byte 0 and re-reads the whole file. + // The fix reads the appended bytes plus five bounded windows: a head and a + // boundary window to plan the resume, both again at the point of use so a + // rollout replaced in between cannot be stitched onto, then the moved + // boundary to record the new resume point. + expect(reparsedFromStart(rolloutPath)).toBe(false) + expect(bytesReadFor(rolloutPath)).toBeLessThan(sizeBeforeAppend) + expect(bytesReadFor(rolloutPath)).toBeLessThanOrEqual(appendedBytes + 5 * BOUNDARY_WINDOW_BYTES) + }) + + it('carries cumulative token totals across the resume boundary', async () => { + const rolloutPath = join(sessionsDir, 'rollout-cumulative.jsonl') + writeFileSync( + rolloutPath, + [ + sessionMeta('session-cumulative'), + // Padding to clear the resumable floor; each record is worth one token + // and leaves the running total at RESUMABLE_RECORDS. + usageRecordRange(0, RESUMABLE_RECORDS), + totalOnlyUsageRecord('2026-05-26T13:00:00.000Z', RESUMABLE_RECORDS + 100), + totalOnlyUsageRecord('2026-05-26T13:01:00.000Z', RESUMABLE_RECORDS + 250) + ].join(''), + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS + 250) + const resumeOffset = recordedResumeOffset(first.processedFiles, rolloutPath) + + // Only the running total is on the wire, so the appended record's delta + // depends entirely on the totals carried out of the previous scan. + appendFileSync( + rolloutPath, + totalOnlyUsageRecord('2026-05-26T13:02:00.000Z', RESUMABLE_RECORDS + 400), + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(parseReadOffsets(rolloutPath)).toEqual([resumeOffset]) + const fromScratch = await scanCodexUsageFiles([], []) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 400) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + }) + + it('still reuses an untouched rollout without reading it', async () => { + const rolloutPath = join(sessionsDir, 'rollout-idle.jsonl') + writeFileSync(rolloutPath, `${sessionMeta('session-idle')}${usageRecordRange(0, 50)}`, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + streamReads.length = 0 + + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(bytesReadFor(rolloutPath)).toBe(0) + expect(totalTokens(second.dailyAggregates)).toBe(50) + }) + + it('tracks byte offsets through CRLF line endings', async () => { + const rolloutPath = join(sessionsDir, 'rollout-crlf.jsonl') + const toCrlf = (text: string): string => text.replaceAll('\n', '\r\n') + writeFileSync( + rolloutPath, + toCrlf(`${sessionMeta('session-crlf')}${usageRecordRange(0, RESUMABLE_RECORDS)}`), + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + const resumeOffset = recordedResumeOffset(first.processedFiles, rolloutPath) + + appendFileSync( + rolloutPath, + toCrlf(usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 3)), + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + // A one-byte-per-line drift would put this offset inside a record. + expect(parseReadOffsets(rolloutPath)).toEqual([resumeOffset]) + const fromScratch = await scanCodexUsageFiles([], []) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 3) + expect(second.sessions).toEqual(fromScratch.sessions) + }) + + it('matches a full rescan after repeated appends', async () => { + const rolloutPath = join(sessionsDir, 'rollout-chatty.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-chatty')}${usageRecordRange(0, RESUMABLE_RECORDS)}`, + 'utf-8' + ) + + let processedFiles: CodexUsagePersistedFile[] = [] + let scanned = await scanCodexUsageFiles([], processedFiles) + processedFiles = scanned.processedFiles + + for (let round = 1; round <= 5; round++) { + const from = RESUMABLE_RECORDS + (round - 1) * 10 + appendFileSync(rolloutPath, usageRecordRange(from, from + 10), 'utf-8') + streamReads.length = 0 + scanned = await scanCodexUsageFiles([], processedFiles) + // Every round after the first has to resume, not restart. + expect(reparsedFromStart(rolloutPath)).toBe(false) + processedFiles = scanned.processedFiles + } + + const fromScratch = await scanCodexUsageFiles([], []) + expect(totalTokens(scanned.dailyAggregates)).toBe(RESUMABLE_RECORDS + 50) + expect(scanned.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(scanned.sessions).toEqual(fromScratch.sessions) + }) + + it('falls back to a full reparse when a rollout is truncated', async () => { + const rolloutPath = join(sessionsDir, 'rollout-truncated.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-truncated')}${usageRecordRange(0, RESUMABLE_RECORDS)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + // Without a recorded resume point the fallback below would be trivial. + expect(recordedResumeOffset(first.processedFiles, rolloutPath)).toBeGreaterThan(0) + + writeFileSync(rolloutPath, `${sessionMeta('session-truncated')}${usageRecordRange(0, 5)}`) + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(totalTokens(second.dailyAggregates)).toBe(5) + }) + + // The point-of-use re-check runs just before the parse read opens, so a + // rollout truncated in the gap still resumes into a file that no longer + // reaches the offset: the stream yields nothing and the whole pre-truncation + // history survives the merge as this scan's answer. + it('reparses from the start when a rollout shrinks during its parse read', async () => { + const rolloutPath = join(sessionsDir, 'rollout-shrinks-mid-read.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-shrinker')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + const resumeOffset = recordedResumeOffset(first.processedFiles, rolloutPath) + + appendFileSync(rolloutPath, usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2), 'utf-8') + const truncated = `${sessionMeta('session-shrinker')}${usageRecordRange(0, 5)}` + onStreamOpen.current = (path, bounded) => { + // Bounded reads are the digest windows; the unbounded one is the parse + // read, which opens after every check this scan is going to make. + if (path === rolloutPath && !bounded) { + onStreamOpen.current = null + writeFileSync(path, truncated, 'utf-8') + } + } + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(onStreamOpen.current).toBeNull() + // Resumed at the recorded offset, then restarted: both halves are the point. + expect(parseReadOffsets(rolloutPath)).toEqual([resumeOffset, 0]) + expect(totalTokens(second.dailyAggregates)).toBe(5) + expect(second.sessions[0]?.eventCount).toBe(5) + // Nothing resumable may survive either: the recorded prefix is gone. + const third = await scanCodexUsageFiles([], second.processedFiles) + expect(totalTokens(third.dailyAggregates)).toBe(5) + }) + + // The other direction, and the worse half: a replacement *longer* than the + // recorded offset reads full windows, so no short read can fire. The cached + // context — session id, cwd, model, running totals — is stitched onto an + // unrelated file's records, running cumulative-delta arithmetic across two + // files that have nothing to do with each other. + // + // Token totals and daily aggregates come out byte-identical to a cold scan + // here: the stale prefix contributes exactly as many events as the resumed + // read skips. Attribution is the only surviving signal, so the oracle is the + // session shape — a totals-based one is provably blind to this. + it('reparses from the start when a rollout is replaced by a larger file mid-scan', async () => { + const driverPath = join(sessionsDir, 'aaaa-driver.jsonl') + const targetPath = join(sessionsDir, 'zzzz-grower.jsonl') + writeFileSync(driverPath, `${sessionMeta('session-driver')}${usageRecordRange(120, 123)}`) + writeFileSync( + targetPath, + `${sessionMeta('session-grower')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + ) + + const first = await scanCodexUsageFiles([], []) + const resumeOffset = + first.processedFiles.find((file) => file.path === targetPath)?.parseResumeState + ?.parsedBytes ?? 0 + + appendFileSync(driverPath, usageRecordRange(123, 124), 'utf-8') + appendFileSync(targetPath, usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2), 'utf-8') + const replacement = `${sessionMeta('session-other')}${usageRecordRange(200, 260)}` + // Past the recorded offset, so every digest window still reads its full size. + expect(Buffer.byteLength(replacement)).toBeGreaterThan(resumeOffset) + onStreamOpen.current = (path, bounded) => { + // Lands while the earlier-sorted rollout is being parsed, which is after + // the scanner's discovery loop verified every file's prefix. + if (path === driverPath && !bounded) { + onStreamOpen.current = null + writeFileSync(targetPath, replacement, 'utf-8') + } + } + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(onStreamOpen.current).toBeNull() + // The scan planned to resume here, and the point-of-use check is what sends + // it back to byte 0 before a single suffix byte is read. + expect(resumeOffset).toBeGreaterThan(0) + expect(parseReadOffsets(targetPath)).toEqual([0]) + const fromScratch = await scanCodexUsageFiles([], []) + + // Stitching leaves `session-grower` owning the records of `session-other`. + expect(eventCountsBySession(second.sessions)).toEqual( + eventCountsBySession(fromScratch.sessions) + ) + expect(second.sessions).toEqual(fromScratch.sessions) + }) + + it('falls back to a full reparse when a rollout is rewritten at the same size', async () => { + const rolloutPath = join(sessionsDir, 'rollout-replaced.jsonl') + const original = `${sessionMeta('session-a')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + writeFileSync(rolloutPath, original, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + expect(recordedResumeOffset(first.processedFiles, rolloutPath)).toBeGreaterThan(0) + + // Byte-identical length, different content: only the year changes. + const replacement = original.replaceAll('2026-05-26T', '2027-05-26T') + expect(replacement.length).toBe(original.length) + writeFileSync(rolloutPath, replacement, 'utf-8') + expect(statSync(rolloutPath).size).toBe(original.length) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(second.sessions).toEqual(fromScratch.sessions) + }) + + it('falls back to a full reparse when a rewritten rollout also grows', async () => { + const rolloutPath = join(sessionsDir, 'rollout-rotated.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-a')}${usageRecordRange(0, RESUMABLE_RECORDS)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + expect(recordedResumeOffset(first.processedFiles, rolloutPath)).toBeGreaterThan(0) + + // Rotation: a fresh, longer file lands at the same path. + rmSync(rolloutPath) + writeFileSync( + rolloutPath, + `${sessionMeta('session-b')}${usageRecordRange(0, RESUMABLE_RECORDS + 5).replaceAll('2026-', '2027-')}`, + 'utf-8' + ) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 5) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(second.sessions).toEqual(fromScratch.sessions) + }) + + /** A rollout whose leading records differ but whose trailing records — more + * than a boundary window of them — are byte-identical, at the same length. + * The boundary digest is blind to this by construction. */ + function prefixSwapPair(sessionId: string): { original: string; replacement: string } { + const sharedSuffix = usageRecordRange(20, 40) + expect(sharedSuffix.length).toBeGreaterThan(BOUNDARY_WINDOW_BYTES) + let swappedPrefix = '' + for (let index = 0; index < 20; index++) { + const minute = String(index % 60).padStart(2, '0') + swappedPrefix += usageRecord(`2026-05-26T12:${minute}:00.000Z`, 3, index + 1) + } + const original = `${sessionMeta(sessionId)}${usageRecordRange(0, 20)}${sharedSuffix}` + const replacement = `${sessionMeta(sessionId)}${swappedPrefix}${sharedSuffix}` + expect(replacement.length).toBe(original.length) + return { original, replacement } + } + + // The mirror image of the prefix swap: the head window is byte-identical, so + // only the boundary window is left to notice that trailing records changed. + it('falls back to a full reparse when the records before the offset changed', async () => { + const rolloutPath = join(sessionsDir, 'rollout-tail-swap.jsonl') + const swappedFrom = RESUMABLE_RECORDS + const swappedTo = RESUMABLE_RECORDS + 10 + const sharedHead = `${sessionMeta('session-tail')}${usageRecordRange(0, swappedFrom)}` + expect(sharedHead.length).toBeGreaterThan(BOUNDARY_WINDOW_BYTES) + let heavierTail = '' + for (let index = swappedFrom; index < swappedTo; index++) { + const minute = String(index % 60).padStart(2, '0') + const hour = String(12 + (Math.floor(index / 60) % 4)).padStart(2, '0') + heavierTail += usageRecord(`2026-05-26T${hour}:${minute}:00.000Z`, 3, index + 1) + } + const original = `${sharedHead}${usageRecordRange(swappedFrom, swappedTo)}` + const replacement = `${sharedHead}${heavierTail}` + expect(replacement.length).toBe(original.length) + // Only the bytes inside the boundary window differ, so the head digest is + // blind to this and the boundary digest is the one guard under test. + expect(original.length - sharedHead.length).toBeLessThan(BOUNDARY_WINDOW_BYTES) + writeFileSync(rolloutPath, original, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(swappedTo) + expect(recordedResumeOffset(first.processedFiles, rolloutPath)).toBeGreaterThan(0) + + writeFileSync(rolloutPath, replacement, 'utf-8') + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(totalTokens(second.dailyAggregates)).toBe(swappedFrom + 10 * 3) + }) + + // Rotation: the path is unlinked and recreated. `physicalFileId` cannot carry + // this — ext4 and overlayfs hand the new file the inode the old one freed — + // so the head window is what has to catch it on Linux. + it('falls back to a full reparse when a recreated rollout swapped its prefix', async () => { + const rolloutPath = join(sessionsDir, 'rollout-prefix-swap-rotated.jsonl') + const { original, replacement } = prefixSwapPair('session-prefix-rotated') + writeFileSync(rolloutPath, original, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(40) + + rmSync(rolloutPath) + writeFileSync(rolloutPath, replacement, 'utf-8') + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(totalTokens(second.dailyAggregates)).toBe(80) + }) + + // The same swap written in place. No inode changes on any platform, so the + // head window is the only guard left — this is the case that was missed on + // macOS too, not just on Linux. + it('falls back to a full reparse when a prefix was rewritten in place', async () => { + const rolloutPath = join(sessionsDir, 'rollout-prefix-swap-in-place.jsonl') + const { original, replacement } = prefixSwapPair('session-prefix-in-place') + writeFileSync(rolloutPath, original, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(40) + + const inodeBefore = statSync(rolloutPath).ino + writeFileSync(rolloutPath, replacement, 'utf-8') + // Pins why this test is not a duplicate of the rotation case above. + expect(statSync(rolloutPath).ino).toBe(inodeBefore) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + expect(totalTokens(second.dailyAggregates)).toBe(80) + }) + + // A new session is too short to be worth resuming, so it records no resume + // point and is reparsed whole. Once it grows past the floor it has to start + // resuming, rather than staying on the full-reparse path for the rest of its + // life because the first scan left nothing behind. + it('starts resuming once the prefix grows past the resumable floor', async () => { + const rolloutPath = join(sessionsDir, 'rollout-crosses-floor.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-crosses')}${usageRecordRange(0, 3)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(3) + expect(first.processedFiles[0]?.parseResumeState).toBeNull() + + streamReads.length = 0 + appendFileSync(rolloutPath, usageRecordRange(3, RESUMABLE_RECORDS), 'utf-8') + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS) + // Nothing to resume from yet, so this scan reads the whole file. + expect(parseReadOffsets(rolloutPath)).toEqual([0]) + const resumeOffset = recordedResumeOffset(second.processedFiles, rolloutPath) + expect(resumeOffset).toBeGreaterThan(0) + + streamReads.length = 0 + appendFileSync(rolloutPath, usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2), 'utf-8') + const third = await scanCodexUsageFiles([], second.processedFiles) + expect(totalTokens(third.dailyAggregates)).toBe(RESUMABLE_RECORDS + 2) + expect(parseReadOffsets(rolloutPath)).toEqual([resumeOffset]) + }) + + it('does not double-count a record completed after a partial trailing line', async () => { + const rolloutPath = join(sessionsDir, 'rollout-partial.jsonl') + const complete = usageRecordRange(0, RESUMABLE_RECORDS) + const pending = usageRecord('2026-05-26T12:59:00.000Z', 1, RESUMABLE_RECORDS + 1) + writeFileSync( + rolloutPath, + `${sessionMeta('session-partial')}${complete}${pending.slice(0, 40)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + + // The writer finishes the line and appends one more record. + writeFileSync( + rolloutPath, + `${sessionMeta('session-partial')}${complete}${pending}${usageRecordRange(RESUMABLE_RECORDS + 1, RESUMABLE_RECORDS + 2)}`, + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + // The partial tail sits past the recorded offset, so this resumes onto it. + expect(parseReadOffsets(rolloutPath)).toEqual([ + recordedResumeOffset(first.processedFiles, rolloutPath) + ]) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 2) + expect(second.sessions[0]?.eventCount).toBe(RESUMABLE_RECORDS + 2) + }) + + // The case above stops at the parser: its tail is truncated JSON, so no event + // comes out of it. A tail that is complete JSON with only the newline missing + // is counted, yet the next scan re-reads it — the resume offset must exclude + // it or the record lands in the totals twice. + it('does not double-count a counted tail whose newline was not yet written', async () => { + const rolloutPath = join(sessionsDir, 'rollout-unflushed-newline.jsonl') + const complete = usageRecordRange(0, RESUMABLE_RECORDS) + const pending = usageRecord('2026-05-26T12:59:00.000Z', 1, RESUMABLE_RECORDS + 1) + writeFileSync( + rolloutPath, + `${sessionMeta('session-unflushed')}${complete}${pending.slice(0, -1)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + // The unterminated line is valid JSON, so it is parsed and counted here. + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS + 1) + expect(first.sessions[0]?.eventCount).toBe(RESUMABLE_RECORDS + 1) + // The prefix clears the resumable floor, so suppressing the resume point is + // the only thing that can force the reparse asserted below. + expect(first.processedFiles[0]?.parseResumeState).toBeNull() + + // The writer flushes the newline and appends one more record. + appendFileSync( + rolloutPath, + `\n${usageRecordRange(RESUMABLE_RECORDS + 1, RESUMABLE_RECORDS + 2)}`, + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(parseReadOffsets(rolloutPath)).toEqual([0]) + const fromScratch = await scanCodexUsageFiles([], []) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 2) + expect(second.sessions[0]?.eventCount).toBe(RESUMABLE_RECORDS + 2) + expect(second.dailyAggregates).toEqual(fromScratch.dailyAggregates) + }) + + it('reads appended bytes only when the append shares the cached mtime', async () => { + const rolloutPath = join(sessionsDir, 'rollout-same-mtime.jsonl') + writeFileSync( + rolloutPath, + `${sessionMeta('session-same-mtime')}${usageRecordRange(0, 40)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(40) + streamReads.length = 0 + + appendFileSync(rolloutPath, usageRecordRange(40, 42), 'utf-8') + // A coarse-mtime filesystem reports the append under the cached mtime. + const coarseMtimeMs = statSync(rolloutPath).mtimeMs + const cached = first.processedFiles.map((file) => + file.path === rolloutPath ? { ...file, mtimeMs: coarseMtimeMs } : file + ) + + const second = await scanCodexUsageFiles([], cached) + expect(totalTokens(second.dailyAggregates)).toBe(42) + expect(second.sessions[0]?.eventCount).toBe(42) + expect(reparsedFromStart(rolloutPath)).toBe(false) + }) + + it('keeps fork ownership when the owning rollout grows incrementally', async () => { + const originalPath = join(sessionsDir, 'aaaa-original.jsonl') + const forkPath = join(sessionsDir, 'zzzz-fork.jsonl') + const copiedPrefix = `${sessionMeta('session-fork')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + writeFileSync(originalPath, copiedPrefix, 'utf-8') + writeFileSync( + forkPath, + `${copiedPrefix}${usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS + 2) + expect(first.processedFiles.find((file) => file.path === originalPath)?.ownedEventKeys).toEqual( + expect.arrayContaining([expect.any(String)]) + ) + + const resumeOffset = recordedResumeOffset(first.processedFiles, originalPath) + appendFileSync( + originalPath, + usageRecordRange(RESUMABLE_RECORDS + 2, RESUMABLE_RECORDS + 4), + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(parseReadOffsets(originalPath)).toEqual([resumeOffset]) + // shared + 2 fork-only + 2 newly appended, each counted exactly once. + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 4) + const originalAfter = second.processedFiles.find((file) => file.path === originalPath) + const forkAfter = second.processedFiles.find((file) => file.path === forkPath) + expect(originalAfter?.ownedEventKeys).toHaveLength(RESUMABLE_RECORDS + 2) + expect(forkAfter?.ownedEventKeys).toHaveLength(2) + expect(forkAfter?.hasDeferredClaims).toBe(true) + }) + + it('keeps a new fork from re-claiming events a resumed rollout still owns', async () => { + const originalPath = join(sessionsDir, 'aaaa-origin.jsonl') + const forkPath = join(sessionsDir, 'zzzz-late-fork.jsonl') + const copiedPrefix = `${sessionMeta('session-late')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + writeFileSync(originalPath, copiedPrefix, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(RESUMABLE_RECORDS) + const resumeOffset = recordedResumeOffset(first.processedFiles, originalPath) + + // The owner grows (resume path) in the same cycle a fork of its prefix appears. + appendFileSync( + originalPath, + usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2), + 'utf-8' + ) + writeFileSync( + forkPath, + `${copiedPrefix}${usageRecordRange(RESUMABLE_RECORDS + 2, RESUMABLE_RECORDS + 3)}`, + 'utf-8' + ) + + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(parseReadOffsets(originalPath)).toEqual([resumeOffset]) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 3) + const forkAfter = second.processedFiles.find((file) => file.path === forkPath) + expect(forkAfter?.ownedEventKeys).toHaveLength(1) + expect(forkAfter?.hasDeferredClaims).toBe(true) + }) + + // Pins why the scanner verifies resume points before it seeds event ownership + // rather than leaving it to the parse. A rollout that fails verification must + // not reserve the keys it used to own: a fork holding those same records is + // the only file left that can count them. + it('lets a fork reclaim records a rewritten rollout can no longer own', async () => { + const originalPath = join(sessionsDir, 'aaaa-rewritten.jsonl') + const forkPath = join(sessionsDir, 'zzzz-inheritor.jsonl') + const sharedPrefix = `${sessionMeta('session-shared')}${usageRecordRange(0, 40)}` + writeFileSync(originalPath, sharedPrefix, 'utf-8') + + const first = await scanCodexUsageFiles([], []) + expect(totalTokens(first.dailyAggregates)).toBe(40) + + // The owner is rewritten into an unrelated session, so its resume point no + // longer verifies; a fork carrying its old records appears the same cycle. + writeFileSync( + originalPath, + `${sessionMeta('session-rewritten')}${usageRecordRange(100, 140)}`, + 'utf-8' + ) + writeFileSync(forkPath, `${sharedPrefix}${usageRecordRange(40, 43)}`, 'utf-8') + + const second = await scanCodexUsageFiles([], first.processedFiles) + const fromScratch = await scanCodexUsageFiles([], []) + expect(eventCountsBySession(second.sessions)).toEqual( + eventCountsBySession(fromScratch.sessions) + ) + expect(totalTokens(second.dailyAggregates)).toBe(83) + }) + + it('still reclaims deferred fork claims after an incremental append', async () => { + const originalPath = join(sessionsDir, 'aaaa-owner.jsonl') + const forkPath = join(sessionsDir, 'zzzz-deferred.jsonl') + const copiedPrefix = `${sessionMeta('session-deferred')}${usageRecordRange(0, RESUMABLE_RECORDS)}` + writeFileSync(originalPath, copiedPrefix, 'utf-8') + writeFileSync( + forkPath, + `${copiedPrefix}${usageRecordRange(RESUMABLE_RECORDS, RESUMABLE_RECORDS + 2)}`, + 'utf-8' + ) + + const first = await scanCodexUsageFiles([], []) + const resumeOffset = recordedResumeOffset(first.processedFiles, forkPath) + // The deferring fork is the one that grows, so its deferred flag has to + // survive the incremental merge or the reclaim below never runs. + appendFileSync( + forkPath, + usageRecordRange(RESUMABLE_RECORDS + 2, RESUMABLE_RECORDS + 4), + 'utf-8' + ) + streamReads.length = 0 + const second = await scanCodexUsageFiles([], first.processedFiles) + expect(parseReadOffsets(forkPath)).toEqual([resumeOffset]) + expect(totalTokens(second.dailyAggregates)).toBe(RESUMABLE_RECORDS + 4) + expect(second.processedFiles.find((file) => file.path === forkPath)?.hasDeferredClaims).toBe( + true + ) + + rmSync(originalPath) + const third = await scanCodexUsageFiles([], second.processedFiles) + expect(third.processedFiles).toHaveLength(1) + expect(third.processedFiles[0]?.ownedEventKeys).toHaveLength(RESUMABLE_RECORDS + 4) + expect(totalTokens(third.dailyAggregates)).toBe(RESUMABLE_RECORDS + 4) + }) +}) diff --git a/src/main/codex-usage/scanner-paths.test.ts b/src/main/codex-usage/scanner-paths.test.ts index 2a7cc7ccd41..2f14ee0ad42 100644 --- a/src/main/codex-usage/scanner-paths.test.ts +++ b/src/main/codex-usage/scanner-paths.test.ts @@ -382,6 +382,25 @@ describe('listCodexSessionFiles', () => { expect( result.dailyAggregates.reduce((total, aggregate) => total + aggregate.eventCount, 0) ).toBe(3) + + // A suffix-only parse must not leave a resume offset behind: once the + // bridge markers are gone, a later append has to reparse the whole file so + // the previously skipped prefix is counted. + rmSync(runtimeBridgeMarkerDir, { recursive: true, force: true }) + rmSync(runtimeSessionPath) + writeFileSync( + systemSessionPath, + `${copiedPrefix}${usageRecord('2026-05-26T12:01:00.000Z', 3, 13)}${usageRecord('2026-05-26T12:03:00.000Z', 4, 17)}` + ) + + const afterBridge = await scanCodexUsageFiles([], result.processedFiles) + + expect( + afterBridge.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(17) + expect( + afterBridge.dailyAggregates.reduce((total, aggregate) => total + aggregate.eventCount, 0) + ).toBe(3) }) it('counts token events copied into forked rollout files exactly once', async () => { @@ -594,6 +613,153 @@ describe('listCodexSessionFiles', () => { ).toBe(25) }) + // Bridge markers can appear on a source file that already has a resume state: + // the copy is made after a scan, so the scanner sees a non-legacy cache and a + // legacy suffix offset at once. Resuming would extend the cached full-history + // projection instead of restarting as a suffix-only parse, which loses the + // total-only baseline the suffix depends on and recounts the copied records. + it('reparses in full when bridge markers appear on an already-resumable source', async () => { + const runtimeSessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + const runtimeBridgeMarkerDir = join( + userDataDir, + 'codex-runtime-home', + 'home', + '.orca-session-copies' + ) + const systemSessionsDir = join(fakeHomeDir, '.codex', 'sessions') + mkdirSync(runtimeSessionsDir, { recursive: true }) + mkdirSync(systemSessionsDir, { recursive: true }) + const systemSessionPath = join(systemSessionsDir, 'system.jsonl') + const runtimeSessionPath = join(runtimeSessionsDir, 'system.jsonl') + const meta = `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'legacy-session', cwd: join(fakeHomeDir, 'repo') } + })}\n` + // The prefix has to clear MIN_RESUMABLE_PREFIX_BYTES or scan 1 records no + // resume point and the transition under test never arises. + let padding = '' + for (let index = 0; index < 40; index++) { + const minute = String(index).padStart(2, '0') + padding += usageRecord(`2026-05-26T11:${minute}:00.000Z`, 1, index + 1) + } + const scannedPrefix = `${meta}${padding}${usageRecord('2026-05-26T12:00:00.000Z', 10, 50)}` + writeFileSync(systemSessionPath, scannedPrefix, 'utf-8') + + // No markers yet, so this scan records a plain incremental resume point. + const first = await scanCodexUsageFiles([], []) + expect( + first.processedFiles.find((file) => file.path === systemSessionPath)?.parseResumeState + ?.parsedBytes + ).toBe(Buffer.byteLength(scannedPrefix)) + + // The source grows, then the legacy copy is taken: the copied prefix now + // reaches past the recorded resume offset. + const copiedPrefix = `${scannedPrefix}${usageRecord('2026-05-26T12:01:00.000Z', 3, 53)}` + writeFileSync(systemSessionPath, copiedPrefix, 'utf-8') + writeFileSync(runtimeSessionPath, copiedPrefix, 'utf-8') + mkdirSync(runtimeBridgeMarkerDir, { recursive: true }) + const sourceStat = lstatSync(systemSessionPath) + const targetStat = lstatSync(runtimeSessionPath) + writeFileSync( + join(runtimeBridgeMarkerDir, 'system.jsonl.json'), + `${JSON.stringify({ + sourcePath: systemSessionPath, + sourceSize: sourceStat.size, + sourceMtimeMs: sourceStat.mtimeMs, + targetSize: targetStat.size, + targetMtimeMs: targetStat.mtimeMs + })}\n`, + 'utf-8' + ) + writeFileSync( + systemSessionPath, + [ + copiedPrefix, + totalOnlyUsageRecord('2026-05-26T12:02:00.000Z', 70), + totalOnlyUsageRecord('2026-05-26T12:03:00.000Z', 74) + ].join('') + ) + writeFileSync( + runtimeSessionPath, + `${copiedPrefix}${usageRecord('2026-05-26T12:04:00.000Z', 5, 58)}` + ) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const cold = await scanCodexUsageFiles([], []) + + // 40 padding + 10 + 3 copied prefix, 5 runtime-only, and the source suffix + // contributing 74 - 70 once its leading total-only record reads as baseline. + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(62) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.eventCount, 0) + ).toBe(44) + expect(second.dailyAggregates).toEqual(cold.dailyAggregates) + }) + + // The reuse gate has its own legacy check, separate from the resume gate. A + // cached entry can predate the bridge marker while the source file itself is + // untouched, so (size, mtime) still match and nothing else would stop the + // scan serving a full-history projection for a file that is now parsed + // suffix-only — double-counting the copied prefix against the managed copy. + it('does not reuse a pre-bridge cache once the source became suffix-only', async () => { + const runtimeSessionsDir = join(userDataDir, 'codex-runtime-home', 'home', 'sessions') + const markerDir = join(userDataDir, 'codex-runtime-home', 'home', '.orca-session-copies') + const systemSessionsDir = join(fakeHomeDir, '.codex', 'sessions') + mkdirSync(runtimeSessionsDir, { recursive: true }) + mkdirSync(systemSessionsDir, { recursive: true }) + const systemSessionPath = join(systemSessionsDir, 'system.jsonl') + const runtimeSessionPath = join(runtimeSessionsDir, 'system.jsonl') + const meta = `${JSON.stringify({ + type: 'session_meta', + payload: { id: 'legacy-session', cwd: join(fakeHomeDir, 'repo') } + })}\n` + const copiedPrefix = `${meta}${usageRecord('2026-05-26T12:00:00.000Z', 10)}` + // A total-only tail is what separates the two readings: parsed as a suffix + // it is a baseline worth nothing, carried in a full projection it is a + // delta worth 3. + writeFileSync( + systemSessionPath, + `${copiedPrefix}${totalOnlyUsageRecord('2026-05-26T12:01:00.000Z', 13)}`, + 'utf-8' + ) + + // No marker directory yet, so this is an ordinary full parse. + const first = await scanCodexUsageFiles([], []) + const cachedStat = lstatSync(systemSessionPath) + + // The bridge marker lands afterwards, recording the source as it stood when + // the copy was taken. The source file is not touched. + mkdirSync(markerDir, { recursive: true }) + writeFileSync( + runtimeSessionPath, + `${copiedPrefix}${usageRecord('2026-05-26T12:02:00.000Z', 5, 15)}`, + 'utf-8' + ) + writeFileSync( + join(markerDir, 'system.jsonl.json'), + `${JSON.stringify({ + sourcePath: systemSessionPath, + sourceSize: Buffer.byteLength(copiedPrefix), + sourceMtimeMs: cachedStat.mtimeMs - 5000, + targetSize: Buffer.byteLength(copiedPrefix), + targetMtimeMs: cachedStat.mtimeMs - 5000 + })}\n`, + 'utf-8' + ) + // The reuse gate's own check is the only thing left: the stat still matches. + expect(lstatSync(systemSessionPath).size).toBe(cachedStat.size) + expect(lstatSync(systemSessionPath).mtimeMs).toBe(cachedStat.mtimeMs) + + const second = await scanCodexUsageFiles([], first.processedFiles) + const cold = await scanCodexUsageFiles([], []) + expect( + second.dailyAggregates.reduce((total, aggregate) => total + aggregate.totalTokens, 0) + ).toBe(15) + expect(second.dailyAggregates).toEqual(cold.dailyAggregates) + }) + 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 0d5c1eba526..998025f8a7c 100644 --- a/src/main/codex-usage/scanner.ts +++ b/src/main/codex-usage/scanner.ts @@ -1,117 +1,28 @@ -import { basename } from 'node:path' -import { createReadStream } from 'node:fs' -import { stat } from 'node:fs/promises' -import { createInterface } from 'node:readline' -import { createUsageEventAggregation } from '../usage/usage-event-aggregation' -import { - createUsageWorktreeResolver, - type UsageWorktreeResolver -} from '../usage/usage-worktree-resolver' +import { createUsageWorktreeResolver } from '../usage/usage-worktree-resolver' import { getLegacySourceSkipBytesByPath, listCodexSessionFiles, yieldToEventLoop } from './codex-session-file-discovery' -import { - attributeCodexUsageEvent, - type CodexUsageWorktreeRef -} from './codex-usage-event-attribution' -import { parseCodexUsageRecord, type CodexUsageParseContext } from './codex-usage-record-parser' +import type { CodexUsageWorktreeRef } from './codex-usage-event-attribution' +import { codexUsageAggregation } from './codex-usage-aggregation' +import { getProcessedFileInfo, parseCodexUsageFile } from './codex-rollout-file-parse' +import { resolveCodexRolloutResume } from './codex-rollout-resume-state' import type { - CodexUsageAttributedEvent, CodexUsageDailyAggregate, + CodexUsageParseResumeState, CodexUsagePersistedFile, - CodexUsageProcessedFile, CodexUsageSession } from './types' const YIELD_EVERY_FILES = 10 -export async function getProcessedFileInfo(filePath: string): Promise { - const fileStat = await stat(filePath) - return { - path: filePath, - mtimeMs: fileStat.mtimeMs, - size: fileStat.size - } -} - -type CodexUsageMetric = { hasInferredPricing: boolean } - -const codexUsageAggregation = createUsageEventAggregation< - CodexUsageAttributedEvent, - CodexUsageMetric ->({ - metric: { - empty: () => ({ hasInferredPricing: false }), - fromEvent: (event) => ({ hasInferredPricing: event.hasInferredPricing }), - fold: (target, source) => { - target.hasInferredPricing ||= source.hasInferredPricing - } - }, - cloneSessionForMerge: (session) => ({ - ...session, - locationBreakdown: session.locationBreakdown.map((entry) => ({ ...entry })), - modelBreakdown: session.modelBreakdown.map((entry) => ({ ...entry })), - locationModelBreakdown: session.locationModelBreakdown.map((entry) => ({ ...entry })) - }) -}) - const { finalizeSessions, mergeSessions, mergeDailyAggregates, sortDailyAggregates } = codexUsageAggregation -export async function parseCodexUsageFile( - filePath: string, - resolveWorktree: UsageWorktreeResolver, - options: { skipInitialBytes?: number; claimEventKey?: (eventKey: string) => boolean } = {} -): Promise { - const processedFile = await getProcessedFileInfo(filePath) - const lines = createInterface({ - input: createReadStream(filePath, { - encoding: 'utf-8', - start: options.skipInitialBytes ?? 0 - }), - crlfDelay: Infinity - }) - const events: CodexUsageAttributedEvent[] = [] - const context: CodexUsageParseContext = { - sessionId: basename(filePath, '.jsonl'), - sessionCwd: null, - currentCwd: null, - currentModel: null, - previousTotals: null, - // Why: suffix-only legacy copy parsing lacks the copied prefix context. A - // leading total-only snapshot is a baseline, not the suffix's billable delta. - totalOnlyBaselinePending: (options.skipInitialBytes ?? 0) > 0 - } - - const ownedEventKeys = new Set() - let hasDeferredClaims = false - for await (const line of lines) { - const parsed = parseCodexUsageRecord(line, context) - if (!parsed) { - continue - } - // Why: fork/resume rollouts start with a copied prefix of the parent file. - // 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) - const attributed = await attributeCodexUsageEvent(parsed, resolveWorktree) - if (attributed) { - events.push(attributed) - } - } - - return { - ...processedFile, - ...codexUsageAggregation.aggregate(events), - ownedEventKeys: [...ownedEventKeys], - hasDeferredClaims - } +type CodexRolloutResumePlan = { + state: CodexUsageParseResumeState + previous: CodexUsagePersistedFile } export async function scanCodexUsageFiles( @@ -141,6 +52,7 @@ export async function scanCodexUsageFiles( ) const reusedByPath = new Map() + const resumeByPath = new Map() const pathsToParse: string[] = [] for (const [index, filePath] of files.entries()) { const legacySourceSkipBytes = legacySourceSkipBytesByPath.get(filePath) ?? 0 @@ -159,6 +71,15 @@ export async function scanCodexUsageFiles( if (canReuse) { reusedByPath.set(filePath, previous) } else { + // Why: rollouts are append-only and grow all day, so re-reading each one + // from byte 0 dominated scans (#20940). A reclaim or a legacy suffix + // offset still needs the whole file, so neither may resume. + if (!mustReclaimDeferred && legacySourceSkipBytes === 0 && previous) { + const state = await resolveCodexRolloutResume(filePath, previous) + if (state) { + resumeByPath.set(filePath, { state, previous }) + } + } pathsToParse.push(filePath) } if ((index + 1) % YIELD_EVERY_FILES === 0) { @@ -169,13 +90,14 @@ export async function scanCodexUsageFiles( // Why: resuming or forking a Codex session copies the parent rollout's // token_count records into a new file, so per-file parsing re-counts the // whole copied history once per descendant (#8006). Cross-file ownership - // counts each record for exactly one file; cached files keep the claims - // they persisted, and new files claim in sorted-path order so rescans stay - // deterministic. + // counts each record for exactly one file; cached and resumed files keep the + // claims they persisted, and the rest claim in sorted-path order so rescans + // stay deterministic. const eventOwnerByKey = new Map() - for (const [filePath, previous] of reusedByPath) { - for (const eventKey of previous.ownedEventKeys) { - // First cached claim wins so conflicting projections stay deterministic. + for (const filePath of files) { + const retained = reusedByPath.get(filePath) ?? resumeByPath.get(filePath)?.previous + for (const eventKey of retained?.ownedEventKeys ?? []) { + // First retained claim wins so conflicting projections stay deterministic. if (!eventOwnerByKey.has(eventKey)) { eventOwnerByKey.set(eventKey, filePath) } @@ -185,7 +107,8 @@ export async function scanCodexUsageFiles( const parsedByPath = new Map() for (const [index, filePath] of pathsToParse.entries()) { const processed = await parseCodexUsageFile(filePath, resolveWorktree, { - skipInitialBytes: legacySourceSkipBytesByPath.get(filePath) ?? 0, + legacySourceSkipBytes: legacySourceSkipBytesByPath.get(filePath) ?? 0, + resume: resumeByPath.get(filePath), claimEventKey: (eventKey) => { const owner = eventOwnerByKey.get(eventKey) if (owner !== undefined && owner !== filePath) { diff --git a/src/main/codex-usage/store-automation-usage.test.ts b/src/main/codex-usage/store-automation-usage.test.ts index 187799c17a4..73cd5ce9ccc 100644 --- a/src/main/codex-usage/store-automation-usage.test.ts +++ b/src/main/codex-usage/store-automation-usage.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it, vi } from 'vitest' import type { CodexUsagePersistedState } from './types' -import { createStoreWithState, setupCodexUsageStoreEnv } from './store-test-harness' +import { scanCodexUsageFiles } from './scanner' +import { + createStoreWithState, + createWorktreeUsageSession, + setupCodexUsageStoreEnv +} from './store-test-harness' const { getPathMock } = vi.hoisted(() => ({ getPathMock: vi.fn(() => '/tmp/orca-test-userdata') @@ -28,70 +33,7 @@ describe('CodexUsageStore', () => { lastScanCompletedAt: 2, lastScanError: null }, - sessions: [ - { - sessionId: 'session-1', - firstTimestamp: '2026-04-10T15:00:00.000Z', - lastTimestamp: '2026-04-10T15:05:00.000Z', - primaryModel: 'gpt-5', - hasMixedModels: false, - primaryProjectLabel: 'Repo', - hasMixedLocations: false, - primaryWorktreeId: worktreeId, - primaryRepoId: 'repo-1', - eventCount: 1, - totalInputTokens: 1000, - totalCachedInputTokens: 400, - totalOutputTokens: 250, - totalReasoningOutputTokens: 100, - totalTokens: 1250, - hasInferredPricing: false, - locationBreakdown: [ - { - locationKey: `worktree:${worktreeId}`, - projectLabel: 'Repo', - repoId: 'repo-1', - worktreeId, - eventCount: 1, - inputTokens: 1000, - cachedInputTokens: 400, - outputTokens: 250, - reasoningOutputTokens: 100, - totalTokens: 1250, - hasInferredPricing: false - } - ], - modelBreakdown: [ - { - modelKey: 'gpt-5', - modelLabel: 'gpt-5', - eventCount: 1, - inputTokens: 1000, - cachedInputTokens: 400, - outputTokens: 250, - reasoningOutputTokens: 100, - totalTokens: 1250, - hasInferredPricing: false - } - ], - locationModelBreakdown: [ - { - locationKey: `worktree:${worktreeId}`, - modelKey: 'gpt-5', - modelLabel: 'gpt-5', - repoId: 'repo-1', - worktreeId, - eventCount: 1, - inputTokens: 1000, - cachedInputTokens: 400, - outputTokens: 250, - reasoningOutputTokens: 100, - totalTokens: 1250, - hasInferredPricing: false - } - ] - } - ] + sessions: [createWorktreeUsageSession(worktreeId)] }) const refreshMock = vi.fn().mockResolvedValue({ enabled: true, @@ -126,4 +68,97 @@ describe('CodexUsageStore', () => { expect(refreshMock).toHaveBeenCalledWith(false) }) + it('forces one scan per run and stops re-forcing after a failed attempt', async () => { + const completedAt = new Date('2026-04-10T15:06:00.000Z').getTime() + const scanError = 'EMFILE: too many open files' + const failedScanState = (lastScanStartedAt: number) => ({ + enabled: true, + lastScanStartedAt, + lastScanCompletedAt: completedAt - 60_000, + lastScanError: scanError + }) + const scanStateResult = { + enabled: true, + isScanning: false, + lastScanStartedAt: completedAt - 60_000, + lastScanCompletedAt: completedAt - 60_000, + lastScanError: scanError, + hasAnyCodexData: false + } + const request = { + worktreeId: 'repo-1::/workspace/repo', + terminalSessionId: 'tab-1', + startedAt: completedAt - 120_000, + completedAt + } + + const beforeAttempt = createStoreWithState({ + scanState: failedScanState(completedAt - 60_000) + }) + const beforeRefresh = vi.spyOn(beforeAttempt, 'refresh').mockResolvedValue(scanStateResult) + await beforeAttempt.getAutomationRunUsage(request) + + expect(beforeRefresh).toHaveBeenCalledWith(true) + + // That forced scan failed: it recorded an attempt but no completion. Later + // lookups must not keep forcing a full rescan of all Codex history. + const afterAttempt = createStoreWithState({ scanState: failedScanState(completedAt + 1000) }) + const afterRefresh = vi.spyOn(afterAttempt, 'refresh').mockResolvedValue(scanStateResult) + const usage = await afterAttempt.getAutomationRunUsage(request) + + expect(afterRefresh).toHaveBeenCalledWith(false) + expect(usage.unavailableReason).toBe('scan_failed') + }) + + it('joins a scan that is already in flight when the run finished before it started', async () => { + const worktreeId = 'repo-1::/workspace/repo' + const store = createStoreWithState({ + scanState: { + enabled: true, + lastScanStartedAt: null, + lastScanCompletedAt: null, + lastScanError: null + } + }) + // Prime the worktree fingerprint so an unforced refresh can return early. + await store.refresh(true) + + const completedAt = Date.now() + 10_000 + vi.setSystemTime(new Date(completedAt + 1_000)) + + let startScan = () => {} + let finishScan = () => {} + const scanStarted = new Promise((resolve) => { + startScan = resolve + }) + const scanFinished = new Promise((resolve) => { + finishScan = resolve + }) + vi.mocked(scanCodexUsageFiles).mockImplementationOnce(async () => { + startScan() + await scanFinished + return { + processedFiles: [], + sessions: [createWorktreeUsageSession(worktreeId)], + dailyAggregates: [] + } + }) + + const inFlight = store.refresh(true) + await scanStarted + + const usage = store.getAutomationRunUsage({ + worktreeId, + terminalSessionId: 'session-1', + startedAt: completedAt - 60_000, + completedAt + }) + finishScan() + await inFlight + + // The in-flight scan's start time is not a finished attempt, so the lookup + // forces and rides that scan instead of reading a pre-run cache. + expect((await usage).status).toBe('known') + expect((await usage).providerSessionId).toBe('session-1') + }) }) diff --git a/src/main/codex-usage/store-test-harness.ts b/src/main/codex-usage/store-test-harness.ts index 3bbcdcc5f73..c9e517ede09 100644 --- a/src/main/codex-usage/store-test-harness.ts +++ b/src/main/codex-usage/store-test-harness.ts @@ -15,6 +15,55 @@ export function createEmptyScanResult() { } } +/** One completed session in `worktreeId`, shaped for automation-run attribution. */ +export function createWorktreeUsageSession(worktreeId: string) { + const tokens = { + eventCount: 1, + inputTokens: 1000, + cachedInputTokens: 400, + outputTokens: 250, + reasoningOutputTokens: 100, + totalTokens: 1250, + hasInferredPricing: false + } + return { + sessionId: 'session-1', + firstTimestamp: '2026-04-10T15:00:00.000Z', + lastTimestamp: '2026-04-10T15:05:00.000Z', + primaryModel: 'gpt-5', + hasMixedModels: false, + primaryProjectLabel: 'Repo', + hasMixedLocations: false, + primaryWorktreeId: worktreeId, + primaryRepoId: 'repo-1', + totalInputTokens: 1000, + totalCachedInputTokens: 400, + totalOutputTokens: 250, + totalReasoningOutputTokens: 100, + ...tokens, + locationBreakdown: [ + { + locationKey: `worktree:${worktreeId}`, + projectLabel: 'Repo', + repoId: 'repo-1', + worktreeId, + ...tokens + } + ], + modelBreakdown: [{ modelKey: 'gpt-5', modelLabel: 'gpt-5', ...tokens }], + locationModelBreakdown: [ + { + locationKey: `worktree:${worktreeId}`, + modelKey: 'gpt-5', + modelLabel: 'gpt-5', + repoId: 'repo-1', + worktreeId, + ...tokens + } + ] + } +} + export function createStoreWithState(state: Partial): CodexUsageStore { const store = new CodexUsageStore({ getRepos: () => [], diff --git a/src/main/codex-usage/store.ts b/src/main/codex-usage/store.ts index c187cdf290a..f11121f5d0d 100644 --- a/src/main/codex-usage/store.ts +++ b/src/main/codex-usage/store.ts @@ -140,7 +140,8 @@ export class CodexUsageStore extends UsageProviderStoreLifecycle< async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise { return resolveCodexAutomationRunUsage(input, { getState: () => this.state, - refresh: (force) => this.refresh(force) + refresh: (force) => this.refresh(force), + isScanning: () => this.getScanState().isScanning }) } } diff --git a/src/main/codex-usage/types.ts b/src/main/codex-usage/types.ts index d7e6f058b45..940a8263b0b 100644 --- a/src/main/codex-usage/types.ts +++ b/src/main/codex-usage/types.ts @@ -1,9 +1,34 @@ +import type { CodexUsageRawUsage } from './codex-usage-token-delta' + export type CodexUsageProcessedFile = { path: string mtimeMs: number size: number } +/** Everything needed to resume parsing a grown rollout where the last scan + * stopped, plus the evidence that the already-parsed prefix is still intact. */ +export type CodexUsageParseResumeState = { + /** Offset just past the last line that ended in a newline. Never the raw file + * size: a rollout can be observed mid-write with a partial trailing line. */ + parsedBytes: number + /** Digest of the bytes just before `parsedBytes`. A rewrite or rotation that + * leaves the file at the same length still changes this, unless it left the + * tail of the prefix byte-identical — which is what `headDigest` covers. */ + boundaryDigest: string + /** Digest of the bytes at the start of the parsed prefix. Catches a rewrite + * that replaced the leading records and kept the length and the tail. */ + headDigest: string + /** `dev:ino`, or null where the platform does not report an inode. Not a + * rotation check: ext4 and overlayfs reuse the inode of a recreated path. */ + physicalFileId: string | null + sessionId: string + sessionCwd: string | null + currentCwd: string | null + currentModel: string | null + previousTotals: CodexUsageRawUsage | null +} + export type CodexUsageLocationBreakdown = { locationKey: string projectLabel: string @@ -94,6 +119,9 @@ export type CodexUsagePersistedFile = CodexUsageProcessedFile & { * owner disappears, only deferred files need reparse to reclaim — not the * entire rollout corpus. */ hasDeferredClaims: boolean + /** Null when this file must be reparsed from byte 0 next scan. Absent on + * caches written before incremental resume shipped. */ + parseResumeState?: CodexUsageParseResumeState | null } export type CodexUsagePersistedState = { diff --git a/src/main/usage/automation-usage-scan-forcing.ts b/src/main/usage/automation-usage-scan-forcing.ts new file mode 100644 index 00000000000..85e2fb7e63e --- /dev/null +++ b/src/main/usage/automation-usage-scan-forcing.ts @@ -0,0 +1,27 @@ +type AutomationUsageScanAttempts = { + lastScanStartedAt: number | null + lastScanCompletedAt: number | null +} + +/** + * Whether an automation run's usage lookup must force a provider scan. + * + * Why: attribution needs one finished scan attempt after the run. Keying on the + * attempt instead of its outcome bounds this to a single forced scan per run — a + * persistently failing scan used to re-force on every lookup, forever. + */ +export function shouldForceAutomationUsageScan( + scanState: AutomationUsageScanAttempts, + completedAt: number, + isScanning: boolean +): boolean { + const { lastScanStartedAt, lastScanCompletedAt } = scanState + // Why: an in-flight scan's start time is not a finished attempt yet. Counting + // it sends the lookup down refresh(false), which returns early inside the + // staleness window instead of joining the scan, so the run reads unavailable. + // Forcing here only awaits the scan promise the lifecycle already shares. + const lastFinishedAttempt = isScanning + ? (lastScanCompletedAt ?? 0) + : Math.max(lastScanStartedAt ?? 0, lastScanCompletedAt ?? 0) + return lastFinishedAttempt < completedAt +} diff --git a/src/main/usage/jsonl-line-offsets.test.ts b/src/main/usage/jsonl-line-offsets.test.ts new file mode 100644 index 00000000000..c53880e9d7b --- /dev/null +++ b/src/main/usage/jsonl-line-offsets.test.ts @@ -0,0 +1,83 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { mkdtempSync, rmSync, statSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { readJsonlLinesFromOffset, type JsonlLineAtOffset } from './jsonl-line-offsets' + +let workDir: string + +async function collect(filePath: string, startOffset = 0): Promise { + const lines: JsonlLineAtOffset[] = [] + for await (const entry of readJsonlLinesFromOffset(filePath, startOffset)) { + lines.push(entry) + } + return lines +} + +function writeFixture(name: string, contents: string): string { + const filePath = join(workDir, name) + writeFileSync(filePath, contents, 'utf-8') + return filePath +} + +beforeEach(() => { + workDir = mkdtempSync(join(tmpdir(), 'orca-jsonl-offsets-')) +}) + +afterEach(() => { + rmSync(workDir, { recursive: true, force: true }) +}) + +describe('readJsonlLinesFromOffset', () => { + it('reports byte offsets past each newline', async () => { + const filePath = writeFixture('lf.jsonl', 'ab\ncde\n') + + expect(await collect(filePath)).toEqual([ + { line: 'ab', endOffset: 3, terminated: true }, + { line: 'cde', endOffset: 7, terminated: true } + ]) + }) + + it('strips the carriage return but counts it in the offset', async () => { + const filePath = writeFixture('crlf.jsonl', 'ab\r\ncde\r\n') + + expect(await collect(filePath)).toEqual([ + { line: 'ab', endOffset: 4, terminated: true }, + { line: 'cde', endOffset: 9, terminated: true } + ]) + }) + + it('flags a trailing line with no newline', async () => { + const filePath = writeFixture('partial.jsonl', 'ab\ncd') + + expect(await collect(filePath)).toEqual([ + { line: 'ab', endOffset: 3, terminated: true }, + { line: 'cd', endOffset: 5, terminated: false } + ]) + }) + + it('counts multibyte characters as bytes, not code points', async () => { + const filePath = writeFixture('utf8.jsonl', '"héllo→"\n"next"\n') + const firstLineBytes = Buffer.byteLength('"héllo→"\n', 'utf-8') + + const lines = await collect(filePath) + + expect(lines[0]).toEqual({ line: '"héllo→"', endOffset: firstLineBytes, terminated: true }) + expect(lines[1]?.endOffset).toBe(statSync(filePath).size) + }) + + it('resumes from a mid-file offset', async () => { + const filePath = writeFixture('resume.jsonl', 'one\ntwo\nthree\n') + + expect(await collect(filePath, 4)).toEqual([ + { line: 'two', endOffset: 8, terminated: true }, + { line: 'three', endOffset: 14, terminated: true } + ]) + }) + + it('yields nothing when the offset is already at the end', async () => { + const filePath = writeFixture('end.jsonl', 'one\n') + + expect(await collect(filePath, 4)).toEqual([]) + }) +}) diff --git a/src/main/usage/jsonl-line-offsets.ts b/src/main/usage/jsonl-line-offsets.ts new file mode 100644 index 00000000000..cea99cd4162 --- /dev/null +++ b/src/main/usage/jsonl-line-offsets.ts @@ -0,0 +1,63 @@ +/** + * Streams JSONL lines together with their exact byte offsets so an incremental + * scan can resume at the end of the last complete line. Byte accounting is done + * on raw buffers because `readline` hides how many bytes a line consumed, and a + * CRLF transcript would otherwise drift one byte per line. + */ +import { createReadStream } from 'node:fs' + +const LINE_FEED = 0x0a +const CARRIAGE_RETURN = 0x0d + +export type JsonlLineAtOffset = { + line: string + /** Absolute byte offset just past this line, terminator included. */ + endOffset: number + /** False when the file ended before a newline; the line may still grow. */ + terminated: boolean +} + +function decodeLine(pieces: Buffer[]): string { + const raw = pieces.length === 1 ? pieces[0] : Buffer.concat(pieces) + const end = raw.at(-1) === CARRIAGE_RETURN ? raw.length - 1 : raw.length + return raw.toString('utf-8', 0, end) +} + +export async function* readJsonlLinesFromOffset( + filePath: string, + startOffset: number +): AsyncGenerator { + const stream = createReadStream(filePath, { start: startOffset }) + const pending: Buffer[] = [] + let pendingBytes = 0 + let endOffset = startOffset + + for await (const rawChunk of stream) { + const chunk = Buffer.isBuffer(rawChunk) ? rawChunk : Buffer.from(rawChunk) + let searchFrom = 0 + for (;;) { + const lineFeedIndex = chunk.indexOf(LINE_FEED, searchFrom) + if (lineFeedIndex === -1) { + break + } + const segment = chunk.subarray(searchFrom, lineFeedIndex) + pending.push(segment) + pendingBytes += segment.length + endOffset += pendingBytes + 1 + const line = decodeLine(pending) + pending.length = 0 + pendingBytes = 0 + searchFrom = lineFeedIndex + 1 + yield { line, endOffset, terminated: true } + } + if (searchFrom < chunk.length) { + const remainder = chunk.subarray(searchFrom) + pending.push(remainder) + pendingBytes += remainder.length + } + } + + if (pendingBytes > 0) { + yield { line: decodeLine(pending), endOffset: endOffset + pendingBytes, terminated: false } + } +}