Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 129 additions & 0 deletions apps/server/src/provider/Drivers/codexUsage.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { describe, expect, it } from "@effect/vitest";
import { totalTokens } from "@t3tools/provider-core/server/usage";

import { codexUsageFormat, initialCodexScanState, parseCodexLine } from "./codexUsage.ts";

Expand Down Expand Up @@ -373,6 +374,106 @@ describe("parseCodexLine", () => {
expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(0);
});

it("ignores a replay of the current sequence's first request without rewinding the baseline", () => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
const request = counters(100, 20);
const processed = [
counted(request, request),
counted(request, counters(200, 40)),
counted(request, request),
counted(request, counters(300, 60)),
].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, 120], skipped: 0 });
});

it("bills a genuine restart with the same counters at a new timestamp", () => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
const request = counters(100, 20);
const lines = [
counted(request, request),
counted(request, counters(200, 40)),
counted(request, request).replace("05:17:49.919Z", "05:18:49.919Z"),
counted(counters(30, 5), counters(130, 25)),
];
const processed = lines.map((line) => {
const record = parseCodexLine(line, state);
return record === null ? 0 : totalTokens(record.totals);
});
expect({ processed, skipped: codexUsageFormat.readMalformedRecords?.(state) }).toEqual({
processed: [120, 120, 120, 35],
skipped: 0,
});
});

it.each([
{
reset: true,
markerTotal: counters(0, 0),
nextTotal: counters(130, 25),
processed: [120, 120, 0, 120, 35],
},
{
reset: false,
markerTotal: counters(200, 40),
nextTotal: counters(230, 45),
processed: [120, 120, 0, 0, 35],
},
])(
"bills a request equal to the old sequence start only after a context marker reset the counters (reset: $reset)",
({ markerTotal, nextTotal, processed }) => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
const request = counters(100, 20);
const marker = counted({ ...counters(0, 0), total_tokens: 26387 }, markerTotal, {
model_context_window: 475000,
});
const billed = [
counted(request, request),
counted(request, counters(200, 40)),
marker,
counted(request, request),
counted(counters(30, 5), nextTotal),
].map((line) => {
const record = parseCodexLine(line, state);
return record === null ? 0 : totalTokens(record.totals);
});

expect({
processed: billed,
skipped: codexUsageFormat.readMalformedRecords?.(state),
}).toEqual({ processed, skipped: 0 });
},
);

it("forgets the old sequence start when a malformed pair zeroes the cumulative baseline", () => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
const request = counters(100, 20);
const processed = [
counted(request, request),
counted(counters(30, 5), counters(0, 0)),
counted(request, request),
counted(counters(30, 5), counters(130, 25)),
].map((line) => {
const record = parseCodexLine(line, state);
return record === null ? 0 : totalTokens(record.totals);
});

expect({
processed,
skipped: codexUsageFormat.readMalformedRecords?.(state),
}).toEqual({ processed: [120, 0, 120, 35], skipped: 1 });
});

it("retains initial tier evidence only for the own OpenAI session and exact context until native settings arrive", () => {
const meta = (id: string, model_provider?: string) =>
JSON.stringify({ type: "session_meta", payload: { id, model_provider } });
Expand Down Expand Up @@ -535,5 +636,33 @@ describe("parseCodexLine", () => {
);
expect(record).not.toBeNull();
});

it("forgets the old sequence start when a copied zero baseline rebases the counters", () => {
const state = initialCodexScanState();
const forkInstant = "2026-08-01T05:00:00.000Z";
const request = counters(100, 20);
const billed = (line: string) => {
const record = parseCodexLine(line, state);
return record === null ? 0 : totalTokens(record.totals);
};
parseCodexLine(stamped("2026-08-01T04:59:00.000Z", turnContext), state);
const first = billed(stamped("2026-08-01T04:59:01.000Z", counted(request, request)));
parseCodexLine(meta({ id: "child", timestamp: forkInstant, forkedFromId: "parent" }), state);
const copied = billed(
stamped(
"2026-08-01T05:00:00.001Z",
counted({ ...counters(0, 0), total_tokens: 26387 }, counters(0, 0)),
),
);
const afterBurst = billed(stamped("2026-08-01T05:00:06.000Z", counted(request, request)));
const next = billed(
stamped("2026-08-01T05:00:06.100Z", counted(counters(30, 5), counters(130, 25))),
);

expect({
processed: [first, copied, afterBurst, next],
skipped: codexUsageFormat.readMalformedRecords?.(state),
}).toEqual({ processed: [120, 0, 120, 35], skipped: 0 });
});
});
});
19 changes: 17 additions & 2 deletions apps/server/src/provider/Drivers/codexUsage.ts
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,12 @@ function readCodexTokenUsage(value: unknown): CodexTokenUsage | null {
return isCodexTokenUsage(usage) ? usage : null;
}

// A zeroed baseline rebases the counters, so the sequence start measured against it is forgotten.
function rebaseCodexUsage(state: CodexScanState, cumulative: CodexTokenUsage) {
state.lastCumulativeUsage = cumulative;
if (cumulative.input_tokens + cumulative.output_tokens === 0) delete state.sequenceStartSignature;
}

Comment thread
coderabbitai[bot] marked this conversation as resolved.
function advanceCodexUsage(previous: CodexTokenUsage | null, last: CodexTokenUsage) {
return readCodexTokenUsage(
Object.fromEntries(CODEX_USAGE_FIELDS.map((key) => [key, (previous?.[key] ?? 0) + last[key]])),
Expand All @@ -71,6 +77,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.
sequenceStartSignature: Schema.mutableKey(Schema.optionalKey(Schema.String)),
malformedRecords: Schema.mutableKey(TokenCount),
sawSessionMeta: Schema.mutableKey(Schema.Boolean),
modelProvider: Schema.mutableKey(Schema.String),
Expand Down Expand Up @@ -193,7 +201,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord |
state.forkCopyAnchorMs = 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) state.lastCumulativeUsage = cumulative;
if (cumulative !== null) rebaseCodexUsage(state, cumulative);
else if (!hasCumulative && last !== null && signature !== state.lastUsageSignature) {
state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, last);
}
Expand All @@ -206,7 +214,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord |

const previous = state.lastCumulativeUsage;
// A valid snapshot remains the next baseline even when its paired request is malformed.
if (cumulative !== null) state.lastCumulativeUsage = cumulative;
if (cumulative !== null) rebaseCodexUsage(state, cumulative);
if (last === null || (hasCumulative && cumulative === null)) {
state.malformedRecords += 1;
return null;
Expand Down Expand Up @@ -236,6 +244,12 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord |
if (cumulative !== null) {
// A session resumed in a new Codex process restarts its cumulative counters at this request.
const restarted = CODEX_USAGE_FIELDS.every((key) => cumulative[key] === last[key]);
const sequenceStart = JSON.stringify([timestampMs, last]);
// Only the same event is a replay; a later request can restart with identical counters.
if (restarted && sequenceStart === state.sequenceStartSignature) {
state.lastCumulativeUsage = previous;
return null;
}
// Forks and paginated continuations can inherit totals without copying a counter snapshot.
// Only bill the latest request, and require the unseen prefix to be valid usage too.
const reconciled =
Expand All @@ -248,6 +262,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord |
state.malformedRecords += 1;
return null;
}
if (restarted) state.sequenceStartSignature = sequenceStart;
} else {
if (signature === state.lastUsageSignature) return null;
// Advance expected totals for billed legacy requests so a cumulative re-emission is stale.
Expand Down
Loading