1import type { UsageProviderKind } from "@t3tools/contracts";
2import type {
3 TranscriptUsageFormat,
4 UsageRecord,
5 UsageSpeed,
6} from "@t3tools/provider-core/server/usage";
7import * as Schema from "effect/Schema";
8
9import { CLAUDE_SYNTHETIC_MODEL } from "../provider/Drivers/claudeUsage.ts";
10import { mergeUsageCopies } from "./usageAggregation.ts";
11import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";
12
13const USAGE_SCAN_CACHE_VERSION = 15 as const;
14const SPEED_COMPATIBLE_SINCE_VERSION = 4;
15
16/**
17 * An older server sharing that directory cannot read a newer cache and would
18 * replace it, dropping saved usage for deleted transcripts. Separate files keep
19 * the newer one; older files are removed once it is saved.
20 */
21export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v15.json";
22export const LEGACY_SCAN_CACHE_FILE_NAMES = [
23 "usage-scan-cache-v14.json",
24 "usage-scan-cache-v13.json",
25 "usage-scan-cache-v12.json",
26 "usage-scan-cache-v11.json",
27 "usage-scan-cache-v10.json",
28 "usage-scan-cache-v9.json",
29 "usage-scan-cache-v8.json",
30 "usage-scan-cache-v7.json",
31 "usage-scan-cache-v6.json",
32 "usage-scan-cache-v5.json",
33 "usage-scan-cache.json",
34] as const;
35
36const SPEEDS: readonly UsageSpeed[] = ["standard", "fast", "ultrafast"];
37
38export interface CachedFile {
39 readonly size: number;
40 readonly mtimeMs: number;
41 readonly provider: UsageProviderKind;
42 readonly records: readonly UsageRecord[];
43 /** Unterminated tail records stay separate because incremental parsing rereads them. */
44 readonly tailRecords: readonly UsageRecord[];
45 /** Whole-file count, including diagnostics from the tentative tail. */
46 readonly malformedRecords: number;
47 readonly position: TranscriptParsePosition;
48}
49
50export type ScanCache = Map<string, CachedFile>;
51
52type SerializedRecord = readonly [
53 timestampMs: number,
54 modelIndex: number,
55 sessionIndex: number,
56 uncachedInputTokens: number,
57 cachedInputTokens: number,
58 cacheCreationTokens: number,
59 outputTokens: number,
60 reasoningTokens: number,
61 dedupeKey: string | null,
62 reportedCostUsd: number | null,
63 speed: number,
64 rateModelIndex?: number | null,
65 unresolvedServiceTierTurnId?: string | null,
66 cacheCreation1hTokens?: number,
67];
68
69interface SerializedFile {
70 readonly s: number;
71 readonly m: number;
72 readonly p: UsageProviderKind;
73 readonly r: readonly SerializedRecord[];
74 readonly t: readonly SerializedRecord[];
75 readonly mr: number;
76 readonly o: number;
77 readonly gl: number;
78 readonly gh: number;
79 readonly cs: unknown;
80}
81
82interface SerializedCache {
83 readonly version: number;
84 readonly models: readonly string[];
85 readonly sessions: readonly string[];
86 readonly files: Readonly<Record<string, SerializedFile>>;
87}
88
89interface InternTables {
90 readonly models: string[];
91 readonly sessions: string[];
92 readonly modelIndex: Map<string, number>;
93 readonly sessionIndex: Map<string, number>;
94}
95
96function makeInternTables(): InternTables {
97 return { models: [], sessions: [], modelIndex: new Map(), sessionIndex: new Map() };
98}
99
100function intern(table: string[], index: Map<string, number>, value: string): number {
101 const existing = index.get(value);
102 if (existing !== undefined) return existing;
103 const next = table.length;
104 table.push(value);
105 index.set(value, next);
106 return next;
107}
108
109function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile {
110 const serializeRecord = (record: UsageRecord): SerializedRecord => [
111 record.timestampMs,
112 intern(tables.models, tables.modelIndex, record.model),
113 intern(tables.sessions, tables.sessionIndex, record.sessionId),
114 record.totals.uncachedInputTokens,
115 record.totals.cachedInputTokens,
116 record.totals.cacheCreationTokens,
117 record.totals.outputTokens,
118 record.totals.reasoningTokens,
119 record.dedupeKey,
120 record.reportedCostUsd,
121 SPEEDS.indexOf(record.speed),
122 record.rateModel === undefined
123 ? null
124 : intern(tables.models, tables.modelIndex, record.rateModel),
125 ...(record.cacheCreation1hTokens !== undefined
126 ? ([record.unresolvedServiceTierTurnId ?? null, record.cacheCreation1hTokens] as const)
127 : record.unresolvedServiceTierTurnId !== undefined
128 ? ([record.unresolvedServiceTierTurnId] as const)
129 : ([] as const)),
130 ];
131 return {
132 s: entry.size,
133 m: entry.mtimeMs,
134 p: entry.provider,
135 r: entry.records.map(serializeRecord),
136 t: entry.tailRecords.map(serializeRecord),
137 mr: entry.malformedRecords,
138 o: entry.position.resumeOffset,
139 gl: entry.position.guardLength,
140 gh: entry.position.guardHash,
141 cs: entry.position.state,
142 };
143}
144
145/** Yields the document in pieces so large caches never become one string; growing intern tables preserve indexes, retaining unused strings until restart. */
146export function makeScanCacheWriter(): (
147 cache: ScanCache,
148 extra: Readonly<Record<string, unknown>>,
149) => Iterable<string> {
150 const tables = makeInternTables();
151 function* encode(entries: readonly (readonly [string, CachedFile])[], extra: object) {
152 yield '{"files":{';
153 for (const [index, [path, entry]] of entries.entries()) {
154 yield `${index === 0 ? "" : ","}${JSON.stringify(path)}:${JSON.stringify(serializeFile(entry, tables))}`;
155 }
156 // Encoded after the files, which may have added to the intern tables.
157 const tail = JSON.stringify({
158 ...extra,
159 version: USAGE_SCAN_CACHE_VERSION,
160 models: tables.models,
161 sessions: tables.sessions,
162 });
163 yield `},${tail.slice(1)}`;
164 }
165 // Snapshot now: the encoding runs later, while the file is written.
166 return (cache, extra) => encode([...cache], extra);
167}
168
169function isRecordArray(value: unknown): value is readonly unknown[] {
170 return Array.isArray(value);
171}
172
173export function decodeScanCache(
174 document: unknown,
175 formats: ReadonlyMap<UsageProviderKind, TranscriptUsageFormat<unknown>>,
176): ScanCache {
177 const cache: ScanCache = new Map();
178 const isValidState = makeStateValidators(formats);
179 if (typeof document !== "object" || document === null) return cache;
180
181 const root = document as Partial<SerializedCache>;
182 const version = root.version;
183 if (
184 typeof version !== "number" ||
185 version < SPEED_COMPATIBLE_SINCE_VERSION ||
186 version > USAGE_SCAN_CACHE_VERSION
187 ) {
188 return cache;
189 }
190 if (!isRecordArray(root.models) || !isRecordArray(root.sessions)) return cache;
191 if (typeof root.files !== "object" || root.files === null) return cache;
192
193 if (!root.models.every((value) => typeof value === "string")) return cache;
194 if (!root.sessions.every((value) => typeof value === "string")) return cache;
195 const models = root.models as readonly string[];
196 const sessions = root.sessions as readonly string[];
197
198 // Reject entire corrupt entries; partial survivors would appear warm and hide usage until the file changes.
199 const decodeRecords = (
200 rows: readonly unknown[],
201 provider: UsageProviderKind,
202 ): UsageRecord[] | null => {
203 const records: UsageRecord[] = [];
204 for (const row of rows) {
205 if (!isRecordArray(row) || row.length < 11) return null;
206 const [
207 timestampMs,
208 modelIndex,
209 sessionIndex,
210 uncached,
211 cached,
212 cacheCreation,
213 output,
214 reasoning,
215 dedupeKey,
216 reportedCostUsd,
217 speedIndex,
218 rateModelIndex,
219 unresolvedServiceTierTurnId,
220 cacheCreation1h = 0,
221 ] = row as SerializedRecord;
222 const speed = typeof speedIndex === "number" ? SPEEDS[speedIndex] : undefined;
223
224 const model = typeof modelIndex === "number" ? models[modelIndex] : undefined;
225 const rateModel = typeof rateModelIndex === "number" ? models[rateModelIndex] : undefined;
226 if (
227 typeof timestampMs !== "number" ||
228 !Number.isFinite(timestampMs) ||
229 model === undefined ||
230 (unresolvedServiceTierTurnId != null && typeof unresolvedServiceTierTurnId !== "string") ||
231 !Number.isFinite(cacheCreation1h) ||
232 (rateModelIndex != null && rateModel === undefined) ||
233 !Number.isFinite(uncached) ||
234 !Number.isFinite(cached) ||
235 !Number.isFinite(cacheCreation) ||
236 !Number.isFinite(output) ||
237 !Number.isFinite(reasoning) ||
238 speed === undefined
239 ) {
240 return null;
241 }
242
243 // Unchanged and deleted transcripts may only exist in the cache, so the
244 // parser cannot remove their previously saved local placeholders.
245 if (provider === "claude" && model === CLAUDE_SYNTHETIC_MODEL) continue;
246
247 records.push({
248 provider,
249 timestampMs,
250 model,
251 ...(rateModel === undefined ? {} : { rateModel }),
252 sessionId: (typeof sessionIndex === "number" ? sessions[sessionIndex] : undefined) ?? "",
253 totals: {
254 uncachedInputTokens: uncached,
255 cachedInputTokens: cached,
256 cacheCreationTokens: cacheCreation,
257 outputTokens: output,
258 reasoningTokens: reasoning,
259 },
260 ...(cacheCreation1h > 0 ? { cacheCreation1hTokens: cacheCreation1h } : {}),
261 reportedCostUsd: typeof reportedCostUsd === "number" ? reportedCostUsd : null,
262 speed,
263 ...(version >= 7 && unresolvedServiceTierTurnId != null
264 ? { unresolvedServiceTierTurnId }
265 : {}),
266 dedupeKey: typeof dedupeKey === "string" ? dedupeKey : null,
267 });
268 }
269 return records;
270 };
271
272 for (const [path, raw] of Object.entries(root.files)) {
273 if (typeof raw !== "object" || raw === null) continue;
274 const entry = raw as Partial<SerializedFile>;
275 if (typeof entry.s !== "number" || typeof entry.m !== "number") continue;
276 if (entry.p === undefined) continue;
277 const format = formats.get(entry.p);
278 if (format === undefined) continue;
279 if (!isRecordArray(entry.r) || !isRecordArray(entry.t)) continue;
280 const malformedRecords = entry.mr === undefined ? 0 : entry.mr;
281 if (!Number.isSafeInteger(malformedRecords) || malformedRecords < 0) continue;
282 // Position fields feed byte offsets and a Buffer allocation in the reader,
283 // so anything outside their real ranges must reject the entry: a bogus
284 // guard length would otherwise fail every parse of the file, silently
285 // dropping its usage instead of costing the documented cold re-parse.
286 if (
287 typeof entry.o !== "number" ||
288 !Number.isSafeInteger(entry.o) ||
289 entry.o < 0 ||
290 typeof entry.gl !== "number" ||
291 !Number.isSafeInteger(entry.gl) ||
292 entry.gl < 0 ||
293 entry.gl > GUARD_LENGTH ||
294 entry.gl > entry.o ||
295 typeof entry.gh !== "number" ||
296 !Number.isFinite(entry.gh)
297 ) {
298 continue;
299 }
300 // Keep history for deleted rollouts, but re-parse incompatible reducer state.
301 // v13 adds Claude 1-hour cache writes; v14 uses Codex request ledgers.
302 // v15 repairs skipped Codex requests whose counters rewound; clean v14 entries stay served.
303 const legacy =
304 (entry.p === "claude" && version < 13) ||
305 (format.state !== undefined &&
306 (version < 6 ||
307 (entry.p === "codex" && (version < 14 || (version < 15 && malformedRecords > 0)))));
308 // A corrupt state disqualifies the entry: resuming with it would attach
309 // appended usage to the wrong model or replay fork-copied history.
310 if (!legacy && !isValidState.get(entry.p)?.(entry.cs)) continue;
311 // Before v15 the state lacks the latest snapshot time that keeps a re-emitted older snapshot
312 // from billing twice, so a grown file re-parses whole instead of resuming.
313 const resumable = !legacy && !(entry.p === "codex" && version < 15);
314
315 const provider = entry.p;
316 const records = decodeRecords(entry.r, provider);
317 const tailRecords = decodeRecords(entry.t, provider);
318 if (records === null || tailRecords === null) continue;
319
320 cache.set(path, {
321 size: legacy ? -1 : entry.s,
322 mtimeMs: entry.m,
323 provider,
324 records,
325 tailRecords,
326 malformedRecords,
327 position: resumable
328 ? {
329 resumeOffset: entry.o,
330 guardLength: entry.gl,
331 guardHash: entry.gh,
332 state: entry.cs,
333 }
334 : { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null },
335 });
336 }
337
338 return cache;
339}
340
341function makeStateValidators(
342 formats: ReadonlyMap<UsageProviderKind, TranscriptUsageFormat<unknown>>,
343): ReadonlyMap<UsageProviderKind, (value: unknown) => boolean> {
344 return new Map(
345 [...formats].map(([provider, format]) => {
346 const isState = format.state === undefined ? undefined : Schema.is(format.state.schema);
347 return [
348 provider,
349 (value: unknown) => value === null || (isState !== undefined && isState(value)),
350 ] as const;
351 }),
352 );
353}
354
355export function pruneScanCache(cache: ScanCache, retentionCutoffMs: number): number {
356 let removed = 0;
357 for (const [path, entry] of cache) {
358 if (entry.mtimeMs < retentionCutoffMs) {
359 cache.delete(path);
360 removed += 1;
361 }
362 }
363 return removed;
364}
365
366/** Pass only newline-terminated records: an unterminated tail can still change, so the aggregator merges it per scan. */
367export function dedupeWithinFile(records: readonly UsageRecord[]): readonly UsageRecord[] {
368 const kept: UsageRecord[] = [];
369 const firstIndex = new Map<string, number>();
370 for (const record of records) {
371 if (record.dedupeKey !== null) {
372 const index = firstIndex.get(record.dedupeKey);
373 if (index !== undefined) {
374 kept[index] = mergeUsageCopies(kept[index]!, record);
375 continue;
376 }
377 firstIndex.set(record.dedupeKey, kept.length);
378 }
379 kept.push(record);
380 }
381 return kept;
382}