Skip to content
Merged
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
117 changes: 117 additions & 0 deletions apps/server/src/provider/Drivers/codexUsage.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,123 @@ describe("parseCodexLine", () => {
}
});

it.each(["continuation", "fork", "subagent"])(
"bills only the latest request when a %s has no initial counter baseline",
(source) => {
const state = initialCodexScanState();
parseCodexLine(
JSON.stringify({
type: "session_meta",
timestamp: "2026-08-01T05:17:41.000Z",
payload: {
id: "child",
model_provider: "openai",
...(source === "fork" ? { forked_from_id: "parent" } : {}),
...(source === "subagent"
? { source: { subagent: { thread_spawn: { parent_thread_id: "parent" } } } }
: { source: "vscode" }),
},
}),
state,
);
parseCodexLine(turnContext, state);
const request = {
...counters(100, 20),
cached_input_tokens: 30,
cache_write_input_tokens: 10,
reasoning_output_tokens: 5,
};
const cumulative = {
...counters(500, 60),
cached_input_tokens: 200,
cache_write_input_tokens: 20,
reasoning_output_tokens: 15,
};
const first = counted(request, cumulative);
expect(parseCodexLine(first, state)).toMatchObject({
sessionId: "child",
totals: {
uncachedInputTokens: 60,
cachedInputTokens: 30,
cacheCreationTokens: 10,
outputTokens: 20,
reasoningTokens: 5,
},
});
expect(parseCodexLine(first, state)).toBeNull();
expect(parseCodexLine(counted(counters(10, 2), cumulative), state)).toBeNull();
expect(
parseCodexLine(
counted(counters(10, 2), {
...cumulative,
input_tokens: 510,
output_tokens: 62,
total_tokens: 572,
}),
state,
)?.totals.outputTokens,
).toBe(2);
expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(0);
},
);

it.each([
{
...counters(90, 40),
cached_input_tokens: 60,
cache_write_input_tokens: 20,
reasoning_output_tokens: 10,
},
{
...counters(200, 10),
cached_input_tokens: 60,
cache_write_input_tokens: 20,
reasoning_output_tokens: 5,
},
{
...counters(200, 40),
cached_input_tokens: 20,
cache_write_input_tokens: 20,
reasoning_output_tokens: 10,
},
{
...counters(200, 40),
cached_input_tokens: 60,
cache_write_input_tokens: 5,
reasoning_output_tokens: 10,
},
{
...counters(200, 40),
cached_input_tokens: 60,
cache_write_input_tokens: 20,
reasoning_output_tokens: 3,
},
{
...counters(110, 40),
cached_input_tokens: 60,
cache_write_input_tokens: 20,
reasoning_output_tokens: 10,
},
{
...counters(200, 30),
cached_input_tokens: 60,
cache_write_input_tokens: 20,
reasoning_output_tokens: 20,
},
])("rejects an initial snapshot with an impossible inherited baseline: %j", (cumulative) => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
const request = {
...counters(100, 20),
cached_input_tokens: 30,
cache_write_input_tokens: 10,
reasoning_output_tokens: 5,
};
expect(parseCodexLine(counted(request, cumulative), state)).toBeNull();
expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(1);
expect(state.lastCumulativeUsage).toEqual(cumulative);
});

it.each(["request", "cumulative"])("skips malformed %s counters and recovers", (malformed) => {
const state = initialCodexScanState();
parseCodexLine(turnContext, state);
Expand Down
13 changes: 9 additions & 4 deletions apps/server/src/provider/Drivers/codexUsage.ts
Original file line number Diff line number Diff line change
Expand Up @@ -243,10 +243,15 @@ 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]);
if (
!restarted &&
!CODEX_USAGE_FIELDS.every((key) => cumulative[key] - (previous?.[key] ?? 0) === last[key])
) {
// 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 =
previous === null
? readCodexTokenUsage(
Object.fromEntries(CODEX_USAGE_FIELDS.map((key) => [key, cumulative[key] - last[key]])),
) !== null
: CODEX_USAGE_FIELDS.every((key) => cumulative[key] - previous[key] === last[key]);
if (!restarted && !reconciled) {
state.malformedRecords += 1;
return null;
}
Expand Down
19 changes: 14 additions & 5 deletions apps/server/src/usage/UsageService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1881,7 +1881,7 @@ describe("UsageService", () => {
}).pipe(Effect.scoped),
);

