authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-09-01 11:59:16+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 02:59:16-07:00
log8f1ef8b9eb72ac6cd8510364608e01dc7febf9c1
tree5e5df7c15de7122cc8e2ed533646ebd645357331
parent0bfb6df34b26dfe0162db6c09dca00bc8c5a5ec4
signature Signed by PGP key B5690EEEBB952194

perf(server): scan only appended transcript bytes for usage summaries (#9024)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Theo Browne <me@t3.gg>

6 files changed, 963 insertions(+), 90 deletions(-)

apps/server/src/usage/UsageService.test.ts created+226
......@@ -0,0 +1,226 @@
1// @effect-diagnostics nodeBuiltinImport:off - the suite seeds and grows real
2// transcript trees on disk, outside the service's Effect FileSystem.
3import * as NodeFSP from "node:fs/promises";
4import * as NodeOS from "node:os";
5import * as NodePath from "node:path";
6
7import { assert, describe, it } from "@effect/vitest";
8import * as NodeServices from "@effect/platform-node/NodeServices";
9import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";
10import { UsageDay, type UsageSummaryInput } from "@t3tools/contracts";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as Fiber from "effect/Fiber";
14import * as Layer from "effect/Layer";
15import * as Scheduler from "effect/Scheduler";
16import { HttpClient, HttpClientResponse } from "effect/unstable/http";
17
18import * as ServerConfig from "../config.ts";
19import * as ServerSettings from "../serverSettings.ts";
20import * as UsageService from "./UsageService.ts";
21
22function claudeLine(id: number, outputTokens: number): string {
23 return `${JSON.stringify({
24 type: "assistant",
25 timestamp: "2026-08-01T10:00:00Z",
26 requestId: `req_${id}`,
27 sessionId: "session-1",
28 message: {
29 id: `msg_${id}`,
30 model: "claude-fable-5",
31 usage: { input_tokens: 10, output_tokens: outputTokens },
32 },
33 })}\n`;
34}
35
36const WINDOW: UsageSummaryInput = {
37 timeZone: "UTC",
38 sinceDay: UsageDay.make("2026-07-31"),
39 untilDay: UsageDay.make("2026-08-02"),
40};
41
42const setup = Effect.gen(function* () {
43 const home = yield* Effect.promise(() =>
44 NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-service-test-")),
45 );
46 yield* Effect.addFinalizer(() =>
47 Effect.promise(() => NodeFSP.rm(home, { recursive: true, force: true })),
48 );
49 const transcriptDir = NodePath.join(home, "claude", "projects", "proj");
50 yield* Effect.promise(() => NodeFSP.mkdir(transcriptDir, { recursive: true }));
51 return {
52 home,
53 transcript: NodePath.join(transcriptDir, "session.jsonl"),
54 settings: {
55 providers: {
56 claudeAgent: { homePath: NodePath.join(home, "claude") },
57 codex: { homePath: NodePath.join(home, "codex") },
58 },
59 },
60 };
61});
62
63const serviceLayers = (input: {
64 readonly prefix: string;
65 readonly home: string;
66 readonly settings: Parameters<typeof ServerSettings.layerTest>[0];
67 readonly onRatesFetch?: () => void;
68}) =>
69 ServerConfig.layerTest(process.cwd(), { prefix: input.prefix }).pipe(
70 Layer.provideMerge(NodeServices.layer),
71 Layer.provideMerge(ServerSettings.layerTest(input.settings)),
72 Layer.provideMerge(
73 Layer.succeed(
74 HttpClient.HttpClient,
75 HttpClient.make((request) =>
76 Effect.sync(() => {
77 input.onRatesFetch?.();
78 // Unparsable rates: every scan retries the fetch, which makes the
79 // fetch count a boundary-level observation of how many scans ran.
80 return HttpClientResponse.fromWeb(request, Response.json({}));
81 }),
82 ),
83 ),
84 ),
85 Layer.provideMerge(
86 Layer.succeed(HostProcessEnvironment, { GROK_HOME: NodePath.join(input.home, "grok") }),
87 ),
88 );
89
90function totalOutputTokens(summary: { buckets: readonly { totals: { outputTokens: number } }[] }) {
91 return summary.buckets.reduce((sum, bucket) => sum + bucket.totals.outputTokens, 0);
92}
93
94describe("UsageService", () => {
95 it.live("counts appended usage on a rescan of a grown transcript", () =>
96 Effect.gen(function* () {
97 const { transcript, settings, home } = yield* setup;
98 yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5)));
99
100 const service = yield* UsageService.make.pipe(
101 Effect.provide(serviceLayers({ prefix: "usage-service-grow-test", home, settings })),
102 );
103
104 const first = yield* service.readSummary(WINDOW);
105 assert.strictEqual(totalOutputTokens(first), 5);
106
107 yield* Effect.promise(() => NodeFSP.appendFile(transcript, claudeLine(2, 7)));
108 const second = yield* service.readSummary(WINDOW);
109 assert.strictEqual(totalOutputTokens(second), 12);
110 }).pipe(Effect.scoped),
111 );
112
113 it.live("shares one scan between concurrent identical requests", () =>
114 Effect.gen(function* () {
115 const { transcript, settings, home } = yield* setup;
116 yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5)));
117
118 let ratesFetches = 0;
119 const service = yield* UsageService.make.pipe(
120 Effect.provide(
121 serviceLayers({
122 prefix: "usage-service-flight-test",
123 home,
124 settings,
125 onRatesFetch: () => {
126 ratesFetches += 1;
127 },
128 }),
129 ),
130 );
131
132 const [first, second] = yield* Effect.all(
133 [service.readSummary(WINDOW), service.readSummary(WINDOW)],
134 { concurrency: 2 },
135 );
136 assert.deepStrictEqual(first, second);
137 assert.strictEqual(ratesFetches, 1);
138
139 // A later request is fresh work again, not a stale cached answer.
140 yield* service.readSummary(WINDOW);
141 assert.strictEqual(ratesFetches, 2);
142 }).pipe(Effect.scoped),
143 );
144
145 it.live("does not orphan an in-flight scan when its first caller is interrupted", () =>
146 Effect.gen(function* () {
147 const { settings, home } = yield* setup;
148 const service = yield* UsageService.make.pipe(
149 Effect.provide(
150 serviceLayers({ prefix: "usage-service-interruption-test", home, settings }),
151 ),
152 );
153
154 let orphanedAt: number | undefined;
155 for (let interruptAt = 1; interruptAt <= 31; interruptAt += 1) {
156 const tasks: Array<() => void> = [];
157 const dispatcher: Scheduler.SchedulerDispatcher = {
158 scheduleTask: (task) => tasks.push(task),
159 flush: () => {
160 let task: (() => void) | undefined;
161 while ((task = tasks.shift()) !== undefined) task();
162 },
163 };
164
165 let requestFiber: Fiber.Fiber<unknown, unknown> | undefined;
166 let requestChecks = 0;
167 const scheduler: Scheduler.Scheduler = {
168 executionMode: "async",
169 makeDispatcher: () => dispatcher,
170 shouldYield: (fiber) => {
171 if (fiber !== requestFiber) return false;
172 requestChecks += 1;
173 if (requestChecks !== interruptAt) return false;
174 fiber.interruptUnsafe();
175 return true;
176 },
177 };
178
179 // Each candidate needs a distinct key because the broken case leaves
180 // its entry in the service's private in-flight map. The invalid window
181 // keeps the real scan synchronous once its detached fiber starts.
182 const input: UsageSummaryInput = {
183 ...WINDOW,
184 sinceDay: UsageDay.make("2026-09-01"),
185 untilDay: UsageDay.make(`2026-08-${String(interruptAt).padStart(2, "0")}`),
186 };
187 const first = yield* service
188 .readSummary(input)
189 .pipe(
190 Effect.exit,
191 Effect.provideService(Scheduler.Scheduler, scheduler),
192 Effect.forkChild,
193 );
194 requestFiber = first;
195 yield* Effect.yieldNow;
196 dispatcher.flush();
197
198 const second = yield* service.readSummary(input).pipe(
199 Effect.match({
200 onFailure: (error) => error.reason,
201 onSuccess: () => "success" as const,
202 }),
203 Effect.provideService(Scheduler.Scheduler, scheduler),
204 Effect.forkChild,
205 );
206 yield* Effect.yieldNow;
207 dispatcher.flush();
208 const secondExit = second.pollUnsafe();
209 if (secondExit === undefined) {
210 second.interruptUnsafe();
211 orphanedAt = interruptAt;
212 break;
213 }
214 if (Exit.isFailure(secondExit)) {
215 assert.fail("the matching request fiber was interrupted");
216 }
217 assert.strictEqual(secondExit.value, "invalidWindow");
218 }
219
220 assert.isUndefined(
221 orphanedAt,
222 `interruption left the next matching request pending at scheduler check ${orphanedAt}`,
223 );
224 }).pipe(Effect.scoped),
225 );
226});
apps/server/src/usage/UsageService.ts+138-27
......@@ -7,7 +7,8 @@
77 *
88 * Transcripts are append-only, so parsed records are memoised per file by
99 * `(size, mtime)`. A cold 30-day scan of ~1.4 GB lands around 2-3 seconds; warm
10 * scans only reparse files that changed.
10 * scans only reparse files that changed, and a file that merely grew resumes
11 * from its cached parse position so only the appended bytes are read.
1112 *
1213 * @module UsageService
1314 */
......@@ -26,6 +27,7 @@ import * as Cause from "effect/Cause";
2627import * as Clock from "effect/Clock";
2728import * as Context from "effect/Context";
2829import * as DateTime from "effect/DateTime";
30import * as Deferred from "effect/Deferred";
2931import * as Effect from "effect/Effect";
3032import * as FileSystem from "effect/FileSystem";
3133import * as Layer from "effect/Layer";
......@@ -272,7 +274,14 @@ export const make = Effect.gen(function* () {
272274 );
273275 });
274276
275 /** Parses one transcript, reusing the cached result when it is unchanged. */
277 /**
278 * Parses one transcript, reusing the cached result when it is unchanged.
279 *
280 * A file that only grew re-parses from the cached position, so an actively
281 * written multi-hundred-megabyte rollout costs its appended bytes per scan
282 * rather than a full re-read. The reader verifies the position's guard bytes
283 * and silently restarts from byte 0 when they no longer match.
284 */
276285 const readFileRecords = (
277286 filePath: string,
278287 size: number,
......@@ -289,23 +298,85 @@ export const make = Effect.gen(function* () {
289298 cached.mtimeMs === mtimeMs &&
290299 cached.provider === provider
291300 ) {
292 return cached.records;
301 return cached.tailRecords.length === 0
302 ? cached.records
303 : [...cached.records, ...cached.tailRecords];
293304 }
294305
295 const parsed = yield* Effect.promise(() => readTranscriptRecords(filePath, provider));
306 // Only a strictly grown file may resume. Same size with a new mtime, or
307 // a shrunken file, means rewritten content; re-parse it whole.
308 const resumeFrom =
309 cached !== undefined && cached.provider === provider && size > cached.size
310 ? cached.position
311 : undefined;
312
313 const parsed = yield* Effect.promise(() =>
314 readTranscriptRecords(filePath, provider, resumeFrom),
315 );
296316 // A read failure is not an empty transcript: caching it under this
297317 // (size, mtime) would silently drop the file's usage until it changes.
298318 if (parsed === null) return [];
299 // Stored already de-duplicated within the file, which is 99% of all
300 // duplicates. The aggregator still runs the cross-file dedupe pass.
301 const records = dedupeWithinFile(parsed);
302319
303 fileCache.set(filePath, { size, mtimeMs, provider, records });
320 // Stored already de-duplicated within the file, which is 99% of all
321 // duplicates. The aggregator still runs the cross-file dedupe pass. One
322 // seen set spans the cached base, the new lines, and the tail so a
323 // resumed parse dedupes exactly like a full one.
324 const base = parsed.resumed && cached !== undefined ? cached.records : [];
325 const seen = new Set<string>();
326 const records = dedupeWithinFile([...base, ...parsed.records], seen);
327 const tailRecords = dedupeWithinFile(parsed.tailRecords, seen);
328
329 fileCache.set(filePath, {
330 size,
331 mtimeMs,
332 provider,
333 records,
334 tailRecords,
335 position: parsed.position,
336 });
304337 cacheDirty = true;
305 return records;
338 return tailRecords.length === 0 ? records : [...records, ...tailRecords];
306339 });
307340
308 const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) {
341 /** One provider directory's walk and parse, before rates are involved. */
342 interface ScannedDir {
343 readonly provider: UsageProviderKind;
344 readonly dir: string;
345 readonly volumeId: string;
346 /** Parsed records per file, or `null` when the directory does not exist. */
347 readonly files:
348 | readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]
349 | null;
350 }
351
352 const collectDirs = Effect.fn("UsageService.collectDirs")(function* (windowStartMs: number) {
353 // The home resolvers ask for `Path` themselves; satisfy them from the
354 // instance we already hold so the scan stays context-free.
355 const dirs = yield* resolveTranscriptDirs().pipe(Effect.provideService(Path.Path, path));
356 const scanned: ScannedDir[] = [];
357 for (const { provider, dir, fileName } of dirs) {
358 const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir));
359 const exists = yield* fileSystem
360 .exists(dir)
361 .pipe(Effect.catchCause(() => Effect.succeed(false)));
362 if (!exists) {
363 scanned.push({ provider, dir, volumeId, files: null });
364 continue;
365 }
366 const files = yield* Effect.promise(() =>
367 listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }),
368 );
369 const parsedFiles: { path: string; records: readonly UsageRecord[] }[] = [];
370 for (const file of files) {
371 const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider);
372 parsedFiles.push({ path: file.path, records });
373 }
374 scanned.push({ provider, dir, volumeId, files: parsedFiles });
375 }
376 return scanned;
377 });
378
379 const scanSummary = Effect.fn("UsageService.scanSummary")(function* (input: UsageSummaryInput) {
309380 if (input.sinceDay > input.untilDay) {
310381 return yield* new UsageReadError({
311382 reason: "invalidWindow",
......@@ -338,13 +409,9 @@ export const make = Effect.gen(function* () {
338409 }
339410
340411 const startedAtMs = yield* Clock.currentTimeMillis;
341 yield* ensureRates();
342412 yield* ensureScanCacheLoaded;
343413
344414 const hostId = NodeOS.hostname();
345 // The home resolvers ask for `Path` themselves; satisfy them from the
346 // instance we already hold so `readSummary` stays context-free.
347 const dirs = yield* resolveTranscriptDirs().pipe(Effect.provideService(Path.Path, path));
348415 const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`);
349416 if (Option.isNone(windowStart)) {
350417 return yield* new UsageReadError({
......@@ -355,6 +422,13 @@ export const make = Effect.gen(function* () {
355422 const windowStartMs =
356423 (hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS;
357424
425 // Pricing only matters once records are aggregated, so the rate table
426 // loads while transcripts stream instead of gating them: a cold rates
427 // fetch on a slow network no longer delays the scan by its own timeout.
428 const [, scannedDirs] = yield* Effect.all([ensureRates(), collectDirs(windowStartMs)], {
429 concurrency: 2,
430 });
431
358432 const aggregator = new UsageAggregator({
359433 timeZone: input.timeZone,
360434 sinceDay: input.sinceDay,
......@@ -368,13 +442,8 @@ export const make = Effect.gen(function* () {
368442 const livePaths = new Set<string>();
369443 const walkedRoots: string[] = [];
370444
371 for (const { provider, dir, fileName } of dirs) {
372 const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir));
373 const exists = yield* fileSystem
374 .exists(dir)
375 .pipe(Effect.catchCause(() => Effect.succeed(false)));
376
377 if (!exists) {
445 for (const { provider, dir, volumeId, files } of scannedDirs) {
446 if (files === null) {
378447 sources.push({
379448 fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId },
380449 status: "missing",
......@@ -388,9 +457,6 @@ export const make = Effect.gen(function* () {
388457 }
389458
390459 walkedRoots.push(dir);
391 const files = yield* Effect.promise(() =>
392 listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }),
393 );
394460 let scannedFiles = 0;
395461 let skippedFiles = 0;
396462 // Distinct per directory. Buckets carry per-cell session counts, but a
......@@ -399,13 +465,12 @@ export const make = Effect.gen(function* () {
399465
400466 for (const file of files) {
401467 livePaths.add(file.path);
402 const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider);
403 if (records.length === 0) {
468 if (file.records.length === 0) {
404469 skippedFiles += 1;
405470 continue;
406471 }
407472 scannedFiles += 1;
408 for (const record of records) {
473 for (const record of file.records) {
409474 // Only sessions that contributed in-window count: the mtime slack
410475 // admits boundary files whose records fall outside the range.
411476 if (aggregator.add(record) && record.sessionId.length > 0) {
......@@ -459,6 +524,52 @@ export const make = Effect.gen(function* () {
459524 } satisfies UsageSummary;
460525 });
461526
527 /**
528 * In-flight scans by window, so concurrent identical requests (the usage
529 * page open on two clients at once) share one scan instead of racing over
530 * the same corpus twice.
531 */
532 const inflightScans = new Map<string, Deferred.Deferred<UsageSummary, UsageReadError>>();
533
534 const scanKey = (input: UsageSummaryInput): string =>
535 JSON.stringify([
536 input.timeZone,
537 input.sinceDay,
538 input.untilDay,
539 input.resolution ?? "day",
540 input.sinceTime ?? null,
541 input.untilTime ?? null,
542 ]);
543
544 const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) {
545 const key = scanKey(input);
546 const deferred = yield* Effect.uninterruptible(
547 Effect.gen(function* () {
548 const existing = inflightScans.get(key);
549 if (existing !== undefined) return existing;
550
551 // Enrollment and detached-fiber creation must be atomic. Otherwise a
552 // canceled first caller can leave a Deferred with no scan to finish it.
553 const created = Deferred.makeUnsafe<UsageSummary, UsageReadError>();
554 inflightScans.set(key, created);
555 // Detached so one departing client cannot tear the scan out from under
556 // the fibers awaiting it; a finished scan warms the cache either way.
557 yield* scanSummary(input).pipe(
558 Effect.onExit((exit) =>
559 Effect.sync(() => inflightScans.delete(key)).pipe(
560 Effect.andThen(Deferred.done(created, exit)),
561 ),
562 ),
563 Effect.forkDetach,
564 );
565 return created;
566 }),
567 );
568 // Waiting stays interruptible. The detached scan continues for other
569 // callers and still warms the cache if this caller leaves.
570 return yield* Deferred.await(deferred);
571 });
572
462573 return { readSummary } as const;
463574});
464575
apps/server/src/usage/usageScanCache.test.ts+73-2
......@@ -5,6 +5,7 @@ import {
55 dedupeWithinFile,
66 encodeScanCache,
77 pruneScanCache,
8 type CachedFile,
89 type ScanCache,
910} from "./usageScanCache.ts";
1011import type { UsageRecord } from "./usageTranscripts.ts";
......@@ -28,10 +29,27 @@ function record(overrides: Partial<UsageRecord> = {}): UsageRecord {
2829 };
2930}
3031
32function position(overrides: Partial<CachedFile["position"]> = {}): CachedFile["position"] {
33 return {
34 resumeOffset: 120,
35 guardLength: 64,
36 guardHash: 0xdeadbeef,
37 codexState: null,
38 ...overrides,
39 };
40}
41
3142function cacheWith(entries: readonly [string, number, readonly UsageRecord[]][]): ScanCache {
3243 const cache: ScanCache = new Map();
3344 for (const [path, mtimeMs, records] of entries) {
34 cache.set(path, { size: records.length * 10, mtimeMs, provider: "claude", records });
45 cache.set(path, {
46 size: records.length * 10,
47 mtimeMs,
48 provider: "claude",
49 records,
50 tailRecords: [],
51 position: position(),
52 });
3553 }
3654 return cache;
3755}
......@@ -49,14 +67,67 @@ describe("scan cache round trip", () => {
4967 records: [
5068 record({ provider: "grok", model: "grok-4.5-build", dedupeKey: "s:p:grok-4.5-build" }),
5169 ],
70 tailRecords: [record({ provider: "grok", model: "grok-4.5-build", dedupeKey: null })],
71 position: position({ resumeOffset: 30, guardLength: 30, guardHash: 123 }),
72 });
73 original.set("/codex.jsonl", {
74 size: 80,
75 mtimeMs: 400,
76 provider: "codex",
77 records: [record({ provider: "codex", model: "gpt-5.2-codex", dedupeKey: null })],
78 tailRecords: [],
79 position: position({
80 codexState: {
81 model: "gpt-5.2-codex",
82 sessionId: "session-c",
83 lastUsageSignature: '{"input_tokens":1}',
84 sawSessionMeta: true,
85 suppressingForkCopies: false,
86 forkCopyAnchorMs: 0,
87 },
88 }),
5289 });
5390
5491 const restored = decodeScanCache(JSON.parse(JSON.stringify(encodeScanCache(original))));
5592
56 expect(restored.size).toBe(3);
93 expect(restored.size).toBe(4);
5794 expect(restored.get("/a.jsonl")).toEqual(original.get("/a.jsonl"));
5895 expect(restored.get("/b.jsonl")).toEqual(original.get("/b.jsonl"));
5996 expect(restored.get("/grok.jsonl")).toEqual(original.get("/grok.jsonl"));
97 expect(restored.get("/codex.jsonl")).toEqual(original.get("/codex.jsonl"));
98 });
99
100 it("drops an entry whose persisted parse state is corrupt", () => {
101 // Resuming with a bad reducer state would attach appended usage to the
102 // wrong model or replay fork-copied history; that entry must cold parse.
103 const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]]));
104 const poisoned = {
105 ...encoded,
106 files: {
107 "/a.jsonl": { ...encoded.files["/a.jsonl"]!, cs: { model: 42 } },
108 },
109 };
110
111 expect(decodeScanCache(JSON.parse(JSON.stringify(poisoned))).has("/a.jsonl")).toBe(false);
112 });
113
114 it("drops an entry whose guard length is outside the supported range", () => {
115 // The guard length sizes a Buffer in the reader; a bogus value would make
116 // every parse of that file fail and silently drop its usage.
117 const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]]));
118 const poisoned = {
119 ...encoded,
120 files: { "/a.jsonl": { ...encoded.files["/a.jsonl"]!, gl: 1e20 } },
121 };
122
123 expect(decodeScanCache(JSON.parse(JSON.stringify(poisoned))).has("/a.jsonl")).toBe(false);
124 });
125
126 it("rejects a document from the previous cache version", () => {
127 const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]]));
128 const previous = { ...encoded, version: 2 };
129
130 expect(decodeScanCache(JSON.parse(JSON.stringify(previous))).size).toBe(0);
60131 });
61132
62133 it("interns repeated model and session strings", () => {
apps/server/src/usage/usageScanCache.ts+138-38
......@@ -19,17 +19,28 @@ import * as NodePath from "node:path";
1919
2020import type { UsageProviderKind } from "@t3tools/contracts";
2121
22import type { UsageRecord } from "./usageTranscripts.ts";
22import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";
23import type { CodexScanState, UsageRecord } from "./usageTranscripts.ts";
2324
2425// v2: Codex fork-copy suppression changed what a file parses to, so v1
2526// entries would keep serving double-counted records forever.
26export const USAGE_SCAN_CACHE_VERSION = 2 as const;
27// v3: entries carry the parse position and reducer state so a grown file
28// re-parses only its appended bytes instead of starting over.
29export const USAGE_SCAN_CACHE_VERSION = 3 as const;
2730
2831export interface CachedFile {
2932 readonly size: number;
3033 readonly mtimeMs: number;
3134 readonly provider: UsageProviderKind;
35 /** Records from newline-terminated lines, up to `position.resumeOffset`. */
3236 readonly records: readonly UsageRecord[];
37 /**
38 * Records from a trailing segment the writer had not newline-terminated at
39 * parse time. Kept apart from `records` because an incremental parse
40 * re-reads that segment and would otherwise double count it.
41 */
42 readonly tailRecords: readonly UsageRecord[];
43 readonly position: TranscriptParsePosition;
3344}
3445
3546export type ScanCache = Map<string, CachedFile>;
......@@ -57,6 +68,14 @@ interface SerializedFile {
5768 readonly m: number;
5869 readonly p: UsageProviderKind;
5970 readonly r: readonly SerializedRecord[];
71 /** Tail records; see `CachedFile.tailRecords`. */
72 readonly t: readonly SerializedRecord[];
73 /** Parse position: resume offset, guard length, guard hash. */
74 readonly o: number;
75 readonly gl: number;
76 readonly gh: number;
77 /** Codex reducer state at `o`; `null` for stateless providers. */
78 readonly cs: CodexScanState | null;
6079}
6180
6281interface SerializedCache {
......@@ -82,24 +101,31 @@ export function encodeScanCache(cache: ScanCache): SerializedCache {
82101 return next;
83102 };
84103
104 const serializeRecord = (record: UsageRecord): SerializedRecord => [
105 record.timestampMs,
106 intern(models, modelIndex, record.model),
107 intern(sessions, sessionIndex, record.sessionId),
108 record.totals.uncachedInputTokens,
109 record.totals.cachedInputTokens,
110 record.totals.cacheCreationTokens,
111 record.totals.outputTokens,
112 record.totals.reasoningTokens,
113 record.dedupeKey,
114 record.reportedCostUsd,
115 ];
116
85117 const files: Record<string, SerializedFile> = {};
86118 for (const [path, entry] of cache) {
87119 files[path] = {
88120 s: entry.size,
89121 m: entry.mtimeMs,
90122 p: entry.provider,
91 r: entry.records.map((record) => [
92 record.timestampMs,
93 intern(models, modelIndex, record.model),
94 intern(sessions, sessionIndex, record.sessionId),
95 record.totals.uncachedInputTokens,
96 record.totals.cachedInputTokens,
97 record.totals.cacheCreationTokens,
98 record.totals.outputTokens,
99 record.totals.reasoningTokens,
100 record.dedupeKey,
101 record.reportedCostUsd,
102 ]),
123 r: entry.records.map(serializeRecord),
124 t: entry.tailRecords.map(serializeRecord),
125 o: entry.position.resumeOffset,
126 gl: entry.position.guardLength,
127 gh: entry.position.guardHash,
128 cs: entry.position.codexState,
103129 };
104130 }
105131
......@@ -133,24 +159,16 @@ export function decodeScanCache(document: unknown): ScanCache {
133159 const models = root.models as readonly string[];
134160 const sessions = root.sessions as readonly string[];
135161
136 for (const [path, raw] of Object.entries(root.files)) {
137 if (typeof raw !== "object" || raw === null) continue;
138 const entry = raw as Partial<SerializedFile>;
139 if (typeof entry.s !== "number" || typeof entry.m !== "number") continue;
140 if (entry.p !== "claude" && entry.p !== "codex" && entry.p !== "grok") continue;
141 if (!isRecordArray(entry.r)) continue;
142
143 const provider: UsageProviderKind = entry.p;
162 // Any corrupt row disqualifies the whole entry. Keeping the survivors
163 // under the original (size, mtime) would read as a valid warm hit and the
164 // file would never be re-parsed, silently losing the dropped rows' usage.
165 const decodeRecords = (
166 rows: readonly unknown[],
167 provider: UsageProviderKind,
168 ): UsageRecord[] | null => {
144169 const records: UsageRecord[] = [];
145 // Any corrupt row disqualifies the whole entry. Keeping the survivors
146 // under the original (size, mtime) would read as a valid warm hit and the
147 // file would never be re-parsed, silently losing the dropped rows' usage.
148 let corrupt = false;
149 for (const row of entry.r) {
150 if (!isRecordArray(row) || row.length < 10) {
151 corrupt = true;
152 break;
153 }
170 for (const row of rows) {
171 if (!isRecordArray(row) || row.length < 10) return null;
154172 const [
155173 timestampMs,
156174 modelIndex,
......@@ -175,8 +193,7 @@ export function decodeScanCache(document: unknown): ScanCache {
175193 !Number.isFinite(output) ||
176194 !Number.isFinite(reasoning)
177195 ) {
178 corrupt = true;
179 break;
196 return null;
180197 }
181198
182199 records.push({
......@@ -195,14 +212,89 @@ export function decodeScanCache(document: unknown): ScanCache {
195212 dedupeKey: typeof dedupeKey === "string" ? dedupeKey : null,
196213 });
197214 }
215 return records;
216 };
198217
199 if (corrupt) continue;
200 cache.set(path, { size: entry.s, mtimeMs: entry.m, provider, records });
218 for (const [path, raw] of Object.entries(root.files)) {
219 if (typeof raw !== "object" || raw === null) continue;
220 const entry = raw as Partial<SerializedFile>;
221 if (typeof entry.s !== "number" || typeof entry.m !== "number") continue;
222 if (entry.p !== "claude" && entry.p !== "codex" && entry.p !== "grok") continue;
223 if (!isRecordArray(entry.r) || !isRecordArray(entry.t)) continue;
224 // Position fields feed byte offsets and a Buffer allocation in the reader,
225 // so anything outside their real ranges must reject the entry: a bogus
226 // guard length would otherwise fail every parse of the file, silently
227 // dropping its usage instead of costing the documented cold re-parse.
228 if (
229 typeof entry.o !== "number" ||
230 !Number.isSafeInteger(entry.o) ||
231 entry.o < 0 ||
232 typeof entry.gl !== "number" ||
233 !Number.isSafeInteger(entry.gl) ||
234 entry.gl < 0 ||
235 entry.gl > GUARD_LENGTH ||
236 entry.gl > entry.o ||
237 typeof entry.gh !== "number" ||
238 !Number.isFinite(entry.gh)
239 ) {
240 continue;
241 }
242 const codexState = decodeCodexState(entry.cs);
243 if (codexState === undefined) continue;
244
245 const provider: UsageProviderKind = entry.p;
246 const records = decodeRecords(entry.r, provider);
247 const tailRecords = decodeRecords(entry.t, provider);
248 if (records === null || tailRecords === null) continue;
249
250 cache.set(path, {
251 size: entry.s,
252 mtimeMs: entry.m,
253 provider,
254 records,
255 tailRecords,
256 position: {
257 resumeOffset: entry.o,
258 guardLength: entry.gl,
259 guardHash: entry.gh,
260 codexState,
261 },
262 });
201263 }
202264
203265 return cache;
204266}
205267
268/**
269 * Validates a persisted Codex reducer state. Returns `undefined` for a corrupt
270 * value, which disqualifies the entry: resuming with a bad state would attach
271 * appended usage to the wrong model or replay fork-copied history.
272 */
273function decodeCodexState(value: unknown): CodexScanState | null | undefined {
274 if (value === null) return null;
275 if (typeof value !== "object") return undefined;
276 const state = value as Partial<CodexScanState>;
277 if (
278 typeof state.model !== "string" ||
279 typeof state.sessionId !== "string" ||
280 (state.lastUsageSignature !== null && typeof state.lastUsageSignature !== "string") ||
281 typeof state.sawSessionMeta !== "boolean" ||
282 typeof state.suppressingForkCopies !== "boolean" ||
283 typeof state.forkCopyAnchorMs !== "number" ||
284 !Number.isFinite(state.forkCopyAnchorMs)
285 ) {
286 return undefined;
287 }
288 return {
289 model: state.model,
290 sessionId: state.sessionId,
291 lastUsageSignature: state.lastUsageSignature ?? null,
292 sawSessionMeta: state.sawSessionMeta,
293 suppressingForkCopies: state.suppressingForkCopies,
294 forkCopyAnchorMs: state.forkCopyAnchorMs,
295 };
296}
297
206298export interface PruneOptions {
207299 /** Files the walk just saw. Only meaningful inside the walked window. */
208300 readonly livePaths: ReadonlySet<string>;
......@@ -251,9 +343,17 @@ export function pruneScanCache(cache: ScanCache, options: PruneOptions): number
251343 return removed;
252344}
253345
254/** Within-file de-duplication, applied before an entry is cached. */
255export function dedupeWithinFile(records: readonly UsageRecord[]): readonly UsageRecord[] {
256 const seen = new Set<string>();
346/**
347 * Within-file de-duplication, applied before an entry is cached.
348 *
349 * Callers stitching an incremental parse together pass one `seen` set across
350 * the line and tail record batches so the whole file stays deduplicated as a
351 * unit; the set is mutated in place.
352 */
353export function dedupeWithinFile(
354 records: readonly UsageRecord[],
355 seen: Set<string> = new Set(),
356): readonly UsageRecord[] {
257357 const kept: UsageRecord[] = [];
258358 for (const record of records) {
259359 if (record.dedupeKey !== null) {
apps/server/src/usage/usageTranscriptReader.test.ts created+210
......@@ -0,0 +1,210 @@
1// @effect-diagnostics nodeBuiltinImport:off - resume coverage writes, appends
2// to, and truncates real transcript files byte-exactly, mirroring the reader's
3// own deliberate node:fs usage.
4import * as NodeFSP from "node:fs/promises";
5import * as NodeOS from "node:os";
6import * as NodePath from "node:path";
7
8import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest";
9
10import { readTranscriptRecords } from "./usageTranscriptReader.ts";
11
12let dir: string;
13
14beforeEach(async () => {
15 dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "usage-reader-test-"));
16});
17
18afterEach(async () => {
19 await NodeFSP.rm(dir, { recursive: true, force: true });
20});
21
22function claudeLine(id: number, outputTokens: number): string {
23 return `${JSON.stringify({
24 type: "assistant",
25 timestamp: "2026-08-01T10:00:00Z",
26 requestId: `req_${id}`,
27 sessionId: "session-1",
28 message: {
29 id: `msg_${id}`,
30 model: "claude-fable-5",
31 usage: { input_tokens: 10, output_tokens: outputTokens },
32 },
33 })}\n`;
34}
35
36function codexMetaLine(): string {
37 return `${JSON.stringify({
38 type: "session_meta",
39 timestamp: "2026-08-01T10:00:00Z",
40 payload: { type: "session_meta", id: "codex-session-1" },
41 })}\n`;
42}
43
44function codexModelLine(model: string): string {
45 return `${JSON.stringify({
46 type: "turn_context",
47 timestamp: "2026-08-01T10:00:01Z",
48 payload: { type: "turn_context", model },
49 })}\n`;
50}
51
52function codexUsageLine(outputTokens: number, secondsOffset: number): string {
53 return `${JSON.stringify({
54 type: "event_msg",
55 timestamp: `2026-08-01T10:00:${String(secondsOffset).padStart(2, "0")}Z`,
56 payload: {
57 type: "token_count",
58 info: { last_token_usage: { input_tokens: 100, output_tokens: outputTokens } },
59 },
60 })}\n`;
61}
62
63describe("readTranscriptRecords resume", () => {
64 it("parses only appended lines when resuming a grown file", async () => {
65 const path = NodePath.join(dir, "claude.jsonl");
66 await NodeFSP.writeFile(path, claudeLine(1, 5) + claudeLine(2, 7));
67 const first = await readTranscriptRecords(path, "claude");
68 assert.isNotNull(first);
69 assert.strictEqual(first.records.length, 2);
70 assert.isFalse(first.resumed);
71
72 await NodeFSP.appendFile(path, claudeLine(3, 11));
73 const second = await readTranscriptRecords(path, "claude", first.position);
74 assert.isNotNull(second);
75 assert.isTrue(second.resumed);
76 assert.strictEqual(second.records.length, 1);
77 assert.strictEqual(second.records[0]?.totals.outputTokens, 11);
78
79 // The stitched result matches a from-scratch parse of the whole file.
80 const full = await readTranscriptRecords(path, "claude");
81 assert.isNotNull(full);
82 assert.deepStrictEqual([...first.records, ...second.records], [...full.records]);
83 });
84
85 it("carries the Codex reducer state across the resume boundary", async () => {
86 const path = NodePath.join(dir, "rollout.jsonl");
87 await NodeFSP.writeFile(path, codexMetaLine() + codexModelLine("gpt-5.2-codex"));
88 const first = await readTranscriptRecords(path, "codex");
89 assert.isNotNull(first);
90 assert.strictEqual(first.records.length, 0);
91
92 // The appended usage event has no turn_context or session_meta of its own;
93 // model and session must come from the state captured before the boundary.
94 await NodeFSP.appendFile(path, codexUsageLine(9, 5));
95 const second = await readTranscriptRecords(path, "codex", first.position);
96 assert.isNotNull(second);
97 assert.isTrue(second.resumed);
98 assert.strictEqual(second.records.length, 1);
99 assert.strictEqual(second.records[0]?.model, "gpt-5.2-codex");
100 assert.strictEqual(second.records[0]?.sessionId, "codex-session-1");
101 });
102
103 it("suppresses a Codex duplicate usage event that straddles the boundary", async () => {
104 const path = NodePath.join(dir, "rollout.jsonl");
105 await NodeFSP.writeFile(
106 path,
107 codexMetaLine() + codexModelLine("gpt-5.2-codex") + codexUsageLine(9, 5),
108 );
109 const first = await readTranscriptRecords(path, "codex");
110 assert.isNotNull(first);
111 assert.strictEqual(first.records.length, 1);
112
113 // Codex re-emits an unchanged token_count on stream boundaries; the copy
114 // lands after the resume point and must still be dropped.
115 await NodeFSP.appendFile(path, codexUsageLine(9, 5) + codexUsageLine(21, 8));
116 const second = await readTranscriptRecords(path, "codex", first.position);
117 assert.isNotNull(second);
118 assert.isTrue(second.resumed);
119 assert.deepStrictEqual(
120 second.records.map((record) => record.totals.outputTokens),
121 [21],
122 );
123 });
124
125 it("defers an unterminated trailing line to tailRecords, then consumes it once terminated", async () => {
126 const path = NodePath.join(dir, "claude.jsonl");
127 const unterminated = claudeLine(2, 7).trimEnd();
128 await NodeFSP.writeFile(path, claudeLine(1, 5) + unterminated);
129 const first = await readTranscriptRecords(path, "claude");
130 assert.isNotNull(first);
131 assert.strictEqual(first.records.length, 1);
132 assert.strictEqual(first.tailRecords.length, 1);
133 assert.strictEqual(first.tailRecords[0]?.totals.outputTokens, 7);
134
135 // Completing the line and appending another re-reads from the resume
136 // point, so the once-tail record arrives exactly once as a line record.
137 await NodeFSP.appendFile(path, `\n${claudeLine(3, 11)}`);
138 const second = await readTranscriptRecords(path, "claude", first.position);
139 assert.isNotNull(second);
140 assert.isTrue(second.resumed);
141 assert.deepStrictEqual(
142 second.records.map((record) => record.totals.outputTokens),
143 [7, 11],
144 );
145 assert.strictEqual(second.tailRecords.length, 0);
146 });
147
148 it("re-parses from the start when the guard bytes no longer match", async () => {
149 const path = NodePath.join(dir, "claude.jsonl");
150 await NodeFSP.writeFile(path, claudeLine(1, 5));
151 const first = await readTranscriptRecords(path, "claude");
152 assert.isNotNull(first);
153
154 // Same path, larger size, different content: a replaced file, not growth.
155 await NodeFSP.writeFile(path, claudeLine(4, 13) + claudeLine(5, 17));
156 const second = await readTranscriptRecords(path, "claude", first.position);
157 assert.isNotNull(second);
158 assert.isFalse(second.resumed);
159 assert.deepStrictEqual(
160 second.records.map((record) => record.totals.outputTokens),
161 [13, 17],
162 );
163 });
164
165 it("re-parses from the start when the file shrank below the resume point", async () => {
166 const path = NodePath.join(dir, "claude.jsonl");
167 await NodeFSP.writeFile(path, claudeLine(1, 5) + claudeLine(2, 7));
168 const first = await readTranscriptRecords(path, "claude");
169 assert.isNotNull(first);
170
171 await NodeFSP.writeFile(path, claudeLine(3, 11));
172 const second = await readTranscriptRecords(path, "claude", first.position);
173 assert.isNotNull(second);
174 assert.isFalse(second.resumed);
175 assert.deepStrictEqual(
176 second.records.map((record) => record.totals.outputTokens),
177 [11],
178 );
179 });
180
181 it("parses a line larger than one stream chunk", async () => {
182 // Tool-heavy transcripts carry multi-megabyte single lines; they arrive
183 // split across many chunks and must reassemble into one record.
184 const path = NodePath.join(dir, "claude.jsonl");
185 const bigLine = `${JSON.stringify({
186 type: "assistant",
187 timestamp: "2026-08-01T10:00:00Z",
188 requestId: "req_big",
189 sessionId: "session-1",
190 padding: "x".repeat(512 * 1024),
191 message: {
192 id: "msg_big",
193 model: "claude-fable-5",
194 usage: { input_tokens: 10, output_tokens: 42 },
195 },
196 })}\n`;
197 await NodeFSP.writeFile(path, bigLine + claudeLine(2, 7));
198
199 const parsed = await readTranscriptRecords(path, "claude");
200 assert.isNotNull(parsed);
201 assert.deepStrictEqual(
202 parsed.records.map((record) => record.totals.outputTokens),
203 [42, 7],
204 );
205 });
206
207 it("returns null for an unreadable file", async () => {
208 assert.isNull(await readTranscriptRecords(NodePath.join(dir, "missing.jsonl"), "claude"));
209 });
210});
apps/server/src/usage/usageTranscriptReader.ts+178-23
......@@ -4,16 +4,19 @@
44 *
55 * Isolated here so the rest of the usage code stays on Effect's `FileSystem`.
66 * The direct `node:fs` streaming is deliberate: a cold 30-day window is ~1.4 GB
7 * across ~1,500 files, and `readline` over a read stream is roughly an order of
7 * across ~1,500 files, and buffer-level streaming is roughly an order of
88 * magnitude cheaper than materialising each file. The equivalent Effect stream
99 * pipeline is idiomatic but not fast enough to sit behind a page load.
1010 *
11 * Transcripts are append-only, so a parse also reports the byte position it
12 * stopped at. A later scan of the same file resumes from that position and
13 * parses only the appended bytes, which is what keeps a warm scan cheap while a
14 * session is actively writing a multi-hundred-megabyte rollout.
15 *
1116 * @module usageTranscriptReader
1217 */
13import * as NodeFS from "node:fs";
1418import * as NodeFSP from "node:fs/promises";
1519import * as NodePath from "node:path";
16import * as NodeReadline from "node:readline";
1720
1821import type { UsageProviderKind } from "@t3tools/contracts";
1922
......@@ -23,6 +26,7 @@ import {
2326 parseClaudeLine,
2427 parseCodexLine,
2528 parseGrokLine,
29 type CodexScanState,
2630 type UsageRecord,
2731} from "./usageTranscripts.ts";
2832
......@@ -32,6 +36,56 @@ export interface TranscriptFile {
3236 readonly mtimeMs: number;
3337}
3438
39/**
40 * Where a parse stopped, with enough state to continue from there.
41 *
42 * The guard hash fingerprints the bytes immediately before `resumeOffset`. A
43 * resume only proceeds when those bytes still match: transcripts are
44 * append-only by design, but a rotated or rewritten file silently mis-parsed
45 * from the middle would corrupt usage totals. The window is a cheap tripwire
46 * for those realistic failure shapes, all of which disturb the file's tail at
47 * that exact offset; it deliberately does not hash the whole prefix, which
48 * would cost the full re-read the resume exists to avoid.
49 */
50export interface TranscriptParsePosition {
51 /** Byte offset just past the last newline-terminated line consumed. */
52 readonly resumeOffset: number;
53 /** Length of the fingerprinted window ending at `resumeOffset`. */
54 readonly guardLength: number;
55 /** FNV-1a hash of that window. */
56 readonly guardHash: number;
57 /** Codex reducer state as of `resumeOffset`; `null` for stateless providers. */
58 readonly codexState: CodexScanState | null;
59}
60
61export interface TranscriptParseResult {
62 /** Records from newline-terminated lines at or after the parse start. */
63 readonly records: readonly UsageRecord[];
64 /**
65 * Records from a trailing segment the writer has not newline-terminated yet.
66 * Kept out of `records` because `position` deliberately excludes that
67 * segment: the next scan re-reads it once the writer finishes the line.
68 */
69 readonly tailRecords: readonly UsageRecord[];
70 readonly position: TranscriptParsePosition;
71 /** Whether the parse continued from `resumeFrom` rather than byte 0. */
72 readonly resumed: boolean;
73}
74
75/** 64 bytes of JSONL tail is ample to distinguish a replaced file. */
76export const GUARD_LENGTH = 64;
77const NEWLINE = 0x0a;
78const CARRIAGE_RETURN = 0x0d;
79
80function fnv1a(buffer: Buffer): number {
81 let hash = 0x811c9dc5;
82 for (let index = 0; index < buffer.length; index += 1) {
83 hash ^= buffer[index]!;
84 hash = Math.imul(hash, 0x01000193);
85 }
86 return hash >>> 0;
87}
88
3589/**
3690 * Lists `.jsonl` transcripts under `root` last modified at or after `sinceMs`.
3791 *
......@@ -100,6 +154,25 @@ export async function readDirectoryVolumeId(path: string): Promise<string> {
100154 }
101155}
102156
157async function guardMatches(
158 handle: NodeFSP.FileHandle,
159 position: TranscriptParsePosition,
160): Promise<boolean> {
161 if (position.guardLength <= 0 || position.guardLength > GUARD_LENGTH) return false;
162 try {
163 const window = Buffer.alloc(position.guardLength);
164 const { bytesRead } = await handle.read(
165 window,
166 0,
167 position.guardLength,
168 position.resumeOffset - position.guardLength,
169 );
170 return bytesRead === position.guardLength && fnv1a(window) === position.guardHash;
171 } catch {
172 return false;
173 }
174}
175
103176/**
104177 * Streams one transcript and returns the usage records it contains, or `null`
105178 * when the file could not be read.
......@@ -109,6 +182,10 @@ export async function readDirectoryVolumeId(path: string): Promise<string> {
109182 * under the same `(size, mtime)` key would silently drop that file's usage
110183 * until the file next changes.
111184 *
185 * With `resumeFrom`, parsing continues from that position when its guard bytes
186 * still match, so only appended lines are read; otherwise the whole file is
187 * re-parsed from the start and `resumed` reports `false`.
188 *
112189 * Codex carries the active model on `turn_context` lines that hold no usage of
113190 * their own, so those still have to pass through the reducer to keep model
114191 * attribution correct.
......@@ -116,43 +193,121 @@ export async function readDirectoryVolumeId(path: string): Promise<string> {
116193export async function readTranscriptRecords(
117194 filePath: string,
118195 provider: UsageProviderKind,
119): Promise<readonly UsageRecord[] | null> {
120 const records: UsageRecord[] = [];
121 const codexState = initialCodexScanState();
196 resumeFrom?: TranscriptParsePosition,
197): Promise<TranscriptParseResult | null> {
198 let handle: NodeFSP.FileHandle;
199 try {
200 handle = await NodeFSP.open(filePath, "r");
201 } catch {
202 return null;
203 }
122204
123205 try {
124 const lines = NodeReadline.createInterface({
125 input: NodeFS.createReadStream(filePath, { encoding: "utf8" }),
126 crlfDelay: Infinity,
127 });
206 let codexState = initialCodexScanState();
207 let resumed = false;
208 let start = 0;
209 if (
210 resumeFrom !== undefined &&
211 resumeFrom.resumeOffset > 0 &&
212 (provider !== "codex" || resumeFrom.codexState !== null) &&
213 (await guardMatches(handle, resumeFrom))
214 ) {
215 if (resumeFrom.codexState !== null) codexState = { ...resumeFrom.codexState };
216 start = resumeFrom.resumeOffset;
217 resumed = true;
218 }
128219
129 for await (const line of lines) {
220 const parseLine = (line: string, state: CodexScanState, out: UsageRecord[]): void => {
130221 if (provider === "codex") {
131222 if (
132223 !mightCarryUsage(line, provider) &&
133224 !line.includes('"turn_context"') &&
134225 !line.includes('"session_meta"')
135226 ) {
136 continue;
227 return;
137228 }
138 const record = parseCodexLine(line, codexState);
139 if (record !== null) records.push(record);
140 continue;
229 const record = parseCodexLine(line, state);
230 if (record !== null) out.push(record);
231 return;
141232 }
142
233 if (!mightCarryUsage(line, provider)) return;
143234 if (provider === "grok") {
144 if (!mightCarryUsage(line, provider)) continue;
145 for (const grokRecord of parseGrokLine(line)) records.push(grokRecord);
235 for (const grokRecord of parseGrokLine(line)) out.push(grokRecord);
236 return;
237 }
238 const record = parseClaudeLine(line);
239 if (record !== null) out.push(record);
240 };
241
242 const toLineString = (lineBuffer: Buffer): string => {
243 const content =
244 lineBuffer.length > 0 && lineBuffer[lineBuffer.length - 1] === CARRIAGE_RETURN
245 ? lineBuffer.subarray(0, -1)
246 : lineBuffer;
247 return content.toString("utf8");
248 };
249
250 const records: UsageRecord[] = [];
251 // Buffer-level line splitting rather than `readline`, because resuming
252 // needs byte-exact offsets and decoded strings cannot provide them.
253 // Newline-free chunks are collected rather than concatenated as they
254 // arrive, so a single huge line costs one copy instead of one per chunk.
255 let resumeOffset = start;
256 let pendingChunks: Buffer[] = [];
257 const stream = handle.createReadStream({
258 start,
259 autoClose: false,
260 }) as AsyncIterable<Buffer>;
261 for await (const chunk of stream) {
262 if (!chunk.includes(NEWLINE)) {
263 pendingChunks.push(chunk);
146264 continue;
147265 }
266 const buffer: Buffer =
267 pendingChunks.length === 0 ? chunk : Buffer.concat([...pendingChunks, chunk]);
268 pendingChunks = [];
269 let lineStart = 0;
270 for (;;) {
271 const newlineIndex = buffer.indexOf(NEWLINE, lineStart);
272 if (newlineIndex === -1) break;
273 parseLine(toLineString(buffer.subarray(lineStart, newlineIndex)), codexState, records);
274 lineStart = newlineIndex + 1;
275 }
276 resumeOffset += lineStart;
277 if (lineStart < buffer.length) pendingChunks.push(buffer.subarray(lineStart));
278 }
148279
149 if (!mightCarryUsage(line, provider)) continue;
150 const record = parseClaudeLine(line);
151 if (record !== null) records.push(record);
280 // A trailing segment without its newline is parsed for this result but not
281 // consumed: a writer may still be appending to it, and counting a half
282 // record now and its full form later would double count.
283 const tailRecords: UsageRecord[] = [];
284 if (pendingChunks.length > 0) {
285 const pending = pendingChunks.length === 1 ? pendingChunks[0]! : Buffer.concat(pendingChunks);
286 if (pending.length > 0) parseLine(toLineString(pending), { ...codexState }, tailRecords);
152287 }
288
289 const guardLength = Math.min(GUARD_LENGTH, resumeOffset);
290 let guardHash = 0;
291 if (guardLength > 0) {
292 const window = Buffer.alloc(guardLength);
293 await handle.read(window, 0, guardLength, resumeOffset - guardLength);
294 guardHash = fnv1a(window);
295 }
296
297 return {
298 records,
299 tailRecords,
300 position: {
301 resumeOffset,
302 guardLength,
303 guardHash,
304 codexState: provider === "codex" ? codexState : null,
305 },
306 resumed,
307 };
153308 } catch {
154309 return null;
310 } finally {
311 await handle.close().catch(() => undefined);
155312 }
156
157 return records;
158313}