authorgravatar for admin@snappeh.comTristan Knight <admin@snappeh.com> 2026-10-12 01:39:55+01:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-12 00:39:55+00:00
log7f02b849584332368a5489631d3286409eed90a8
tree6c30b755624619a6bad42664c6c56de910f56eba
parent499a53c307ec6a0fbff5c83efde355729bd7d343
signature Signed by PGP key B5690EEEBB952194

perf(usage): the first usage scan skips decoding transcript lines that carry no usage (#17255)

6 files changed, 93 insertions(+), 19 deletions(-)

apps/server/src/provider/Drivers/claudeUsage.ts+1-1
......@@ -98,7 +98,7 @@ export const claudeUsageFormat: TranscriptUsageFormat<void> = {
9898 costUSD: true,
9999 message: { id: true, model: true, usage: true },
100100 },
101 mightCarryUsage: (line) => line.includes('"usage"'),
101 usageMarkers: ['"usage"'],
102102 parseLine: (line) => orEmpty(parseClaudeLine(line)),
103103 parseProjected: (projected) => orEmpty(parseClaudeRecord(projected)),
104104};
apps/server/src/provider/Drivers/codexUsage.ts+1-5
......@@ -231,11 +231,7 @@ export const codexUsageFormat: TranscriptUsageFormat<CodexScanState> = {
231231 },
232232 // `turn_context`, `thread_settings_applied` and `session_meta` lines hold no
233233 // usage, but they carry the model, tier and session the reducer needs.
234 mightCarryUsage: (line) =>
235 line.includes('"token_count"') ||
236 line.includes('"turn_context"') ||
237 line.includes('"thread_settings_applied"') ||
238 line.includes('"session_meta"'),
234 usageMarkers: ['"token_count"', '"turn_context"', '"thread_settings_applied"', '"session_meta"'],
239235 parseLine: (line, state) => orEmpty(parseCodexLine(line, state)),
240236 parseProjected: (projected, state) => orEmpty(parseCodexRecord(projected, state)),
241237 state: { initial: initialCodexScanState, schema: CodexScanState },
apps/server/src/usage/usageTranscriptReader.test.ts+55-1
......@@ -8,7 +8,7 @@ import * as NodePath from "node:path";
88import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest";
99
1010import { TEST_FORMATS } from "./usageTestFormats.ts";
11import { readTranscriptRecords } from "./usageTranscriptReader.ts";
11import { readTranscriptRecords, transcriptLineGate } from "./usageTranscriptReader.ts";
1212
1313let dir: string;
1414
......@@ -218,9 +218,63 @@ describe("readTranscriptRecords resume", () => {
218218 );
219219 });
220220
221 it("keeps a CRLF Codex reducer line whose marker straddles a stream chunk", async () => {
222 const path = NodePath.join(dir, "rollout.jsonl");
223 const crlf = (line: string) => line.replace("\n", "\r\n");
224 const modelLine = crlf(codexModelLine("gpt-5.2-codex"));
225 // Multi-byte tool output that fails the gate, sized so the 256 KiB chunk
226 // boundary falls inside the model line's `"turn_context"` marker.
227 const head = crlf(codexMetaLine());
228 const fillerBytes =
229 256 * 1024 - Buffer.byteLength(head) - modelLine.indexOf("turn_context") - 4;
230 const fillerEnvelope = crlf(`${JSON.stringify({ type: "response_item", output: "" })}\n`);
231 const padding = "é".repeat((fillerBytes - Buffer.byteLength(fillerEnvelope)) / 2);
232 const filler = crlf(`${JSON.stringify({ type: "response_item", output: padding })}\n`);
233 const content = head + filler + modelLine + crlf(codexUsageLine(40, 2));
234 await NodeFSP.writeFile(path, content);
235
236 const result = await readTranscriptRecords(path, TEST_FORMATS.codex);
237 assert.isNotNull(result);
238 assert.strictEqual(result.records.length, 1);
239 assert.strictEqual(result.records[0]?.model, "gpt-5.2-codex");
240 assert.strictEqual(result.records[0]?.sessionId, "codex-session-1");
241 assert.strictEqual(result.position.resumeOffset, Buffer.byteLength(content));
242 });
243
221244 it("returns null for an unreadable file", async () => {
222245 assert.isNull(
223246 await readTranscriptRecords(NodePath.join(dir, "missing.jsonl"), TEST_FORMATS.claude),
224247 );
225248 });
226249});
250
251describe("transcriptLineGate", () => {
252 const gates = {
253 one: transcriptLineGate(['"usage"']),
254 several: transcriptLineGate(['"token_count"', '"session_meta"', '"a.b(c)"']),
255 };
256 const markers = { one: ['"usage"'], several: ['"token_count"', '"session_meta"', '"a.b(c)"'] };
257
258 it("passes only lines containing one of its markers", () => {
259 for (const kind of ["one", "several"] as const) {
260 for (const marker of markers[kind]) {
261 assert.isTrue(gates[kind](Buffer.from(`{"payload":{${marker}:1}}`)));
262 // The same text without its quotes, or with one character changed.
263 assert.isFalse(gates[kind](Buffer.from(`{"text":"see ${marker.slice(1, -1)} here"}`)));
264 assert.isFalse(gates[kind](Buffer.from(`{${marker.replace(/[._]|usage/, "x")}:1}`)));
265 }
266 assert.isFalse(gates[kind](Buffer.alloc(0)));
267 }
268 });
269
270 it("matches markers beside multi-byte and malformed UTF-8", () => {
271 const invalid = Buffer.from([0xff, 0xc3]);
272 for (const kind of ["one", "several"] as const) {
273 for (const marker of markers[kind]) {
274 assert.isTrue(gates[kind](Buffer.from(`{"text":"日本語 é",${marker}:1}`)));
275 assert.isTrue(gates[kind](Buffer.concat([invalid, Buffer.from(marker), invalid])));
276 assert.isFalse(gates[kind](Buffer.from(`{"text":"日本語 é ${marker.slice(1, -1)}"}`)));
277 }
278 }
279 });
280});
apps/server/src/usage/usageTranscriptReader.ts+30-7
......@@ -206,6 +206,26 @@ async function guardMatches(
206206 }
207207}
208208
209/**
210 * Builds the byte-level test for a format's `usageMarkers`: whether a line
211 * contains any of them.
212 *
213 * One marker is a single `Buffer` search. Several would be one pass over the
214 * line each, so they are matched together as one alternation over the line's
215 * latin1 view instead: latin1 maps every byte to one character without
216 * validating it, so an ASCII marker matches exactly where its bytes do.
217 */
218export function transcriptLineGate(markers: readonly [string, ...string[]]) {
219 if (markers.length === 1) {
220 const marker = Buffer.from(markers[0]);
221 return (line: Buffer) => line.includes(marker);
222 }
223 const pattern = new RegExp(
224 markers.map((marker) => marker.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|"),
225 );
226 return (line: Buffer) => pattern.test(line.toString("latin1"));
227}
228
209229/**
210230 * Streams one transcript and returns the usage records it contains, or `null`
211231 * when the file could not be read.
......@@ -259,11 +279,6 @@ export async function readTranscriptRecords<State>(
259279 resumed = true;
260280 }
261281
262 const parseLine = (line: string, lineState: State, out: UsageRecord[]): void => {
263 if (!format.mightCarryUsage(line)) return;
264 for (const record of format.parseLine(line, lineState)) out.push(record);
265 };
266
267282 const toLineString = (lineBuffer: Buffer): string => {
268283 const content =
269284 lineBuffer.length > 0 && lineBuffer[lineBuffer.length - 1] === CARRIAGE_RETURN
......@@ -272,6 +287,14 @@ export async function readTranscriptRecords<State>(
272287 return content.toString("utf8");
273288 };
274289
290 // Gate on the raw bytes so the lines that fail it, which is most of them,
291 // are never decoded.
292 const mightCarryUsage = transcriptLineGate(format.usageMarkers);
293 const parseLine = (lineBuffer: Buffer, lineState: State, out: UsageRecord[]): void => {
294 if (!mightCarryUsage(lineBuffer)) return;
295 for (const record of format.parseLine(toLineString(lineBuffer), lineState)) out.push(record);
296 };
297
275298 const records: UsageRecord[] = [];
276299 // Byte offsets remain independent of UTF-8 decoding. Only complete lines
277300 // commit the resume point; an unfinished tail is replayed on the next scan.
......@@ -310,7 +333,7 @@ export async function readTranscriptRecords<State>(
310333 pendingChunks.length === 1
311334 ? pendingChunks[0]!
312335 : Buffer.concat(pendingChunks, pendingBytes);
313 parseLine(toLineString(line), lineState, out);
336 parseLine(line, lineState, out);
314337 }
315338 pendingChunks = [];
316339 pendingBytes = 0;
......@@ -333,7 +356,7 @@ export async function readTranscriptRecords<State>(
333356 // Most lines fit in the current chunk. Avoid buffering/streaming
334357 // machinery on this hot path.
335358 if (!streaming && pendingBytes === 0) {
336 parseLine(toLineString(chunk.subarray(lineStart, newlineIndex)), state, records);
359 parseLine(chunk.subarray(lineStart, newlineIndex), state, records);
337360 } else {
338361 append(chunk.subarray(lineStart, newlineIndex));
339362 finish(state, records);
packages/provider-core/src/server/usage.ts+5-4
......@@ -90,11 +90,12 @@ export type SelectedFields = { readonly [key: string]: true | SelectedFields };
9090export interface TranscriptUsageFormat<State> {
9191 readonly selectFields: SelectedFields;
9292 /**
93 * Cheap substring gate applied before `JSON.parse`. Transcripts are mostly
94 * tool output; skipping lines that cannot matter is worth an order of
95 * magnitude. Lines that only update `State` must pass too.
93 * ASCII substrings, at least one of which every line that matters contains.
94 * Transcripts are mostly tool output, so the engine matches these against a
95 * line's raw bytes and neither decodes nor parses the rest, which is worth an
96 * order of magnitude. Lines that only update `State` must match too.
9697 */
97 mightCarryUsage(line: string): boolean;
98 readonly usageMarkers: readonly [string, ...string[]];
9899 parseLine(line: string, state: State): readonly UsageRecord[];
99100 parseProjected(projected: unknown, state: State): readonly UsageRecord[];
100101 /**
packages/provider-grok/src/server/usage.ts+1-1
......@@ -218,7 +218,7 @@ export const grokUsageFormat: TranscriptUsageFormat<void> = {
218218 update: { sessionUpdate: true, prompt_id: true, usage: true },
219219 },
220220 },
221 mightCarryUsage: (line) => line.includes('"turn_completed"'),
221 usageMarkers: ['"turn_completed"'],
222222 parseLine: (line) => parseGrokLine(line),
223223 parseProjected: (projected) => parseGrokRecord(projected),
224224};