it.live.each([4, 5, 6, 7])(
it.live.each([4, 5, 6, 7, 8])(
"upgrades a v%s cache: reparses live Codex usage, keeps deleted rollouts and the old file",
(version) =>
Effect.gen(function* () {
Expand All @@ -1903,7 +1903,10 @@ describe("UsageService", () => {
timestamp: "2026-08-01T10:00:00Z",
payload: {
type: "token_count",
info: { last_token_usage: { input_tokens: 0, output_tokens: outputTokens } },
info: {
last_token_usage: { input_tokens: 0, output_tokens: outputTokens },
total_token_usage: { input_tokens: 100, output_tokens: outputTokens + 30 },
},
},
},
]
Expand All @@ -1929,13 +1932,14 @@ describe("UsageService", () => {
yield* first.awaitPersisted;

const legacy = yield* Effect.promise(async () => {
const livePath = await NodeFSP.realpath(live);
const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as {
files: Record<string, { r: unknown[][]; cs: Record<string, unknown> }>;
files: Record<string, { r: unknown[][]; mr: number; cs: Record<string, unknown> }>;
};
for (const [path, file] of Object.entries(document.files)) {
file.r = file.r.map((row) => [
...row.slice(0, 6),
path === live ? 99 : row[6],
path === livePath ? 99 : row[6],
...row.slice(7, 10),
version === 4 ? 0 : 1,
]);
Expand All @@ -1944,6 +1948,11 @@ describe("UsageService", () => {
delete file.cs.lastCumulativeUsage;
delete file.cs.malformedRecords;
}
if (version === 8 && path === livePath) {
file.r = [];
file.mr = 1;
file.cs.malformedRecords = 1;
}
}
const text = encodeUnknownJsonString({ ...document, version });
await NodeFSP.rm(cachePath);
Expand Down Expand Up @@ -1971,7 +1980,7 @@ describe("UsageService", () => {
60 + 20 * (version === 4 ? 1 : 2),
);
yield* upgraded.awaitPersisted;
assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v8.json");
assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v9.json");
assert.strictEqual(
yield* Effect.promise(() => NodeFSP.readFile(legacyPath, "utf8")),
legacy,
Expand Down
17 changes: 16 additions & 1 deletion apps/server/src/usage/usageScanCache.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -162,13 +162,14 @@ describe("scan cache round trip", () => {
expect(restored).toEqual(original);
});

it.each([5, 6, 7])(
it.each([5, 6, 7, 8])(
"retains v%s history and compatible positions but invalidates obsolete state",
(version) => {
const original = cacheWith([["/gone.jsonl", 100, [record({ speed: "fast" })]]]);
original.set("/gone-codex.jsonl", {
...original.get("/gone.jsonl")!,
provider: "codex",
malformedRecords: 1,
records: [record({ provider: "codex", speed: "fast", dedupeKey: null })],
position: position({
state: {
Expand All @@ -191,6 +192,11 @@ describe("scan cache round trip", () => {
malformedRecords: 3,
position: position({ state: { sessionId: "session-a" } }),
});
const cleanCodex = {
...original.get("/gone-codex.jsonl")!,
malformedRecords: 0,
};
original.set("/clean-codex.jsonl", cleanCodex);
const encoded = encodeScanCache(original);
const entry = encoded.files["/gone.jsonl"]!;
const legacy = {
Expand Down Expand Up @@ -219,6 +225,15 @@ describe("scan cache round trip", () => {
size: -1,
position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null },
});
expect(restored.get("/clean-codex.jsonl")).toEqual(
version >= 8
? cleanCodex
: {
...cleanCodex,
size: -1,
position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null },
},
);
expect(restored.get("/gone-pi.jsonl")).toEqual(
version >= 6
? original.get("/gone-pi.jsonl")
Expand Down
11 changes: 7 additions & 4 deletions apps/server/src/usage/usageScanCache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,17 @@ import * as Schema from "effect/Schema";
import { CLAUDE_SYNTHETIC_MODEL } from "../provider/Drivers/claudeUsage.ts";
import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";

const USAGE_SCAN_CACHE_VERSION = 8 as const;
const USAGE_SCAN_CACHE_VERSION = 9 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
* both.
*/
export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v8.json";
export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v9.json";
export const LEGACY_SCAN_CACHE_FILE_NAMES = [
"usage-scan-cache-v8.json",
"usage-scan-cache-v7.json",
"usage-scan-cache-v6.json",
"usage-scan-cache-v5.json",
Expand Down Expand Up @@ -311,9 +312,11 @@ export function decodeScanCache(
continue;
}
// Keep history for deleted rollouts, but re-parse incompatible reducer state.
// Older Codex entries lack provenance state (v7) or misread compaction and resume counters (v8).
// v9 repairs skipped initial Codex requests; clean v8 entries can still resume.
const legacy =
format.state !== undefined && (version < 6 || (entry.p === "codex" && version < 8));
format.state !== undefined &&
(version < 6 ||
(entry.p === "codex" && (version < 8 || (version < 9 && malformedRecords > 0))));
// 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;
Expand Down
59 changes: 59 additions & 0 deletions apps/server/src/usage/usageTranscriptReader.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,65 @@ describe("readTranscriptRecords resume", () => {
assert.strictEqual(second.tailRecords.length, 0);
});

it.each([
{ mode: "buffered", streamingThresholdBytes: Infinity },
{ mode: "streaming", streamingThresholdBytes: 0 },
])(
"resumes an initial Codex carried-total tail without rebilling it ($mode)",
async (options) => {
const path = NodePath.join(dir, "continuation.jsonl");
const usage = (input: number, output: number) => ({
input_tokens: input,
output_tokens: output,
});
const event = (last: unknown, cumulative: unknown) =>
JSON.stringify({
type: "event_msg",
timestamp: "2026-08-01T10:00:05Z",
payload: {
type: "token_count",
info: { last_token_usage: last, total_token_usage: cumulative },
},
});
const firstLine = event(usage(100, 20), usage(500, 60));
await NodeFSP.writeFile(
path,
codexMetaLine() + codexModelLine("gpt-6-astra") + codexTierLine("ultrafast") + firstLine,
);
const first = await readTranscriptRecords(path, TEST_FORMATS.codex, undefined, options);
assert.isNotNull(first);
assert.strictEqual(first.records.length, 0);
assert.deepStrictEqual(
first.tailRecords.map((r) => [r.totals.outputTokens, r.speed]),
[[20, "ultrafast"]],
);
assert.strictEqual(first.malformedRecords, 0);

await NodeFSP.appendFile(path, `\n${firstLine}\n${event(usage(10, 2), usage(510, 62))}\n`);
const resumed = await readTranscriptRecords(
path,
TEST_FORMATS.codex,
JSON.parse(JSON.stringify(first.position)),
options,
);
const full = await readTranscriptRecords(path, TEST_FORMATS.codex, undefined, options);
assert.isNotNull(resumed);
assert.isNotNull(full);
assert.isTrue(resumed.resumed);
assert.deepStrictEqual(resumed.records, full.records);
assert.deepStrictEqual(
resumed.records.map((r) => [r.totals.outputTokens, r.speed]),
[
[20, "ultrafast"],
[2, "ultrafast"],
],
);
assert.deepStrictEqual(resumed.tailRecords, []);
assert.strictEqual(resumed.malformedRecords, 0);
assert.deepStrictEqual(resumed.position, full.position);
},
);

it("re-parses from the start when the guard bytes no longer match", async () => {
const path = NodePath.join(dir, "claude.jsonl");
await NodeFSP.writeFile(path, claudeLine(1, 5));
Expand Down
Loading