fix(server): reconcile Codex usage counters

apps/server/src/usage/UsageService.test.ts

@@ -102,6 +102,87 @@ function totalOutputTokens(summary: { buckets: readonly { totals: { outputTokens
 }
 
 describe("UsageService", () => {
+  const usage = (input: number, output: number) => ({
+    input_tokens: input,
+    output_tokens: output,
+    total_tokens: input + output,
+  });
+  const line = (last: unknown, total: unknown) =>
+    JSON.stringify({
+      type: "event_msg",
+      timestamp: "2026-08-01T10:00:05Z",
+      payload: {
+        type: "token_count",
+        info: { last_token_usage: last, total_token_usage: total },
+      },
+    }) + "\n";
+  const model =
+    JSON.stringify({
+      type: "turn_context",
+      payload: { model: "gpt-5.4" },
+    }) + "\n";
+
+  it.live("reports inconsistent Codex counters through warm scans, restart, and recovery", () =>
+    Effect.gen(function* () {
+      const { settings, home } = yield* setup;
+      const dir = NodePath.join(home, "codex", "sessions");
+      const path = NodePath.join(dir, "rollout.jsonl");
+      yield* Effect.promise(() => NodeFSP.mkdir(dir, { recursive: true }));
+      yield* Effect.promise(() =>
+        NodeFSP.writeFile(
+          path,
+          model +
+            line(usage(100, 20), usage(100, 20)) +
+            line(usage(20, 5), usage(100, 20)) +
+            line(usage(10, 2), usage(130, 28)),
+        ),
+      );
+      yield* Effect.gen(function* () {
+        const service = yield* UsageService.make;
+        for (let scan = 0; scan < 2; scan++) {
+          const result = yield* service.readSummary(WINDOW);
+          assert.strictEqual(totalOutputTokens(result), 20);
+          const source = result.sources.find((source) => source.fingerprint.provider === "codex");
+          assert.strictEqual(source?.status, "partial");
+          assert.strictEqual(source?.malformedRecords, 1);
+          assert.include(source?.message ?? "", "Inconsistent usage records: 1");
+        }
+        const restarted = yield* UsageService.make;
+        // A warm cache restores diagnostics without touching the transcript.
+        const open = vi
+          .spyOn(NodeFSP, "open")
+          .mockRejectedValue(new Error("unexpected transcript read"));
+        const restored = yield* restarted
+          .readSummary(WINDOW)
+          .pipe(Effect.ensuring(Effect.sync(() => open.mockRestore())));
+        assert.strictEqual(totalOutputTokens(restored), 20);
+        assert.strictEqual(
+          restored.sources.find((source) => source.fingerprint.provider === "codex")
+            ?.malformedRecords,
+          1,
+        );
+        yield* Effect.promise(() => NodeFSP.appendFile(path, line(usage(10, 2), usage(140, 30))));
+        const appended = yield* restarted.readSummary(WINDOW);
+        assert.strictEqual(totalOutputTokens(appended), 22);
+        assert.strictEqual(
+          appended.sources.find((source) => source.fingerprint.provider === "codex")
+            ?.malformedRecords,
+          1,
+        );
+        yield* Effect.promise(() =>
+          NodeFSP.writeFile(path, model + line(usage(100, 20), usage(100, 20))),
+        );
+        const repaired = yield* restarted.readSummary(WINDOW);
+        const source = repaired.sources.find((source) => source.fingerprint.provider === "codex");
+        assert.strictEqual(totalOutputTokens(repaired), 20);
+        assert.strictEqual(source?.status, "ok");
+        assert.strictEqual(source?.malformedRecords, 0);
+      }).pipe(
+        Effect.provide(serviceLayers({ prefix: "usage-counter-reconciliation", home, settings })),
+      );
+    }).pipe(Effect.scoped),
+  );
+
   it.live("reprices unchanged transcripts when custom prices are added, edited, or removed", () =>
     Effect.gen(function* () {
       const { transcript, settings, home } = yield* setup;

apps/server/src/usage/UsageService.ts

@@ -317,7 +317,10 @@ export const make = Effect.gen(function* () {
     size: number,
     mtimeMs: number,
     provider: UsageProviderKind,
-  ): Effect.Effect<readonly UsageRecord[] | null> =>
+  ): Effect.Effect<{
+    readonly records: readonly UsageRecord[];
+    readonly malformedRecords: number;
+  } | null> =>
     Effect.gen(function* () {
       const cached = fileCache.get(filePath);
       // Provider is part of the identity: if both providers were ever pointed
@@ -328,9 +331,13 @@ export const make = Effect.gen(function* () {
         cached.mtimeMs === mtimeMs &&
         cached.provider === provider
       ) {
-        return cached.tailRecords.length === 0
-          ? cached.records
-          : [...cached.records, ...cached.tailRecords];
+        return {
+          records:
+            cached.tailRecords.length === 0
+              ? cached.records
+              : [...cached.records, ...cached.tailRecords],
+          malformedRecords: cached.malformedRecords,
+        };
       }
 
       // Only a strictly grown file may resume. Same size with a new mtime, or
@@ -362,10 +369,14 @@ export const make = Effect.gen(function* () {
         provider,
         records,
         tailRecords,
+        malformedRecords: parsed.malformedRecords,
         position: parsed.position,
       });
       cacheDirty = true;
-      return tailRecords.length === 0 ? records : [...records, ...tailRecords];
+      return {
+        records: tailRecords.length === 0 ? records : [...records, ...tailRecords],
+        malformedRecords: parsed.malformedRecords,
+      };
     });
 
   /** One provider directory's walk and parse, before rates are involved. */
@@ -375,6 +386,7 @@ export const make = Effect.gen(function* () {
     readonly volumeId: string;
     readonly status: "ok" | "partial" | "missing" | "failed";
     readonly failedEntries: number;
+    readonly malformedRecords: number;
     readonly files: readonly {
       readonly path: string;
       readonly records: readonly UsageRecord[] | null;
@@ -396,17 +408,20 @@ export const make = Effect.gen(function* () {
       );
       const parsedFiles: { path: string; records: readonly UsageRecord[] | null }[] = [];
       let failedEntries = listing.failedEntries;
+      let malformedRecords = 0;
       for (const file of listing.files) {
-        const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider);
-        if (records === null) failedEntries += 1;
-        parsedFiles.push({ path: file.path, records });
+        const parsed = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider);
+        if (parsed === null) failedEntries += 1;
+        malformedRecords += parsed?.malformedRecords ?? 0;
+        parsedFiles.push({ path: file.path, records: parsed?.records ?? null });
       }
       scanned.push({
         provider,
         dir,
         volumeId,
         files: parsedFiles,
         failedEntries,
+        malformedRecords,
         status: listing.status === "ok" && failedEntries > 0 ? "partial" : listing.status,
       });
     }
@@ -484,7 +499,15 @@ export const make = Effect.gen(function* () {
     const livePaths = new Set<string>();
     const walkedRoots: string[] = [];
 
-    for (const { provider, dir, volumeId, files, status, failedEntries } of scannedDirs) {
+    for (const {
+      provider,
+      dir,
+      volumeId,
+      files,
+      status,
+      failedEntries,
+      malformedRecords,
+    } of scannedDirs) {
       if (status === "missing" || status === "failed") {
         sources.push({
           fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId },
@@ -527,14 +550,14 @@ export const make = Effect.gen(function* () {
 
       sources.push({
         fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId },
-        status,
+        status: malformedRecords > 0 ? "partial" : status,
         scannedFiles,
         skippedFiles,
-        malformedRecords: 0,
+        malformedRecords,
         distinctSessions: sessionIds.size,
         message:
-          status === "partial"
-            ? `Usage is incomplete: ${failedEntries} transcript files or directory entries could not be read.`
+          failedEntries > 0 || malformedRecords > 0
+            ? `Usage is incomplete. Unreadable transcript files or directory entries: ${failedEntries}. Inconsistent usage records: ${malformedRecords}.`
             : null,
       });
     }

apps/server/src/usage/usageScanCache.test.ts

@@ -8,7 +8,7 @@ import {
   type CachedFile,
   type ScanCache,
 } from "./usageScanCache.ts";
-import type { UsageRecord } from "./usageTranscripts.ts";
+import { initialCodexScanState, type UsageRecord } from "./usageTranscripts.ts";
 
 function record(overrides: Partial<UsageRecord> = {}): UsageRecord {
   return {
@@ -47,6 +47,7 @@ function cacheWith(entries: readonly [string, number, readonly UsageRecord[]][])
       mtimeMs,
       provider: "claude",
       records,
+      malformedRecords: 0,
       tailRecords: [],
       position: position(),
     });
@@ -67,6 +68,7 @@ describe("scan cache round trip", () => {
       records: [
         record({ provider: "grok", model: "grok-4.5-build", dedupeKey: "s:p:grok-4.5-build" }),
       ],
+      malformedRecords: 0,
       tailRecords: [record({ provider: "grok", model: "grok-4.5-build", dedupeKey: null })],
       position: position({ resumeOffset: 30, guardLength: 30, guardHash: 123 }),
     });
@@ -75,11 +77,21 @@ describe("scan cache round trip", () => {
       mtimeMs: 400,
       provider: "codex",
       records: [record({ provider: "codex", model: "gpt-5.2-codex", dedupeKey: null })],
+      malformedRecords: 2,
       tailRecords: [],
       position: position({
         codexState: {
           model: "gpt-5.2-codex",
           sessionId: "session-c",
+          malformedRecords: 2,
+          lastCumulativeUsage: {
+            input_tokens: 10,
+            cached_input_tokens: 0,
+            cache_write_input_tokens: 0,
+            output_tokens: 5,
+            reasoning_output_tokens: 0,
+            total_tokens: 15,
+          },
           lastUsageSignature: '{"input_tokens":1}',
           sawSessionMeta: true,
           suppressingForkCopies: false,
@@ -111,6 +123,33 @@ describe("scan cache round trip", () => {
     expect(decodeScanCache(JSON.parse(JSON.stringify(poisoned))).has("/a.jsonl")).toBe(false);
   });
 
+  it.each([
+    { malformedRecords: -1 },
+    { lastCumulativeUsage: { input_tokens: 10, output_tokens: 5 } },
+    {
+      lastCumulativeUsage: {
+        input_tokens: 10,
+        output_tokens: 5,
+        cached_input_tokens: 0,
+        cache_write_input_tokens: 0,
+        reasoning_output_tokens: 6,
+        total_tokens: 15,
+      },
+    },
+  ])("discards corrupt counter state so the transcript is read again %#", (overrides) => {
+    const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]]));
+    const poisoned = {
+      ...encoded,
+      files: {
+        "/a.jsonl": {
+          ...encoded.files["/a.jsonl"]!,
+          cs: { ...initialCodexScanState(), ...overrides },
+        },
+      },
+    };
+    expect(decodeScanCache(JSON.parse(JSON.stringify(poisoned))).size).toBe(0);
+  });
+
   it("drops an entry whose guard length is outside the supported range", () => {
     // The guard length sizes a Buffer in the reader; a bogus value would make
     // every parse of that file fail and silently drop its usage.
@@ -125,7 +164,7 @@ describe("scan cache round trip", () => {
 
   it("rejects a document from the previous cache version", () => {
     const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]]));
-    const previous = { ...encoded, version: 2 };
+    const previous = { ...encoded, version: 3 };
 
     expect(decodeScanCache(JSON.parse(JSON.stringify(previous))).size).toBe(0);
   });

apps/server/src/usage/usageScanCache.ts

@@ -20,13 +20,14 @@ import * as NodePath from "node:path";
 import type { UsageProviderKind } from "@t3tools/contracts";
 
 import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";
-import type { CodexScanState, UsageRecord } from "./usageTranscripts.ts";
+import { readCodexTokenUsage, type CodexScanState, type UsageRecord } from "./usageTranscripts.ts";
 
 // v2: Codex fork-copy suppression changed what a file parses to, so v1
 // entries would keep serving double-counted records forever.
 // v3: entries carry the parse position and reducer state so a grown file
 // re-parses only its appended bytes instead of starting over.
-const USAGE_SCAN_CACHE_VERSION = 3 as const;
+// v4: reconcile Codex cumulative counters and retain malformed-record counts.
+const USAGE_SCAN_CACHE_VERSION = 4 as const;
 
 export interface CachedFile {
   readonly size: number;
@@ -40,6 +41,7 @@ export interface CachedFile {
    * re-reads that segment and would otherwise double count it.
    */
   readonly tailRecords: readonly UsageRecord[];
+  readonly malformedRecords: number;
   readonly position: TranscriptParsePosition;
 }
 
@@ -70,6 +72,7 @@ interface SerializedFile {
   readonly r: readonly SerializedRecord[];
   /** Tail records; see `CachedFile.tailRecords`. */
   readonly t: readonly SerializedRecord[];
+  readonly mc: number;
   /** Parse position: resume offset, guard length, guard hash. */
   readonly o: number;
   readonly gl: number;
@@ -122,6 +125,7 @@ export function encodeScanCache(cache: ScanCache): SerializedCache {
       p: entry.provider,
       r: entry.records.map(serializeRecord),
       t: entry.tailRecords.map(serializeRecord),
+      mc: entry.malformedRecords,
       o: entry.position.resumeOffset,
       gl: entry.position.guardLength,
       gh: entry.position.guardHash,
@@ -239,6 +243,7 @@ export function decodeScanCache(document: unknown): ScanCache {
     ) {
       continue;
     }
+    if (typeof entry.mc !== "number" || !Number.isSafeInteger(entry.mc) || entry.mc < 0) continue;
     const codexState = decodeCodexState(entry.cs);
     if (codexState === undefined) continue;
 
@@ -253,6 +258,7 @@ export function decodeScanCache(document: unknown): ScanCache {
       provider,
       records,
       tailRecords,
+      malformedRecords: entry.mc,
       position: {
         resumeOffset: entry.o,
         guardLength: entry.gl,
@@ -281,11 +287,25 @@ function decodeCodexState(value: unknown): CodexScanState | null | undefined {
     typeof state.sawSessionMeta !== "boolean" ||
     typeof state.suppressingForkCopies !== "boolean" ||
     typeof state.forkCopyAnchorMs !== "number" ||
-    !Number.isFinite(state.forkCopyAnchorMs)
+    !Number.isFinite(state.forkCopyAnchorMs) ||
+    typeof state.malformedRecords !== "number" ||
+    !Number.isSafeInteger(state.malformedRecords) ||
+    state.malformedRecords < 0
   ) {
     return undefined;
   }
+  const cumulative = readCodexTokenUsage(state.lastCumulativeUsage);
+  if (
+    state.lastCumulativeUsage !== null &&
+    (cumulative === null ||
+      Object.entries(cumulative).some(
+        ([key, count]) => (state.lastCumulativeUsage as Record<string, unknown>)[key] !== count,
+      ))
+  )
+    return undefined;
   return {
+    lastCumulativeUsage: cumulative,
+    malformedRecords: state.malformedRecords,
     model: state.model,
     sessionId: state.sessionId,
     lastUsageSignature: state.lastUsageSignature ?? null,