fix(usage): bill the Codex request a compaction's counter rewind drops

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

@@ -59,6 +59,20 @@ describe("parseCodexLine", () => {
         info: { last_token_usage: last, total_token_usage: total, ...extra },
       },
     });
+  const usage = (input: number, cached: number, output: number, reasoning = 0) => ({
+    input_tokens: input,
+    cached_input_tokens: cached,
+    cache_write_input_tokens: 0,
+    output_tokens: output,
+    reasoning_output_tokens: reasoning,
+    total_tokens: input + output,
+  });
+  // A snapshot from a session whose compaction rewound the cumulative counters.
+  const rewound = (
+    timestamp: string,
+    last: readonly [number, number, number, number],
+    total: readonly [number, number, number, number],
+  ) => counted(usage(...last), usage(...total)).replace("2026-08-01T05:17:49.919Z", timestamp);
 
   it("attributes usage to the model from the preceding turn context", () => {
     const state = initialCodexScanState();
@@ -573,6 +587,27 @@ describe("parseCodexLine", () => {
     }).toEqual({ processed: [120, 120, 0, 120], skipped: 0 });
   });
 
+  it("keeps the baseline's timestamp when it ignores a restart replay", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    const request = counters(100, 20);
+    const at = (line: string, time: string) => line.replace("05:17:49.919Z", time);
+    const processed = [
+      at(counted(request, request), "05:17:00.000Z"),
+      at(counted(request, counters(200, 40)), "05:18:00.000Z"),
+      at(counted(request, request), "05:17:00.000Z"),
+      // An older snapshot is not a rewind against the 05:18 baseline.
+      at(counted(counters(50, 10), counters(150, 30)), "05:17:30.000Z"),
+    ].map((line) => {
+      const record = parseCodexLine(line, state);
+      return record === null ? 0 : totalTokens(record.totals);
+    });
+    expect({ processed, skipped: codexUsageFormat.readMalformedRecords?.(state) }).toEqual({
+      processed: [120, 120, 0, 0],
+      skipped: 1,
+    });
+  });
+
   it("bills a genuine restart with the same counters at a new timestamp", () => {
     const state = initialCodexScanState();
     parseCodexLine(turnContext, state);
@@ -593,6 +628,169 @@ describe("parseCodexLine", () => {
     });
   });
 
