fix(usage): count Codex request ledgers and replace archived cache entries

apps/server/src/provider/Drivers/codexUsage.test.ts

@@ -450,6 +450,61 @@ 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);
+    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(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);
+  });
+
   it("does not bill a context-window estimate or lose the following real request", () => {
     const state = initialCodexScanState();
     parseCodexLine(turnContext, state);

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,31 @@ 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) {
+      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]),
+        },
+      ];
+    }
+    state.requestLedgerActive = false;
     (state.unreportedRequests ??= []).push({ ...codexAttribution(state, timestampMs), usage });
     return [];
   }
@@ -278,6 +298,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 +444,7 @@ export const codexUsageFormat: TranscriptUsageFormat<CodexScanState> = {
       session_id: true,
       model_provider: true,
       turn_id: true,
+      response_id: true,
       model: true,
       thread_settings: { service_tier: true },
       forked_from_id: true,

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<string, unknown>; 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<string, unknown>;
+          };
+          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"))

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,

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,19 +225,23 @@ 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,
         position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null },
       });
       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(

apps/server/src/usage/usageScanCache.ts

@@ -10,16 +10,17 @@ 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;
 
 /**
  * An older server sharing that directory cannot read a newer cache and would
  * 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;

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;