mirror of
https://github.com/stablyai/orca.git
synced 2026-09-22 00:02:31 +00:00
perf(codex-usage): resume rollout scans at the last parsed byte (#21102)
* perf(codex-usage): resume rollout scans at the last parsed byte
Codex rollout files are append-only and grow all day, but any append
changed both mtime and size, so `canReuse` discarded the cached entry and
the scanner re-read the whole file from byte 0 on the Electron main
process. On one real corpus that was 6.59 GB re-read per cycle across
26.63 GB / 21,110 files.
Each parsed file now persists a resume point: the offset just past the
last newline-terminated line, the parse context at that offset (session
id, cwd, model, running totals), a sha256 of the 4 KiB before it, and the
file's dev:ino. A grown file resumes there and merges the appended
rollup into the cached one; anything unproven falls back to a full
reparse — truncation, an in-place rewrite, rotation, a counted tail with
no trailing newline, a legacy copied-session suffix offset, or a file
that must reclaim deferred fork claims. Resume never depends on mtime
equality, so a coarse-mtime filesystem cannot hide an append.
Fixture: a 75,737-byte rollout with a 758-byte append re-read 76,495
bytes before and 8,950 after (the append plus two bounded 4 KiB boundary
windows).
Also bounds the automation-attribution force predicate for both Codex and
Claude: it keyed on `lastScanError`, so a persistently failing scan forced
a fresh full rescan on every single lookup. It now keys on the most recent
scan attempt, which is one forced scan per run regardless of outcome.
* fix(codex-usage): verify the head of a resumed rollout prefix
The resume guard proved only the 4 KiB before the resume offset, and leaned
on dev:ino to catch a rollout that was replaced at the same path. ext4 and
overlayfs hand a recreated file the inode the old one freed, so on Linux that
check passes and a same-length prefix swap resumes over changed history.
Measured 20/20 inode reuse on ext4 and overlayfs, 0/20 on APFS and tmpfs --
which is why the case only failed in CI.
An in-place prefix rewrite kept no inode change on any platform, so that
variant was missed on macOS too.
Digest a bounded window at the start of the parsed prefix as well. When the
two windows meet, one read covers the whole prefix and leaves no gap. The
head window is carried across a resume rather than re-read, so a resumed scan
reads the appended bytes plus three 4 KiB windows.
* test(codex-usage): cover the resume window layout switch
* test(codex-usage): cover the boundary window in isolation
* test(codex-usage): isolate the boundary window with disjoint windows
* fix(codex-usage): restart a rollout parse when its verified prefix is gone
The scanner verifies a rollout's prefix in its first pass and reads it in
the second, so a truncation in between left the merged projection holding
the whole pre-truncation history while `processedFile` was re-stat'd to the
new, smaller size. Size and mtime then matched disk with no resume state
left to reject, so the reuse path served the stale total on every later
scan. The resume-state builder returns null only on a short read, which is
exactly that signal; on it, drop the merge and reparse the file from zero.
Also covers three guards that no test was holding: the unterminated-tail
resume suppression (a tail that is valid JSON minus its newline is counted,
so resuming over it double-counts), the short-read check in
`readWindowDigest` (without it a resume point past EOF verifies against
itself), and the legacy-suffix exclusion in the scanner's resume guard
(bridge markers can appear on a file that already has a resume state).
* fix(codex-usage): re-verify a rollout resume point at the point of use
The scanner verified each resume point while walking the sessions
directory, then parsed the files afterwards, so every file discovered or
parsed in between widened the gap between the check and the read. A
rollout replaced in that gap resumed at the old offset into unrelated
bytes: the cached session id, cwd, model and running totals were stitched
onto another file's records, and because the projection was then re-stat'd
to the new size, the reuse path froze the corrupted numbers. A shrink was
the visible half of this; a replacement larger than the recorded offset
never short-reads and corrupts instead of going stale.
Re-run the full check — inode, head window and boundary window — inside
the parse, against the file about to be read. The short-read fallback
added alongside it still covers the narrower case of a truncation landing
after that check, during the read itself.
Cost, measured on the existing byte oracle: a resumed file now reads
`appended + 5 * 4096` rather than `appended + 3 * 4096`, paid only by
files that changed since the last scan; untouched rollouts still read
nothing. Two byte-total assertions that a 15 KB rollout can no longer
satisfy now assert their intent directly — that the parse read did not
reopen at byte 0 — via a stream oracle that records each read's offset.
* test(codex-usage): pin mid-scan replacement on attribution, not totals
The mid-scan replacement case was written with a heavier replacement so
the token totals diverged, which overstated how visible the defect is.
Rebuilt on the variant where the stale prefix contributes exactly as many
events as the resumed read skips: daily aggregates and token totals then
match a cold scan byte for byte, and the misattribution — 60 records of
one session recorded against another — is the only remaining signal.
Oracle is now the session shape. Removing the point-of-use re-verification
fails it with `session-grower` in place of `session-other`; every
totals-based assertion still passes under that mutation.
* perf(codex-usage): stop resuming a rollout prefix too short to pay for it
Point-of-use re-verification made a resumed scan cost five bounded windows,
which is more than re-reading a small rollout outright. Measured against a
cold reparse of the same file, resuming lost below a 12,288 B prefix and
lost badly under 8 KiB, where the coalesced-window layout rehashed the
whole prefix on each of the three verification passes.
Set the floor at that break-even — 3 * 4096, the point where two
verification passes plus the recorded boundary stop being cheaper than
reading the prefix once — and refuse to record or accept a resume point
below it. Measured: a 12,568 B prefix now reads 21,234 B resumed against
21,514 B cold, and a 76,484 B rollout reads 21,238 B against 84,676 B. No
size band reads more than a cold scan any more; under the floor the
windows are skipped entirely and a scan reads exactly the file.
With every offset past the floor the two windows can no longer overlap, so
the coalesced-layout branch and the empty-window branch are gone. The
floor is also input validation: a persisted offset below it would put the
boundary window at a negative start and throw ERR_OUT_OF_RANGE.
Tests that meant to exercise the resume path were silently reparsing whole
once the floor landed — the suite stayed green while three guards lost
their only coverage. They now size their rollouts off RESUMABLE_RECORDS
and assert the offsets their parse reads actually opened at, so a test
that stops resuming fails instead of passing quietly.
* test(codex-usage): cover the reuse gate's own legacy-bridge check
`scanner.ts` carries the same `legacySourceSkipBytes === 0` term twice and
they are different guards: line 83 gates resuming, line 71 gates reuse.
Only the first had a test, so dropping the second left the suite green.
It is load-bearing. A cached entry can predate the bridge marker while the
source file is untouched, so size and mtime still match and nothing else
stops the scan serving a full-history projection for a file that is now
parsed suffix-only. With a total-only record after the copy point the two
readings diverge — baseline worth nothing against a delta worth three —
and the reused entry reports 18 tokens where a cold scan reports 15.
* fix(codex-usage): annotate the mid-scan seam instead of asserting it
The changed-code quality gate rejects any non-const type assertion, and
`onStreamOpen: { current: null as (...) | null }` is one, so `static
analysis` failed on this PR. A typed local carries the same intent.
* fix(usage): force an automation lookup onto a scan already in flight
`shouldForceAutomationUsageScan` keyed on `max(lastScanStartedAt,
lastScanCompletedAt)`, so a scan that started after the run completed but
is still running counted as a finished attempt. The lookup then called
`refresh(false)`, which returns early inside the 5-minute staleness
window instead of joining the scan, and the run's usage read
`unavailable`. Forcing instead just awaits the shared `scanPromise`.
While a scan is in flight its start time is no longer treated as an
attempt, so the once-per-run bound still holds: a failed scan leaves
`lastScanStartedAt` past the run and stops re-forcing.
The two providers' copies of the predicate were byte-identical, so it now
lives in `src/main/usage/automation-usage-scan-forcing.ts`.
This commit is contained in:
Jinwoo Hong
@@ -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)
|
||||
|
||||
@@ -51,6 +51,42 @@ function createStoreWithState(state: Partial<ClaudeUsagePersistedState>): 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<void>((resolve) => {
|
||||
startScan = resolve
|
||||
})
|
||||
const scanFinished = new Promise<void>((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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -139,7 +139,8 @@ export class ClaudeUsageStore extends UsageProviderStoreLifecycle<
|
||||
async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise<AutomationRunUsage> {
|
||||
return resolveAutomationRunUsage(input, {
|
||||
getState: () => this.state,
|
||||
refresh: (force) => this.refresh(force)
|
||||
refresh: (force) => this.refresh(force),
|
||||
isScanning: () => this.getScanState().isScanning
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<CodexUsageProcessedFile> {
|
||||
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<string, CodexUsageSession>()
|
||||
mergeSessions(sessionsById, previous.sessions)
|
||||
mergeSessions(sessionsById, appended.sessions)
|
||||
const dailyByKey = new Map<string, CodexUsageDailyAggregate>()
|
||||
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<CodexUsagePersistedFile> {
|
||||
// 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<string>()
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
@@ -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<string | null> {
|
||||
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<PrefixDigests | null> {
|
||||
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<string | null> {
|
||||
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<CodexUsageParseResumeState | null> {
|
||||
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<CodexUsageParseResumeState | null> {
|
||||
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
|
||||
}
|
||||
@@ -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 }))
|
||||
})
|
||||
})
|
||||
@@ -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<typeof NodeOs>('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<typeof NodeFs>('node:fs')
|
||||
return {
|
||||
...actual,
|
||||
createReadStream: (
|
||||
path: Parameters<typeof actual.createReadStream>[0],
|
||||
options?: Parameters<typeof actual.createReadStream>[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)
|
||||
})
|
||||
})
|
||||
@@ -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(
|
||||
|
||||
+28
-105
@@ -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<CodexUsageProcessedFile> {
|
||||
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<CodexUsagePersistedFile> {
|
||||
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<string>()
|
||||
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<string, CodexUsagePersistedFile>()
|
||||
const resumeByPath = new Map<string, CodexRolloutResumePlan>()
|
||||
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<string, string>()
|
||||
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<string, CodexUsagePersistedFile>()
|
||||
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) {
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
startScan = resolve
|
||||
})
|
||||
const scanFinished = new Promise<void>((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')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<CodexUsagePersistedState>): CodexUsageStore {
|
||||
const store = new CodexUsageStore({
|
||||
getRepos: () => [],
|
||||
|
||||
@@ -140,7 +140,8 @@ export class CodexUsageStore extends UsageProviderStoreLifecycle<
|
||||
async getAutomationRunUsage(input: AutomationUsageLookupInput): Promise<AutomationRunUsage> {
|
||||
return resolveCodexAutomationRunUsage(input, {
|
||||
getState: () => this.state,
|
||||
refresh: (force) => this.refresh(force)
|
||||
refresh: (force) => this.refresh(force),
|
||||
isScanning: () => this.getScanState().isScanning
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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<JsonlLineAtOffset[]> {
|
||||
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([])
|
||||
})
|
||||
})
|
||||
@@ -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<JsonlLineAtOffset> {
|
||||
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 }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user