+  // Both records below are taken from sessions that dropped one request each after compaction.
+  const compactedSep06 = [
+    // 2026-09-06T20:38:24.552Z: 4,280,396 tokens before the session compacted.
+    rewound("2026-09-06T20:38:24.552Z", [73869, 72448, 456, 13], [4252914, 3953024, 27482, 2045]),
+    // 2026-09-06T21:10:04.346Z: the snapshot rewinds to 3,621,978 and still counts this request.
+    rewound("2026-09-06T21:10:04.346Z", [77213, 74240, 121, 0], [3597853, 3370880, 24125, 1750]),
+    // 2026-09-06T21:10:15.589Z: the next snapshot reconciles against the rewound baseline.
+    rewound("2026-09-06T21:10:15.589Z", [77371, 77056, 202, 0], [3675224, 3447936, 24327, 1750]),
+  ];
+
+  it("bills the request a compaction's counter rewind reports", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    const processed = compactedSep06.map((line) => {
+      const record = parseCodexLine(line, state);
+      return record === null ? 0 : totalTokens(record.totals);
+    });
+
+    expect({
+      processed,
+      skipped: codexUsageFormat.readMalformedRecords?.(state),
+    }).toEqual({ processed: [74325, 77334, 77573], skipped: 0 });
+  });
+
+  // A later turn continues from an earlier snapshot in the same rollout.
+  it("bills a request that rewinds to a snapshot from a previous turn", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    const lines = [
+      ["2026-11-02T10:00:01.000Z", counters(1000, 200), counters(1000, 200)],
+      ["2026-11-02T10:00:02.000Z", counters(100, 20), counters(1100, 220)],
+      ["2026-11-02T10:00:03.000Z", counters(300, 50), counters(1400, 270)],
+      // Continues from the 10:00:02 snapshot: 1,560 cumulative minus 240 reports 1,320.
+      ["2026-11-02T12:00:00.000Z", counters(200, 40), counters(1300, 260)],
+    ] as const;
+    const processed = lines.map(([timestamp, last, total]) => {
+      const record = parseCodexLine(
+        counted(last, total).replace("2026-08-01T05:17:49.919Z", timestamp),
+        state,
+      );
+      return record === null ? 0 : totalTokens(record.totals);
+    });
+
+    expect({
+      processed,
+      skipped: codexUsageFormat.readMalformedRecords?.(state),
+    }).toEqual({ processed: [1200, 120, 350, 240], skipped: 0 });
+  });
+
+  it("bills a rewound request only when its snapshot contains it", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    parseCodexLine(counted(counters(100, 20), counters(100, 20)), state);
+    // The snapshot is behind the request it reports, so it cannot be a baseline for that request.
+    expect(parseCodexLine(counted(counters(30, 5), counters(10, 2)), state)).toBeNull();
+    expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(1);
+  });
+
+  it("does not bill a stale snapshot whose timestamp goes back", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    parseCodexLine(counted(counters(100, 20), counters(100, 20)), state);
+    // A replayed request from before the current baseline, not a rewind.
+    expect(
+      parseCodexLine(
+        counted(counters(30, 5), counters(50, 10)).replace(
+          "2026-08-01T05:17:49.919Z",
+          "2026-08-01T05:16:49.919Z",
+        ),
+        state,
+      ),
+    ).toBeNull();
+    expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(1);
+    expect(
+      parseCodexLine(counted(counters(10, 2), counters(60, 12)), state)?.totals.outputTokens,
+    ).toBe(2);
+  });
+
+  it.each([
+    {
+      name: "a rejected stale snapshot",
+      lines: [
+        ["10:00:10", counters(100, 20), counters(100, 20)],
+        ["10:00:08", counters(20, 4), counters(50, 10)],
+        ["10:00:09", counters(10, 2), counters(30, 6)],
+      ],
+      processed: [120, 0, 0],
+    },
+    {
+      name: "a billed last-only request",
+      lines: [
+        ["10:00:00", counters(100, 20), counters(100, 20)],
+        ["10:00:05", counters(20, 4), undefined],
+        ["10:00:10", counters(30, 6), undefined],
+        ["10:00:05", counters(20, 4), counters(120, 24)],
+      ],
+      processed: [120, 24, 36, 0],
+    },
+  ])("keeps the rewind guard's timestamp from going back after $name", ({ lines, processed }) => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    expect(
+      lines.map(([time, last, total]) => {
+        const record = parseCodexLine(
+          counted(last, total).replace("2026-08-01T05:17:49.919Z", `2026-08-01T${time}.000Z`),
+          state,
+        );
+        return record === null ? 0 : totalTokens(record.totals);
+      }),
+    ).toEqual(processed);
+  });
+
+  it("does not treat a rewind as ordered when a cached state has no baseline timestamp", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    const at = (line: string, time: string) =>
+      line.replace("2026-08-01T05:17:49.919Z", `2026-08-01T${time}.000Z`);
+    const billed = (line: string) => {
+      const record = parseCodexLine(line, state);
+      return record === null ? 0 : totalTokens(record.totals);
+    };
+    const start = at(counted(counters(100, 20), counters(100, 20)), "10:00:00");
+    const second = at(counted(counters(20, 4), counters(120, 24)), "10:00:05");
+    const third = at(counted(counters(30, 6), counters(150, 30)), "10:00:10");
+    expect([start, second, third].map(billed)).toEqual([120, 24, 36]);
+    // Scan states cached before v15 resume without the timestamp.
+    delete state.lastCumulativeTimestampMs;
+
+    const replayedStart = billed(start);
+    const hasTimestampAfterStart = "lastCumulativeTimestampMs" in state;
+    const replayedSecond = billed(second);
+    expect({ replayed: [replayedStart, replayedSecond], hasTimestampAfterStart }).toEqual({
+      replayed: [0, 0],
+      hasTimestampAfterStart: false,
+    });
+  });
+
+  it("warns about pending anonymous requests a counter rewind discards", () => {
+    const state = initialCodexScanState();
+    parseCodexLine(turnContext, state);
+    const at = (line: string, time: string) =>
+      line.replace("2026-08-01T05:17:49.919Z", `2026-08-01T${time}.000Z`);
+    const request = (time: string, usage: unknown) =>
+      JSON.stringify({
+        type: "token_usage_record",
+        timestamp: `2026-08-01T${time}.000Z`,
+        payload: { usage },
+      });
+    const billed = [
+      at(counted(counters(100, 20), counters(100, 20)), "10:00:00"),
+      request("10:00:05", counters(40, 8)),
+      request("10:00:06", counters(20, 4)),
+      // Rewinds below the 120-token baseline and reports only the 10:00:06 request.
+      at(counted(counters(20, 4), counters(60, 12)), "10:00:07"),
+      at(counted(counters(10, 2), counters(70, 14)), "10:00:08"),
+    ].flatMap((line) => parseCodexRecords(line, state));
+
+    expect({
+      processed: billed.map((record) => totalTokens(record.totals)),
+      skipped: codexUsageFormat.readMalformedRecords?.(state),
+    }).toEqual({ processed: [120, 24, 12], skipped: 1 });
+  });
+
   it.each([
     {
       reset: true,

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

 if (restarted && sequenceStart === state.sequenceStartSignature) { 
 if (restarted && sequenceStart === state.sequenceStartSignature) { 
@@ -54,9 +54,18 @@ function readCodexTokenUsage(value: unknown): CodexTokenUsage | null {
   return isCodexTokenUsage(usage) ? usage : null;
 }
 
+// The rewind guard compares against the newest counters seen, so an older record never lowers it.
+function advanceCodexTimestamp(state: CodexScanState, timestampMs: number) {
+  state.lastCumulativeTimestampMs = Math.max(
+    state.lastCumulativeTimestampMs ?? timestampMs,
+    timestampMs,
+  );
+}
+
 // A zeroed baseline rebases the counters, so the sequence start measured against it is forgotten.
-function rebaseCodexUsage(state: CodexScanState, cumulative: CodexTokenUsage) {
+function rebaseCodexUsage(state: CodexScanState, cumulative: CodexTokenUsage, timestampMs: number) {
   state.lastCumulativeUsage = cumulative;
+  advanceCodexTimestamp(state, timestampMs);
   delete state.unreportedRequests;
   if (cumulative.input_tokens + cumulative.output_tokens === 0) delete state.sequenceStartSignature;
 }
@@ -67,6 +76,26 @@ function advanceCodexUsage(previous: CodexTokenUsage | null, last: CodexTokenUsa
   );
 }
 
+/** A rewind is a new baseline when the timestamp held and the snapshot still contains the request it
+ * reports; stale and out-of-order records fail that test, as does a baseline of unknown age (scan
+ * states cached before v15). A delayed replay of an older snapshot with a fresh timestamp is
+ * indistinguishable from a rewind in the fields Codex writes. */
+function isRewoundBaseline(
+  previous: CodexTokenUsage | null,
+  previousTimestampMs: number | undefined,
+  cumulative: CodexTokenUsage,
+  last: CodexTokenUsage,
+  timestampMs: number,
+) {
+  return (
+    previous !== null &&
+    previousTimestampMs !== undefined &&
+    timestampMs >= previousTimestampMs &&
+    cumulative.total_tokens < previous.total_tokens &&
+    CODEX_USAGE_FIELDS.every((key) => cumulative[key] >= last[key])
+  );
+}
+
 const CodexSpeed = Schema.Literals(["standard", "fast", "ultrafast"]);
 
 // What a request is billed as is fixed when it is read, not when a later snapshot reconciles it.
@@ -111,6 +140,18 @@ function unreportedRequests(
   );
 }
 
+// A rewound snapshot accounts only for the request it reports. Earlier pending requests cannot be
+// reconciled against the rewound counters, so they stay unbilled.
+function discardedRequestCount(
+  pending: readonly CodexUnreportedRequest[] | undefined,
+  last: CodexTokenUsage,
+) {
+  const final = pending?.at(-1);
+  if (pending === undefined || final === undefined) return 0;
+  const reported = CODEX_USAGE_FIELDS.every((key) => final.usage[key] === last[key]);
+  return pending.length - (reported ? 1 : 0);
+}
+
 /**
  * Codex `token_count` events carry no model or service tier, so both are carried forward: the model
  * from the most recent `turn_context`, the tier from the most recent `thread_settings_applied`.
@@ -121,7 +162,8 @@ export const CodexScanState = Schema.Struct({
   sessionId: Schema.mutableKey(Schema.String),
   lastUsageSignature: Schema.mutableKey(Schema.NullOr(Schema.String)),
   lastCumulativeUsage: Schema.mutableKey(Schema.NullOr(CodexTokenUsage)),
-  // Optional so scan states cached before this field existed stay valid.
+  // Optional so scan states cached before these fields existed stay valid.
+  lastCumulativeTimestampMs: Schema.mutableKey(Schema.optionalKey(Schema.Finite)),
   sequenceStartSignature: Schema.mutableKey(Schema.optionalKey(Schema.String)),
   // `token_usage_record` usage since the last cumulative snapshot, one entry per record.
   unreportedRequests: Schema.mutableKey(
@@ -290,7 +332,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag
   if (isForkCopy(state, timestampMs)) {
     // Copied history is already billed in the parent, but establishes the child's baseline.
     const signature = last === null ? null : JSON.stringify(last);
-    if (cumulative !== null) rebaseCodexUsage(state, cumulative);
+    if (cumulative !== null) rebaseCodexUsage(state, cumulative, timestampMs);
     else if (!hasCumulative && last !== null && signature !== state.lastUsageSignature) {
       state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, last);
     }
@@ -300,14 +342,15 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag
 
   // Context snapshots can rewind on resume and omit compaction spending; identified requests are authoritative.
   if (state.requestLedgerActive) {
-    if (cumulative !== null) rebaseCodexUsage(state, cumulative);
+    if (cumulative !== null) rebaseCodexUsage(state, cumulative, timestampMs);
     return [];
   }
 
   const previous = state.lastCumulativeUsage;
+  const previousTimestampMs = state.lastCumulativeTimestampMs;
   const pending = state.unreportedRequests;
   // A valid snapshot remains the next baseline even when its paired request is malformed.
-  if (cumulative !== null) rebaseCodexUsage(state, cumulative);
+  if (cumulative !== null) rebaseCodexUsage(state, cumulative, timestampMs);
   if (last === null || (hasCumulative && cumulative === null)) {
     state.malformedRecords += 1;
     return [];
@@ -342,6 +385,9 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag
     // Only the same event is a replay; a later request can restart with identical counters.
     if (restarted && sequenceStart === state.sequenceStartSignature) {
       state.lastCumulativeUsage = previous;
+      // The replay's older timestamp would let a later stale snapshot pass as a rewind.
+      if (previousTimestampMs === undefined) delete state.lastCumulativeTimestampMs;
+      else state.lastCumulativeTimestampMs = previousTimestampMs;
       return [];
     }
     // Forks and paginated continuations can inherit totals without copying a counter snapshot.
@@ -353,12 +399,16 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag
           ) !== null
         : CODEX_USAGE_FIELDS.every((key) => cumulative[key] - previous[key] === last[key]);
     if (!restarted && !reconciled) {
-      const bridged = unreportedRequests(previous, cumulative, last, pending);
-      if (bridged === null) {
-        state.malformedRecords += 1;
-        return [];
+      if (isRewoundBaseline(previous, previousTimestampMs, cumulative, last, timestampMs)) {
+        state.malformedRecords += discardedRequestCount(pending, last);
+      } else {
+        const bridged = unreportedRequests(previous, cumulative, last, pending);
+        if (bridged === null) {
+          state.malformedRecords += 1;
+          return [];
+        }
+        unreported = bridged;
       }
-      unreported = bridged;
     }
     if (restarted) state.sequenceStartSignature = sequenceStart;
   } else {
@@ -370,6 +420,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly Usag
       return [];
     }
     state.lastCumulativeUsage = advanced;
+    advanceCodexTimestamp(state, timestampMs);
   }
   state.lastUsageSignature = signature;
 

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

@@ -2936,7 +2936,7 @@ describe("UsageService", () => {
             60 + 20 * (version === 4 ? 1 : 2),
           );
           yield* upgraded.awaitPersisted;
-          assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v14.json");
+          assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v15.json");
           assert.deepStrictEqual(
             (yield* Effect.promise(() => NodeFSP.readdir(stateDir)))
               .filter((name) => name.startsWith("usage-scan-cache"))