diff --git a/apps/server/src/provider/Drivers/codexUsage.test.ts b/apps/server/src/provider/Drivers/codexUsage.test.ts index 3820ce0cf5234..74812bafb5099 100644 --- a/apps/server/src/provider/Drivers/codexUsage.test.ts +++ b/apps/server/src/provider/Drivers/codexUsage.test.ts @@ -1,7 +1,18 @@ import { describe, expect, it } from "@effect/vitest"; import { totalTokens } from "@t3tools/provider-core/server/usage"; -import { codexUsageFormat, initialCodexScanState, parseCodexLine } from "./codexUsage.ts"; +import { + type CodexScanState, + codexUsageFormat, + initialCodexScanState, + parseCodexLine as parseCodexRecords, +} from "./codexUsage.ts"; + +function parseCodexLine(line: string, state: CodexScanState) { + const records = parseCodexRecords(line, state); + expect(records.length).toBeLessThanOrEqual(1); + return records[0] ?? null; +} describe("parseCodexLine", () => { const sessionMeta = JSON.stringify({ @@ -326,6 +337,119 @@ describe("parseCodexLine", () => { expect(codexUsageFormat.readMalformedRecords?.(state)).toBe(1); }); + it("bills only the reported request when usage records sit between reconciling snapshots", () => { + const state = initialCodexScanState(); + const usageRecord = (usage: unknown) => + JSON.stringify({ + type: "token_usage_record", + timestamp: "2026-08-01T05:17:45.000Z", + payload: { usage }, + }); + parseCodexLine(turnContext, state); + parseCodexLine(counted(counters(100, 20), counters(100, 20)), state); + for (const usage of [counters(40, 5), counters(60, 15), counters(30, 3)]) { + expect(parseCodexLine(usageRecord(usage), state)).toBeNull(); + } + expect( + parseCodexLine(counted(counters(100, 20), counters(200, 40)), state)?.totals.outputTokens, + ).toBe(20); + // Records explained by an earlier snapshot cannot bridge a later gap. + parseCodexLine(usageRecord(counters(10, 2)), state); + expect(parseCodexLine(counted(counters(10, 2), counters(340, 65)), state)).toBeNull(); + expect(state.malformedRecords).toBe(1); + }); + + it("bills each interrupted turn's request with the model, tier and time it was read under", () => { + const state = initialCodexScanState(); + const line = (type: string, time: string, payload: Record) => + JSON.stringify({ type, timestamp: `2026-11-0${time}Z`, payload }); + const context = (time: string, turn_id: string, model: string) => + line("turn_context", time, { turn_id, model }); + const tier = (time: string, service_tier: string) => + line("event_msg", time, { + type: "thread_settings_applied", + thread_settings: { service_tier }, + }); + const request = (time: string, usage: unknown) => line("token_usage_record", time, { usage }); + const snapshot = (time: string, last: unknown, total: unknown) => + line("event_msg", time, { + type: "token_count", + info: { last_token_usage: last, total_token_usage: total }, + }); + const parsed = [ + line("session_meta", "1T23:59:50", { id: "session-a", model_provider: "openai" }), + context("1T23:59:50", "turn-1", "gpt-6-sol"), + request("1T23:59:51", counters(100, 20)), + snapshot("1T23:59:52", counters(100, 20), counters(100, 20)), + // One interrupted turn whose requests straddle midnight. + request("1T23:59:59", counters(150, 10)), + request("2T00:00:01", counters(50, 5)), + tier("2T00:00:02", "priority"), + context("2T00:00:02", "turn-2", "gpt-6-astra"), + request("2T00:00:03", counters(50, 5)), + request("2T00:00:04", counters(30, 5)), + tier("2T00:00:05", "default"), + context("2T00:00:05", "turn-3", "gpt-6-sol"), + request("2T00:00:06", counters(200, 30)), + snapshot("2T00:00:07", counters(200, 30), counters(580, 75)), + ].flatMap((value) => parseCodexRecords(value, state)); + + const at = (time: string) => Date.parse(`2026-11-0${time}Z`); + expect( + parsed.map((record) => ({ + tokens: totalTokens(record.totals), + model: record.model, + speed: record.speed, + timestampMs: record.timestampMs, + unresolvedServiceTierTurnId: record.unresolvedServiceTierTurnId, + })), + ).toEqual([ + { + tokens: 120, + model: "gpt-6-sol", + speed: "standard", + timestampMs: at("1T23:59:52"), + unresolvedServiceTierTurnId: "turn-1", + }, + { + tokens: 160, + model: "gpt-6-sol", + speed: "standard", + timestampMs: at("1T23:59:59"), + unresolvedServiceTierTurnId: "turn-1", + }, + { + tokens: 55, + model: "gpt-6-sol", + speed: "standard", + timestampMs: at("2T00:00:01"), + unresolvedServiceTierTurnId: "turn-1", + }, + { + tokens: 55, + model: "gpt-6-astra", + speed: "fast", + timestampMs: at("2T00:00:03"), + unresolvedServiceTierTurnId: undefined, + }, + { + tokens: 35, + model: "gpt-6-astra", + speed: "fast", + timestampMs: at("2T00:00:04"), + unresolvedServiceTierTurnId: undefined, + }, + { + tokens: 230, + model: "gpt-6-sol", + speed: "standard", + timestampMs: at("2T00:00:07"), + unresolvedServiceTierTurnId: undefined, + }, + ]); + 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); @@ -664,5 +788,26 @@ describe("parseCodexLine", () => { skipped: codexUsageFormat.readMalformedRecords?.(state), }).toEqual({ processed: [120, 0, 120, 35], skipped: 0 }); }); + + it("bills the child's first request when its turn is interrupted before any snapshot", () => { + const state = initialCodexScanState(); + const forkInstant = "2026-08-01T05:00:00.000Z"; + const request = (timestamp: string, usage: unknown) => + JSON.stringify({ type: "token_usage_record", timestamp, payload: { usage } }); + const billed = [ + meta({ id: "child", timestamp: forkInstant, forkedFromId: "parent" }), + stamped(forkInstant, turnContext), + request("2026-08-01T05:00:00.001Z", counters(100, 20)), + stamped("2026-08-01T05:00:00.001Z", counted(counters(100, 20), counters(100, 20))), + request("2026-08-01T05:00:06.000Z", counters(150, 10)), + request("2026-08-01T05:00:10.000Z", counters(200, 30)), + stamped("2026-08-01T05:00:10.000Z", counted(counters(200, 30), counters(450, 60))), + ].flatMap((line) => parseCodexRecords(line, state)); + + expect({ + processed: billed.map((record) => totalTokens(record.totals)), + skipped: codexUsageFormat.readMalformedRecords?.(state), + }).toEqual({ processed: [160, 230], skipped: 0 }); + }); }); }); diff --git a/apps/server/src/provider/Drivers/codexUsage.ts b/apps/server/src/provider/Drivers/codexUsage.ts index 20939e54db799..90f7c78b849f8 100644 --- a/apps/server/src/provider/Drivers/codexUsage.ts +++ b/apps/server/src/provider/Drivers/codexUsage.ts @@ -5,7 +5,6 @@ import * as Schema from "effect/Schema"; import { parseTimestampMs, - totalTokens, type ProviderUsageReader, type TranscriptUsageFormat, type UsageRecord, @@ -58,6 +57,7 @@ function readCodexTokenUsage(value: unknown): CodexTokenUsage | 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; + delete state.unreportedRequests; if (cumulative.input_tokens + cumulative.output_tokens === 0) delete state.sequenceStartSignature; } @@ -67,18 +67,66 @@ function advanceCodexUsage(previous: CodexTokenUsage | null, last: CodexTokenUsa ); } +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. +const CodexUnreportedRequest = Schema.Struct({ + timestampMs: Schema.Finite, + model: Schema.String, + turnId: Schema.String, + speed: CodexSpeed, + sawThreadSettings: Schema.Boolean, + usage: CodexTokenUsage, +}); +type CodexUnreportedRequest = typeof CodexUnreportedRequest.Type; +type CodexRequestAttribution = Omit; + +// An interrupted turn can end before a snapshot reports its last request. Bill those requests only +// when the `token_usage_record`s since the previous snapshot explain the whole gap; `last` is the +// final one. +function unreportedRequests( + previous: CodexTokenUsage | null, + cumulative: CodexTokenUsage, + last: CodexTokenUsage, + pending: readonly CodexUnreportedRequest[] | undefined, +) { + if ( + previous === null || + pending === undefined || + CODEX_USAGE_FIELDS.some( + (key) => + cumulative[key] - previous[key] !== + pending.reduce((sum, request) => sum + request.usage[key], 0), + ) + ) { + return null; + } + const final = pending.at(-1)!; + const remainder = readCodexTokenUsage( + Object.fromEntries(CODEX_USAGE_FIELDS.map((key) => [key, final.usage[key] - last[key]])), + ); + if (remainder === null) return null; + return [...pending.slice(0, -1), { ...final, usage: remainder }].filter( + (request) => request.usage.input_tokens + request.usage.output_tokens > 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`. */ export const CodexScanState = Schema.Struct({ model: Schema.mutableKey(Schema.String), - speed: Schema.mutableKey(Schema.Literals(["standard", "fast", "ultrafast"])), + speed: Schema.mutableKey(CodexSpeed), 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)), + // `token_usage_record` usage since the last cumulative snapshot, one entry per record. + unreportedRequests: Schema.mutableKey( + Schema.optionalKey(Schema.mutable(Schema.Array(CodexUnreportedRequest))), + ), malformedRecords: Schema.mutableKey(TokenCount), sawSessionMeta: Schema.mutableKey(Schema.Boolean), modelProvider: Schema.mutableKey(Schema.String), @@ -120,32 +168,44 @@ function isForkedSessionMeta(payload: Record): boolean { return typeof (spawn as Record)["parent_thread_id"] === "string"; } +// The first usage line past the copied burst ends suppression for the rest of the rollout. +function isForkCopy(state: CodexScanState, timestampMs: number): boolean { + if (!state.suppressingForkCopies) return false; + if (timestampMs - state.forkCopyAnchorMs < FORK_COPY_MAX_GAP_MS) { + state.forkCopyAnchorMs = timestampMs; + return true; + } + state.suppressingForkCopies = false; + state.lastUsageSignature = null; + return false; +} + /** * Cumulative counters validate request deltas. Older events without cumulative counters use * normalized consecutive-request deduplication. */ -export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | null { +export function parseCodexLine(line: string, state: CodexScanState): readonly UsageRecord[] { let parsed: unknown; try { parsed = JSON.parse(line); } catch { - return null; + return []; } return parseCodexRecord(parsed, state); } -function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | null { - if (typeof parsed !== "object" || parsed === null) return null; +function parseCodexRecord(parsed: unknown, state: CodexScanState): readonly UsageRecord[] { + if (typeof parsed !== "object" || parsed === null) return []; const record = parsed as Record; const payload = record["payload"]; - if (typeof payload !== "object" || payload === null) return null; + if (typeof payload !== "object" || payload === null) return []; const payloadRecord = payload as Record; const payloadType = payloadRecord["type"]; if (record["type"] === "session_meta") { // Only the first meta identifies this session; copied ancestor metas would misattribute later records. - if (state.sawSessionMeta) return null; + if (state.sawSessionMeta) return []; state.sawSessionMeta = true; const id = payloadRecord["id"] ?? payloadRecord["session_id"]; if (typeof id === "string") state.sessionId = id; @@ -157,7 +217,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | state.suppressingForkCopies = true; state.forkCopyAnchorMs = metaTimestampMs; } - return null; + return []; } if (record["type"] === "turn_context") { @@ -169,7 +229,18 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | payloadRecord["model"].length > 0 ? payloadRecord["turn_id"] : ""; - return null; + return []; + } + + // Several of these may sit between two snapshots, so they only explain a gap in the counters. + 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 []; + (state.unreportedRequests ??= []).push({ ...codexAttribution(state, timestampMs), usage }); + return []; } if (payloadType === "thread_settings_applied") { @@ -178,53 +249,49 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | if (typeof settings === "object" && settings !== null) { state.speed = codexSpeed((settings as Record)["service_tier"]); } - return null; + return []; } - if (payloadType !== "token_count") return null; + if (payloadType !== "token_count") return []; const info = payloadRecord["info"]; - if (typeof info !== "object" || info === null) return null; + if (typeof info !== "object" || info === null) return []; const infoRecord = info as Record; // Codex may re-emit these counters after the model or timestamp becomes available. const timestampMs = parseTimestampMs(record["timestamp"]); - if (timestampMs === null || state.model.length === 0) return null; + if (timestampMs === null || state.model.length === 0) return []; const last = readCodexTokenUsage(infoRecord["last_token_usage"]); const rawCumulative = infoRecord["total_token_usage"]; const hasCumulative = rawCumulative !== undefined && rawCumulative !== null; const cumulative = hasCumulative ? readCodexTokenUsage(rawCumulative) : null; - if (state.suppressingForkCopies) { - if (timestampMs - state.forkCopyAnchorMs < FORK_COPY_MAX_GAP_MS) { - 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) rebaseCodexUsage(state, cumulative); - else if (!hasCumulative && last !== null && signature !== state.lastUsageSignature) { - state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, last); - } - state.lastUsageSignature = signature; - return null; + 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); + else if (!hasCumulative && last !== null && signature !== state.lastUsageSignature) { + state.lastCumulativeUsage = advanceCodexUsage(state.lastCumulativeUsage, last); } - state.suppressingForkCopies = false; - state.lastUsageSignature = null; + state.lastUsageSignature = signature; + return []; } const previous = state.lastCumulativeUsage; + 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 (last === null || (hasCumulative && cumulative === null)) { state.malformedRecords += 1; - return null; + return []; } // Zero-component counters report context size, not a request: fill_to_context_window, or the // post-compaction marker whose total is the new context size (its cumulative may reset to zero). if (last.input_tokens + last.output_tokens === 0) { state.lastUsageSignature = null; - return null; + return []; } if ( @@ -232,15 +299,16 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | previous !== null && CODEX_USAGE_FIELDS.every((key) => cumulative[key] === previous[key]) ) { - return null; + return []; } if (last.total_tokens !== last.input_tokens + last.output_tokens) { state.malformedRecords += 1; - return null; + return []; } const signature = JSON.stringify(last); + let unreported: readonly CodexUnreportedRequest[] = []; 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]); @@ -248,7 +316,7 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | // 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; + return []; } // 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. @@ -259,26 +327,51 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | ) !== null : CODEX_USAGE_FIELDS.every((key) => cumulative[key] - previous[key] === last[key]); if (!restarted && !reconciled) { - state.malformedRecords += 1; - return null; + const bridged = unreportedRequests(previous, cumulative, last, pending); + if (bridged === null) { + state.malformedRecords += 1; + return []; + } + unreported = bridged; } if (restarted) state.sequenceStartSignature = sequenceStart; } else { - if (signature === state.lastUsageSignature) return null; + if (signature === state.lastUsageSignature) return []; // Advance expected totals for billed legacy requests so a cumulative re-emission is stale. const advanced = advanceCodexUsage(previous, last); if (advanced === null) { state.malformedRecords += 1; - return null; + return []; } state.lastCumulativeUsage = advanced; } state.lastUsageSignature = signature; - const inputTokens = last.input_tokens; - const cachedInputTokens = last.cached_input_tokens; - const cacheCreationTokens = last.cache_write_input_tokens; - const outputTokens = last.output_tokens; + return [ + ...unreported.map((request) => codexRequestRecord(state, request.usage, request)), + codexRequestRecord(state, last, codexAttribution(state, timestampMs)), + ]; +} + +function codexAttribution(state: CodexScanState, timestampMs: number): CodexRequestAttribution { + return { + timestampMs, + model: state.model, + turnId: state.turnId, + speed: state.speed, + sawThreadSettings: state.sawThreadSettings, + }; +} + +function codexRequestRecord( + state: CodexScanState, + usage: CodexTokenUsage, + { timestampMs, model, turnId, speed, sawThreadSettings }: CodexRequestAttribution, +): UsageRecord { + const inputTokens = usage.input_tokens; + const cachedInputTokens = usage.cached_input_tokens; + const cacheCreationTokens = usage.cache_write_input_tokens; + const outputTokens = usage.output_tokens; const totals: UsageTokenTotals = { // Codex reports `input_tokens` inclusive of the cached portion. @@ -287,24 +380,22 @@ function parseCodexRecord(parsed: unknown, state: CodexScanState): UsageRecord | cacheCreationTokens, outputTokens, // Reported inside output_tokens, surfaced separately for the token mix. - reasoningTokens: last.reasoning_output_tokens, + reasoningTokens: usage.reasoning_output_tokens, }; - if (totalTokens(totals) === 0) return null; - return { provider: "codex", timestampMs, - model: state.model, + model, sessionId: state.sessionId, totals, reportedCostUsd: null, - speed: state.speed, - ...(!state.sawThreadSettings && + speed, + ...(!sawThreadSettings && state.modelProvider === "openai" && state.sessionId.length > 0 && - state.turnId.length > 0 - ? { unresolvedServiceTierTurnId: state.turnId } + turnId.length > 0 + ? { unresolvedServiceTierTurnId: turnId } : {}), dedupeKey: null, }; @@ -317,9 +408,6 @@ function codexSpeed(serviceTier: unknown): UsageSpeed { return "standard"; } -const orEmpty = (record: UsageRecord | null): readonly UsageRecord[] => - record === null ? [] : [record]; - export const codexUsageFormat: TranscriptUsageFormat = { selectFields: { type: true, @@ -335,12 +423,19 @@ export const codexUsageFormat: TranscriptUsageFormat = { forked_from_id: true, source: { subagent: { thread_spawn: { parent_thread_id: true } } }, info: { last_token_usage: true, total_token_usage: true }, + usage: true, }, }, // Metadata lines carry model, tier and session even without usage. - usageMarkers: ['"token_count"', '"turn_context"', '"thread_settings_applied"', '"session_meta"'], - parseLine: (line, state) => orEmpty(parseCodexLine(line, state)), - parseProjected: (projected, state) => orEmpty(parseCodexRecord(projected, state)), + usageMarkers: [ + '"token_count"', + '"token_usage_record"', + '"turn_context"', + '"thread_settings_applied"', + '"session_meta"', + ], + parseLine: parseCodexLine, + parseProjected: parseCodexRecord, state: { initial: initialCodexScanState, schema: CodexScanState }, readMalformedRecords: (state) => state.malformedRecords, // A rollout that moved after it was read leaves a copy in both places. diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 227fac19036b1..9cf97cd4c7c4d 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2624,7 +2624,7 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); - it.live.each([4, 5, 6, 7, 8])( + it.live.each([4, 5, 6, 7, 8, 10])( "upgrades a v%s cache: reparses live Codex usage, keeps deleted rollouts and the old file", (version) => Effect.gen(function* () { @@ -2691,7 +2691,7 @@ describe("UsageService", () => { delete file.cs.lastCumulativeUsage; delete file.cs.malformedRecords; } - if (version === 8 && path === livePath) { + if (version >= 8 && path === livePath) { file.r = []; file.mr = 1; file.cs.malformedRecords = 1; @@ -2723,7 +2723,7 @@ describe("UsageService", () => { 60 + 20 * (version === 4 ? 1 : 2), ); yield* upgraded.awaitPersisted; - assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v10.json"); + assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v11.json"); assert.strictEqual( yield* Effect.promise(() => NodeFSP.readFile(legacyPath, "utf8")), legacy, @@ -2750,6 +2750,106 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("re-parses a clean v10 Codex rollout that grew after an interrupted request", () => + Effect.gen(function* () { + const { home, settings } = yield* setup; + const sessions = NodePath.join(home, "codex", "sessions"); + const live = NodePath.join(sessions, "live.jsonl"); + const usage = (input: number, output: number) => ({ + input_tokens: input, + output_tokens: output, + total_tokens: input + output, + }); + const lines = (values: readonly unknown[]) => + values.map((value) => `${encodeUnknownJsonString(value)}\n`).join(""); + const request = (second: number, last: unknown) => ({ + type: "token_usage_record", + timestamp: `2026-08-01T10:00:0${second}Z`, + payload: { usage: last }, + }); + const tokenCount = (second: number, last: unknown, total: unknown) => ({ + type: "event_msg", + timestamp: `2026-08-01T10:00:0${second}Z`, + payload: { + type: "token_count", + info: { last_token_usage: last, total_token_usage: total }, + }, + }); + yield* Effect.promise(async () => { + await NodeFSP.mkdir(sessions, { recursive: true }); + await NodeFSP.writeFile( + live, + lines([ + { type: "session_meta", payload: { id: "live" } }, + { type: "turn_context", payload: { turn_id: "turn-1", model: "gpt-6-astra" } }, + request(1, usage(100, 20)), + tokenCount(1, usage(100, 20), usage(100, 20)), + // Interrupted after this request finished: no token_count reports it. + request(2, usage(150, 10)), + ]), + ); + }); + + yield* Effect.gen(function* () { + const { stateDir } = yield* ServerConfig.ServerConfig; + const tokens = (summary: UsageSummary) => + summary.buckets.reduce((sum, bucket) => sum + totalTokens(bucket.totals), 0); + const first = yield* UsageService.make; + assert.strictEqual(tokens(yield* first.readSummary(WINDOW)), 120); + yield* first.awaitPersisted; + + yield* Effect.promise(async () => { + const cachePath = NodePath.join(stateDir, SCAN_CACHE_FILE_NAME); + const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as { + files: Record }>; + }; + // The v10 parser ignored `token_usage_record`, so its clean state held nothing pending. + for (const file of Object.values(document.files)) { + file.cs = Object.fromEntries( + Object.entries(file.cs).filter(([key]) => !key.startsWith("unreported")), + ); + } + await NodeFSP.rm(cachePath); + await NodeFSP.writeFile( + NodePath.join(stateDir, "usage-scan-cache-v10.json"), + encodeUnknownJsonString({ ...document, version: 10 }), + ); + await NodeFSP.appendFile( + live, + lines([ + { type: "turn_context", payload: { turn_id: "turn-2", model: "gpt-6-astra" } }, + request(5, usage(200, 30)), + tokenCount(5, usage(200, 30), usage(450, 60)), + ]), + ); + }); + + const upgraded = yield* UsageService.make; + const summary = yield* upgraded.readSummary(WINDOW); + assert.deepStrictEqual( + { + tokens: tokens(summary), + malformedRecords: summary.sources.find( + (source) => source.fingerprint.provider === "codex", + )?.malformedRecords, + }, + { tokens: 510, malformedRecords: 0 }, + ); + }).pipe( + Effect.provide( + layerService({ + prefix: "usage-service-cache-v10-interrupted-test", + home, + settings, + ratesDocument: { + "gpt-6-astra": { input_cost_per_token: 0, output_cost_per_token: 1 }, + }, + }), + ), + ); + }).pipe(Effect.scoped), + ); + it.live("preserves saved tokens, costs and sessions after transcript cleanup and restart", () => Effect.gen(function* () { const { transcript, settings, home } = yield* setup; diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index 477b8bb6fc4f6..fa659c4a443b3 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -161,7 +161,7 @@ describe("scan cache round trip", () => { expect(restored).toEqual(original); }); - it.each([5, 6, 7, 8])( + it.each([5, 6, 7, 8, 10])( "retains v%s history and compatible positions but invalidates obsolete state", (version) => { const original = cacheWith([["/gone.jsonl", 100, [record({ speed: "fast" })]]]); @@ -218,25 +218,25 @@ 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-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( - version >= 8 - ? cleanCodex + expect(restored.get("/gone.jsonl")).toEqual( + version >= 10 + ? original.get("/gone.jsonl") : { - ...cleanCodex, + ...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, + position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, + }); expect(restored.get("/gone-pi.jsonl")).toEqual( version >= 6 ? original.get("/gone-pi.jsonl") diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 70c0c3df17e3e..c9032d64c05bd 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -9,7 +9,7 @@ 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 = 10 as const; +const USAGE_SCAN_CACHE_VERSION = 11 as const; const SPEED_COMPATIBLE_SINCE_VERSION = 4; /** @@ -17,8 +17,9 @@ const SPEED_COMPATIBLE_SINCE_VERSION = 4; * replace it, dropping saved usage for deleted transcripts. Separate files keep * both. */ -export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v10.json"; +export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v11.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [ + "usage-scan-cache-v10.json", "usage-scan-cache-v9.json", "usage-scan-cache-v8.json", "usage-scan-cache-v7.json", @@ -291,16 +292,19 @@ export function decodeScanCache( continue; } // Keep history for deleted rollouts, but re-parse incompatible reducer state. - // v9 repairs skipped initial Codex requests; clean v8 entries can still resume. + // v9 and v11 repair skipped Codex requests; clean v8+ entries stay served while unchanged. // v10 adds Claude advisor calls, which no older Claude entry holds. const legacy = (entry.p === "claude" && version < 10) || (format.state !== undefined && (version < 6 || - (entry.p === "codex" && (version < 8 || (version < 9 && malformedRecords > 0))))); + (entry.p === "codex" && (version < 8 || (version < 11 && 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; + // Before v11 a clean scan could end past an interrupted request whose usage its state dropped, + // so a grown file re-parses whole instead of resuming. + const resumable = !legacy && !(entry.p === "codex" && version < 11); const provider = entry.p; const records = decodeRecords(entry.r, provider); @@ -314,14 +318,14 @@ export function decodeScanCache( records, tailRecords, malformedRecords, - position: legacy - ? { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null } - : { + position: resumable + ? { resumeOffset: entry.o, guardLength: entry.gl, guardHash: entry.gh, state: entry.cs, - }, + } + : { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, }); } diff --git a/apps/server/src/usage/usageTranscriptReader.test.ts b/apps/server/src/usage/usageTranscriptReader.test.ts index f2b4fb43b48b4..105884d81ccfb 100644 --- a/apps/server/src/usage/usageTranscriptReader.test.ts +++ b/apps/server/src/usage/usageTranscriptReader.test.ts @@ -7,7 +7,7 @@ import * as NodePath from "node:path"; import * as NodeZlib from "node:zlib"; import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; -import type { TranscriptUsageFormat } from "@t3tools/provider-core/server/usage"; +import { totalTokens, type TranscriptUsageFormat } from "@t3tools/provider-core/server/usage"; import * as Schema from "effect/Schema"; import { TEST_FORMATS } from "./usageTestFormats.ts"; @@ -381,6 +381,82 @@ describe("readTranscriptRecords resume", () => { }, ); + it.each([ + { mode: "buffered", streamingThresholdBytes: Infinity }, + { mode: "streaming", streamingThresholdBytes: 0 }, + ])( + "bills a Codex request whose turn was interrupted before its token_count ($mode)", + async (options) => { + const path = NodePath.join(dir, "rollout.jsonl"); + const usage = (input: number, output: number) => ({ + input_tokens: input, + cached_input_tokens: 0, + cache_write_input_tokens: 0, + output_tokens: output, + reasoning_output_tokens: 0, + total_tokens: input + output, + }); + const context = (turn: number) => ({ + type: "turn_context", + timestamp: `2026-11-02T10:00:0${turn * 4 - 4}.000Z`, + payload: { turn_id: `turn-${turn}`, model: "gpt-6-sol" }, + }); + const request = (turn: number, id: number, timestamp: string, last: unknown) => ({ + type: "token_usage_record", + timestamp, + payload: { turn_id: `turn-${turn}`, response_id: `resp_${id}`, usage: last }, + }); + const tokenCount = (timestamp: string, last: unknown, total: unknown) => ({ + type: "event_msg", + timestamp, + payload: { + type: "token_count", + info: { last_token_usage: last, total_token_usage: total }, + }, + }); + const lines = [ + { + type: "session_meta", + timestamp: "2026-11-02T10:00:00.000Z", + payload: { id: "aborted-turn-repro", model_provider: "openai" }, + }, + context(1), + request(1, 1, "2026-11-02T10:00:01.000Z", usage(100, 20)), + tokenCount("2026-11-02T10:00:01.001Z", usage(100, 20), usage(100, 20)), + // Interrupted after this request finished: no token_count reports it. + request(1, 2, "2026-11-02T10:00:02.000Z", usage(150, 10)), + { + type: "event_msg", + timestamp: "2026-11-02T10:00:03.000Z", + payload: { type: "turn_aborted", turn_id: "turn-1", reason: "interrupted" }, + }, + context(2), + request(2, 3, "2026-11-02T10:00:05.000Z", usage(200, 30)), + tokenCount("2026-11-02T10:00:05.001Z", usage(200, 30), usage(450, 60)), + ]; + // Lines longer than a read chunk take the projected path when streaming. + const padding = "x".repeat(256 * 1024); + await NodeFSP.writeFile( + path, + lines.map((line) => `${JSON.stringify({ padding, ...line })}\n`).join(""), + ); + + const result = await readTranscriptRecords(path, TEST_FORMATS.codex, undefined, options); + assert.isNotNull(result); + const requests = [...result.records, ...result.tailRecords] + .map((record) => totalTokens(record.totals)) + .toSorted((a, b) => a - b); + assert.deepStrictEqual( + { + billedTokens: requests.reduce((sum, tokens) => sum + tokens, 0), + requests, + malformedRecords: result.malformedRecords, + }, + { billedTokens: 510, requests: [120, 160, 230], malformedRecords: 0 }, + ); + }, + ); + 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));