From 1b29bf8379bc6dafa4153ae4bda7ea0bede42768 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Thu, 12 Nov 2026 12:37:12 -0600 Subject: [PATCH 1/4] fix(usage): count Codex request ledgers and replace archived cache entries --- .../src/provider/Drivers/codexUsage.test.ts | 37 +++++++++++++ .../server/src/provider/Drivers/codexUsage.ts | 26 ++++++++-- apps/server/src/usage/UsageService.test.ts | 52 +++++++++++++++++-- apps/server/src/usage/UsageService.ts | 7 +++ apps/server/src/usage/usageScanCache.test.ts | 18 ++++--- apps/server/src/usage/usageScanCache.ts | 11 ++-- 6 files changed, 132 insertions(+), 19 deletions(-) diff --git a/apps/server/src/provider/Drivers/codexUsage.test.ts b/apps/server/src/provider/Drivers/codexUsage.test.ts index 74812bafb5099..de55645cd7e15 100644 --- a/apps/server/src/provider/Drivers/codexUsage.test.ts +++ b/apps/server/src/provider/Drivers/codexUsage.test.ts @@ -450,6 +450,43 @@ describe("parseCodexLine", () => { expect(state.malformedRecords).toBe(0); }); + it("bills request-ledger usage through resumed context counters and compaction", () => { + const state = initialCodexScanState(); + const request = (response_id: string, usage: unknown) => + JSON.stringify({ + type: "token_usage_record", + timestamp: "2026-08-01T05:17:45.000Z", + payload: { response_id, usage }, + }); + parseCodexLine(sessionMeta, state); + parseCodexLine(turnContext, state); + parseCodexLine(counted(counters(100, 20), counters(1000, 200)), state); + // Resume restores older context counters, while the request ledger continues. + expect( + parseCodexLine(request("response-1", counters(80, 10)), state)?.totals.outputTokens, + ).toBe(10); + expect(parseCodexLine(counted(counters(80, 10), counters(580, 110)), state)).toBeNull(); + // Rate-limit refreshes repeat the snapshot without another request. + expect(parseCodexLine(counted(counters(80, 10), counters(580, 110)), state)).toBeNull(); + expect( + parseCodexLine(request("compaction", counters(90, 15)), state)?.totals.outputTokens, + ).toBe(15); + parseCodexLine(turnContext, state); + expect( + parseCodexLine(counted({ ...counters(0, 0), total_tokens: 40 }, counters(580, 110)), state), + ).toBeNull(); + expect(parseCodexLine(request("response-2", counters(30, 5)), state)?.totals.outputTokens).toBe( + 5, + ); + expect(parseCodexLine(counted(counters(30, 5), counters(610, 115)), state)).toBeNull(); + expect(state.malformedRecords).toBe(0); + // A later turn from an older CLI can still report usage through snapshots alone. + parseCodexLine(turnContext, state); + expect( + parseCodexLine(counted(counters(20, 2), counters(630, 117)), state)?.totals.outputTokens, + ).toBe(2); + }); + it("does not bill a context-window estimate or lose the following real request", () => { const state = initialCodexScanState(); parseCodexLine(turnContext, state); diff --git a/apps/server/src/provider/Drivers/codexUsage.ts b/apps/server/src/provider/Drivers/codexUsage.ts index 90f7c78b849f8..a49eb8e75a67d 100644 --- a/apps/server/src/provider/Drivers/codexUsage.ts +++ b/apps/server/src/provider/Drivers/codexUsage.ts @@ -127,6 +127,7 @@ export const CodexScanState = Schema.Struct({ unreportedRequests: Schema.mutableKey( Schema.optionalKey(Schema.mutable(Schema.Array(CodexUnreportedRequest))), ), + requestLedgerActive: Schema.mutableKey(Schema.optionalKey(Schema.Boolean)), malformedRecords: Schema.mutableKey(TokenCount), sawSessionMeta: Schema.mutableKey(Schema.Boolean), modelProvider: Schema.mutableKey(Schema.String), @@ -181,8 +182,8 @@ function isForkCopy(state: CodexScanState, timestampMs: number): boolean { } /** - * Cumulative counters validate request deltas. Older events without cumulative counters use - * normalized consecutive-request deduplication. + * Identified request-ledger entries report spending independently of context snapshots. + * Legacy snapshots validate request deltas against cumulative counters. */ export function parseCodexLine(line: string, state: CodexScanState): readonly UsageRecord[] { let parsed: unknown; @@ -221,6 +222,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag } if (record["type"] === "turn_context") { + state.requestLedgerActive = false; if (typeof payloadRecord["model"] === "string") state.model = payloadRecord["model"]; // Model carry is useful for usage, but history recovery needs this context's own identity. state.turnId = @@ -232,13 +234,24 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag return []; } - // Several of these may sit between two snapshots, so they only explain a gap in the counters. + // Request ledgers include spending that context snapshots can omit. if (record["type"] === "token_usage_record") { const timestampMs = parseTimestampMs(record["timestamp"]); const usage = readCodexTokenUsage(payloadRecord["usage"]); if (timestampMs === null || usage === null || state.model.length === 0) return []; // Copied history is already billed in the parent. if (isForkCopy(state, timestampMs)) return []; + const responseId = payloadRecord["response_id"]; + if (typeof responseId === "string" && responseId.length > 0) { + state.requestLedgerActive = true; + delete state.unreportedRequests; + return [ + { + ...codexRequestRecord(state, usage, codexAttribution(state, timestampMs)), + dedupeKey: JSON.stringify(["codex", state.sessionId, responseId]), + }, + ]; + } (state.unreportedRequests ??= []).push({ ...codexAttribution(state, timestampMs), usage }); return []; } @@ -278,6 +291,12 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag return []; } + // Context snapshots can rewind on resume and omit compaction spending; identified requests are authoritative. + if (state.requestLedgerActive) { + if (cumulative !== null) rebaseCodexUsage(state, cumulative); + return []; + } + const previous = state.lastCumulativeUsage; const pending = state.unreportedRequests; // A valid snapshot remains the next baseline even when its paired request is malformed. @@ -418,6 +437,7 @@ export const codexUsageFormat: TranscriptUsageFormat = { session_id: true, model_provider: true, turn_id: true, + response_id: true, model: true, thread_settings: { service_tier: true }, forked_from_id: true, diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 3644a53bbc396..bff9bdb1f57fd 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -577,9 +577,55 @@ describe("UsageService", () => { assert.strictEqual(totalOutputTokens(rewritten), 71); assert.deepStrictEqual((yield* service.readSummary(WINDOW)).buckets, rewritten.buckets); yield* service.awaitPersisted; + const { stateDir } = yield* ServerConfig.ServerConfig; + const cachePath = NodePath.join(stateDir, SCAN_CACHE_FILE_NAME); + const originalPath = (yield* Effect.promise(() => NodeFSP.realpath(compressedPath))).slice( + 0, + -4, + ); + yield* Effect.promise(async () => { + const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as { + files: Record< + string, + { cs: Record; r: unknown[]; t: unknown[]; mr: number } + >; + }; + const archived = document.files[await NodeFSP.realpath(compressedPath)]; + assert.ok(archived); + document.files[originalPath] = { + ...archived, + r: [], + t: [], + mr: 1, + cs: { ...archived.cs, malformedRecords: 1 }, + }; + await NodeFSP.writeFile(cachePath, encodeUnknownJsonString(document)); + }); const restarted = yield* UsageService.make; - assert.deepStrictEqual((yield* restarted.readSummary(WINDOW)).buckets, rewritten.buckets); + const restored = yield* restarted.readSummary(WINDOW); + assert.deepStrictEqual(restored.buckets, rewritten.buckets); + assert.strictEqual( + restored.sources.find((source) => source.fingerprint.provider === "codex") + ?.malformedRecords, + 0, + ); yield* restarted.awaitPersisted; + yield* Effect.promise(async () => { + const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as { + files: Record; + }; + assert.strictEqual(document.files[originalPath], undefined); + await NodeFSP.rm(compressedPath); + }); + const removed = yield* UsageService.make; + const retained = yield* removed.readSummary(WINDOW); + assert.deepStrictEqual(retained.buckets, rewritten.buckets); + assert.strictEqual( + retained.sources.find((source) => source.fingerprint.provider === "codex") + ?.malformedRecords, + 0, + ); + yield* removed.awaitPersisted; }).pipe( Effect.scoped, Effect.provide(layerService({ prefix: "usage-compressed-codex", home, settings })), @@ -2789,7 +2835,7 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); - it.live.each([4, 5, 6, 7, 8, 10])( + it.live.each([4, 5, 6, 7, 8, 10, 11, 12, 13])( "upgrades a v%s cache: reparses live Codex usage, keeps deleted rollouts, removes the old file", (version) => Effect.gen(function* () { @@ -2890,7 +2936,7 @@ describe("UsageService", () => { 60 + 20 * (version === 4 ? 1 : 2), ); yield* upgraded.awaitPersisted; - assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v13.json"); + assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v14.json"); assert.deepStrictEqual( (yield* Effect.promise(() => NodeFSP.readdir(stateDir))) .filter((name) => name.startsWith("usage-scan-cache")) diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 341b14d8c6280..912ed51c96b3e 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -638,6 +638,13 @@ export const make = Effect.gen(function* () { } return { path, records, malformedRecords }; }); + const livePaths = new Set(files.map((file) => file.path)); + for (const file of read) { + if (file.failed || !file.path.endsWith(".jsonl.zst")) continue; + const originalPath = file.path.slice(0, -4); + // A successfully read archive replaces the original's saved records and diagnostics. + if (!livePaths.has(originalPath) && fileCache.delete(originalPath)) cacheDirty = true; + } const unread = failedPaths + read.filter((file) => file.failed).length; yield* Effect.annotateCurrentSpan({ "usage.files": files.length, diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index a8fba4da62ae6..a6a0f6d3c41df 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -168,7 +168,7 @@ describe("scan cache round trip", () => { expect(restored).toEqual(original); }); - it.each([5, 6, 7, 8, 10])( + it.each([5, 6, 7, 8, 10, 11, 12, 13])( "retains v%s history and compatible positions but invalidates obsolete state", (version) => { const original = cacheWith([["/gone.jsonl", 100, [record({ speed: "fast" })]]]); @@ -225,11 +225,15 @@ describe("scan cache round trip", () => { }; const restored = decodeScanCache(JSON.parse(JSON.stringify(legacy)), TEST_FORMAT_MAP); - expect(restored.get("/gone.jsonl")).toEqual({ - ...original.get("/gone.jsonl"), - size: -1, - position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, - }); + expect(restored.get("/gone.jsonl")).toEqual( + version >= 13 + ? original.get("/gone.jsonl") + : { + ...original.get("/gone.jsonl"), + size: -1, + position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, + }, + ); expect(restored.get("/gone-codex.jsonl")).toEqual({ ...original.get("/gone-codex.jsonl"), size: -1, @@ -237,7 +241,7 @@ describe("scan cache round trip", () => { }); expect(restored.get("/clean-codex.jsonl")).toEqual({ ...cleanCodex, - size: version >= 8 ? cleanCodex.size : -1, + size: -1, position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, }); expect(restored.get("/gone-pi.jsonl")).toEqual( diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 4f1137f3f7a1c..604937a6a5e41 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -10,7 +10,7 @@ import { CLAUDE_SYNTHETIC_MODEL } from "../provider/Drivers/claudeUsage.ts"; import { mergeUsageCopies } from "./usageAggregation.ts"; import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts"; -const USAGE_SCAN_CACHE_VERSION = 13 as const; +const USAGE_SCAN_CACHE_VERSION = 14 as const; const SPEED_COMPATIBLE_SINCE_VERSION = 4; /** @@ -18,8 +18,9 @@ const SPEED_COMPATIBLE_SINCE_VERSION = 4; * replace it, dropping saved usage for deleted transcripts. Separate files keep * the newer one; older files are removed once it is saved. */ -export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v13.json"; +export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v14.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [ + "usage-scan-cache-v13.json", "usage-scan-cache-v12.json", "usage-scan-cache-v11.json", "usage-scan-cache-v10.json", @@ -296,14 +297,12 @@ export function decodeScanCache( continue; } // Keep history for deleted rollouts, but re-parse incompatible reducer state. - // v9 and v11 repair skipped Codex requests; clean v8+ entries stay served while unchanged. + // v14 uses Codex request ledgers, including compaction spending omitted by context snapshots. // v10 adds Claude advisor calls; before v12 a repeated Claude key kept its first, possibly partial, copy. // v13 adds Claude 1-hour cache writes. const legacy = (entry.p === "claude" && version < 13) || - (format.state !== undefined && - (version < 6 || - (entry.p === "codex" && (version < 8 || (version < 11 && malformedRecords > 0))))); + (format.state !== undefined && (version < 6 || (entry.p === "codex" && version < 14))); // A corrupt state disqualifies the entry: resuming with it would attach // appended usage to the wrong model or replay fork-copied history. if (!legacy && !isValidState.get(entry.p)?.(entry.cs)) continue; From dfd3372e76bfbd946a7f428e8645c56aa2bfd605 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Thu, 12 Nov 2026 12:43:23 -0600 Subject: [PATCH 2/4] fix(usage): retain accounting when Codex record formats change --- apps/server/src/provider/Drivers/codexUsage.test.ts | 13 +++++++++++++ apps/server/src/provider/Drivers/codexUsage.ts | 6 ++++++ 2 files changed, 19 insertions(+) diff --git a/apps/server/src/provider/Drivers/codexUsage.test.ts b/apps/server/src/provider/Drivers/codexUsage.test.ts index de55645cd7e15..50560df18d7bb 100644 --- a/apps/server/src/provider/Drivers/codexUsage.test.ts +++ b/apps/server/src/provider/Drivers/codexUsage.test.ts @@ -485,6 +485,19 @@ describe("parseCodexLine", () => { expect( parseCodexLine(counted(counters(20, 2), counters(630, 117)), state)?.totals.outputTokens, ).toBe(2); + expect( + parseCodexLine(request("interrupted", counters(40, 4)), state)?.totals.outputTokens, + ).toBe(4); + parseCodexLine(turnContext, state); + expect( + parseCodexLine(counted(counters(10, 1), counters(680, 122)), state)?.totals.outputTokens, + ).toBe(1); + expect(parseCodexLine(request("", counters(15, 3)), state)).toBeNull(); + parseCodexLine(turnContext, state); + const transition = parseCodexRecords(request("upgraded", counters(25, 5)), state); + expect(transition.map((record) => record.totals.outputTokens)).toEqual([3, 5]); + expect(parseCodexLine(counted(counters(25, 5), counters(720, 130)), state)).toBeNull(); + expect(state.malformedRecords).toBe(0); }); it("does not bill a context-window estimate or lose the following real request", () => { diff --git a/apps/server/src/provider/Drivers/codexUsage.ts b/apps/server/src/provider/Drivers/codexUsage.ts index a49eb8e75a67d..eedcfda94ea3c 100644 --- a/apps/server/src/provider/Drivers/codexUsage.ts +++ b/apps/server/src/provider/Drivers/codexUsage.ts @@ -243,9 +243,15 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag if (isForkCopy(state, timestampMs)) return []; const responseId = payloadRecord["response_id"]; if (typeof responseId === "string" && responseId.length > 0) { + const pending = state.unreportedRequests ?? []; state.requestLedgerActive = true; + for (const request of pending) { + state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, request.usage); + } + state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, usage); delete state.unreportedRequests; return [ + ...pending.map((request) => codexRequestRecord(state, request.usage, request)), { ...codexRequestRecord(state, usage, codexAttribution(state, timestampMs)), dedupeKey: JSON.stringify(["codex", state.sessionId, responseId]), From fdd5aec35f2aa2bd380b579115232c0784f2fac9 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Thu, 12 Nov 2026 12:43:23 -0600 Subject: [PATCH 3/4] test(server): include repository state in HEAD watch expectations --- apps/server/src/vcs/VcsStatusBroadcaster.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 37d767c29c044..ed9351b37e498 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -1555,13 +1555,13 @@ describe("VcsStatusBroadcaster", () => { yield* Effect.gen(function* () { const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; const branches = yield* Stream.toPull(broadcaster.streamBranchStatus({ cwd: "/repo" })); - assert.deepEqual(yield* branches, [{ refName: "main" }]); + assert.deepEqual(yield* branches, [{ refName: "main", isRepo: true }]); // Every checkout removes HEAD.lock, which leaves the watch running. state.currentLocalStatus = { ...baseLocalStatus, refName: "other" }; yield* Queue.offer(events, { _tag: "Remove", path: "HEAD.lock" }); yield* Queue.offer(events, { _tag: "Update", path: "HEAD" }); - assert.deepEqual(yield* branches, [{ refName: "other" }]); + assert.deepEqual(yield* branches, [{ refName: "other", isRepo: true }]); // `git worktree remove` deletes the watched directory; Windows keeps reporting it. gitDirExists = false; From 49d2774b948885bed09a7ec6b739ede3d9b06038 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Thu, 12 Nov 2026 12:48:08 -0600 Subject: [PATCH 4/4] fix(usage): reconcile unkeyed requests after ledger records --- apps/server/src/provider/Drivers/codexUsage.test.ts | 5 +++++ apps/server/src/provider/Drivers/codexUsage.ts | 1 + 2 files changed, 6 insertions(+) diff --git a/apps/server/src/provider/Drivers/codexUsage.test.ts b/apps/server/src/provider/Drivers/codexUsage.test.ts index 50560df18d7bb..a3abb2ccb4fb1 100644 --- a/apps/server/src/provider/Drivers/codexUsage.test.ts +++ b/apps/server/src/provider/Drivers/codexUsage.test.ts @@ -497,6 +497,11 @@ describe("parseCodexLine", () => { const transition = parseCodexRecords(request("upgraded", counters(25, 5)), state); expect(transition.map((record) => record.totals.outputTokens)).toEqual([3, 5]); expect(parseCodexLine(counted(counters(25, 5), counters(720, 130)), state)).toBeNull(); + expect(parseCodexLine(request("", counters(10, 1)), state)).toBeNull(); + expect( + parseCodexLine(counted(counters(10, 1), counters(730, 131)), state)?.totals.outputTokens, + ).toBe(1); + expect(parseCodexLine(counted(counters(10, 1), counters(730, 131)), state)).toBeNull(); expect(state.malformedRecords).toBe(0); }); diff --git a/apps/server/src/provider/Drivers/codexUsage.ts b/apps/server/src/provider/Drivers/codexUsage.ts index eedcfda94ea3c..540e06fd2b8dd 100644 --- a/apps/server/src/provider/Drivers/codexUsage.ts +++ b/apps/server/src/provider/Drivers/codexUsage.ts @@ -258,6 +258,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag }, ]; } + state.requestLedgerActive = false; (state.unreportedRequests ??= []).push({ ...codexAttribution(state, timestampMs), usage }); return []; }