From 7f02b849584332368a5489631d3286409eed90a8 Mon Sep 17 00:00:00 2001 From: Tristan Knight Date: Mon, 12 Oct 2026 01:39:55 +0100 Subject: [PATCH] perf(usage): the first usage scan skips decoding transcript lines that carry no usage (#17255) --- .../src/provider/Drivers/claudeUsage.ts | 2 +- .../server/src/provider/Drivers/codexUsage.ts | 6 +- .../src/usage/usageTranscriptReader.test.ts | 56 ++++++++++++++++++- .../server/src/usage/usageTranscriptReader.ts | 37 +++++++++--- packages/provider-core/src/server/usage.ts | 9 +-- packages/provider-grok/src/server/usage.ts | 2 +- 6 files changed, 93 insertions(+), 19 deletions(-) diff --git a/apps/server/src/provider/Drivers/claudeUsage.ts b/apps/server/src/provider/Drivers/claudeUsage.ts index 2bab4f5e92b62ed04a2f9fd6e44a1a8c578914db..d0ad4202954cc7f39a0a2ac4ef7e3a691cf66da0 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.ts +++ b/apps/server/src/provider/Drivers/claudeUsage.ts @@ -98,7 +98,7 @@ export const claudeUsageFormat: TranscriptUsageFormat = { costUSD: true, message: { id: true, model: true, usage: true }, }, - mightCarryUsage: (line) => line.includes('"usage"'), + usageMarkers: ['"usage"'], parseLine: (line) => orEmpty(parseClaudeLine(line)), parseProjected: (projected) => orEmpty(parseClaudeRecord(projected)), }; diff --git a/apps/server/src/provider/Drivers/codexUsage.ts b/apps/server/src/provider/Drivers/codexUsage.ts index fc5294e205974a01aa220855293013648990e3cd..1ffcb3aecfcd00fee55cb5893798cda2d01f8f7f 100644 --- a/apps/server/src/provider/Drivers/codexUsage.ts +++ b/apps/server/src/provider/Drivers/codexUsage.ts @@ -231,11 +231,7 @@ export const codexUsageFormat: TranscriptUsageFormat = { }, // `turn_context`, `thread_settings_applied` and `session_meta` lines hold no // usage, but they carry the model, tier and session the reducer needs. - mightCarryUsage: (line) => - line.includes('"token_count"') || - line.includes('"turn_context"') || - line.includes('"thread_settings_applied"') || - line.includes('"session_meta"'), + usageMarkers: ['"token_count"', '"turn_context"', '"thread_settings_applied"', '"session_meta"'], parseLine: (line, state) => orEmpty(parseCodexLine(line, state)), parseProjected: (projected, state) => orEmpty(parseCodexRecord(projected, state)), state: { initial: initialCodexScanState, schema: CodexScanState }, diff --git a/apps/server/src/usage/usageTranscriptReader.test.ts b/apps/server/src/usage/usageTranscriptReader.test.ts index e2c4ee1827feac35e39092c2ba942ffbeea98993..e9a39ccb52dfa20e18c5867bca0dca0ce6a87747 100644 --- a/apps/server/src/usage/usageTranscriptReader.test.ts +++ b/apps/server/src/usage/usageTranscriptReader.test.ts @@ -8,7 +8,7 @@ import * as NodePath from "node:path"; import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; import { TEST_FORMATS } from "./usageTestFormats.ts"; -import { readTranscriptRecords } from "./usageTranscriptReader.ts"; +import { readTranscriptRecords, transcriptLineGate } from "./usageTranscriptReader.ts"; let dir: string; @@ -218,9 +218,63 @@ describe("readTranscriptRecords resume", () => { ); }); + it("keeps a CRLF Codex reducer line whose marker straddles a stream chunk", async () => { + const path = NodePath.join(dir, "rollout.jsonl"); + const crlf = (line: string) => line.replace("\n", "\r\n"); + const modelLine = crlf(codexModelLine("gpt-5.2-codex")); + // Multi-byte tool output that fails the gate, sized so the 256 KiB chunk + // boundary falls inside the model line's `"turn_context"` marker. + const head = crlf(codexMetaLine()); + const fillerBytes = + 256 * 1024 - Buffer.byteLength(head) - modelLine.indexOf("turn_context") - 4; + const fillerEnvelope = crlf(`${JSON.stringify({ type: "response_item", output: "" })}\n`); + const padding = "é".repeat((fillerBytes - Buffer.byteLength(fillerEnvelope)) / 2); + const filler = crlf(`${JSON.stringify({ type: "response_item", output: padding })}\n`); + const content = head + filler + modelLine + crlf(codexUsageLine(40, 2)); + await NodeFSP.writeFile(path, content); + + const result = await readTranscriptRecords(path, TEST_FORMATS.codex); + assert.isNotNull(result); + assert.strictEqual(result.records.length, 1); + assert.strictEqual(result.records[0]?.model, "gpt-5.2-codex"); + assert.strictEqual(result.records[0]?.sessionId, "codex-session-1"); + assert.strictEqual(result.position.resumeOffset, Buffer.byteLength(content)); + }); + it("returns null for an unreadable file", async () => { assert.isNull( await readTranscriptRecords(NodePath.join(dir, "missing.jsonl"), TEST_FORMATS.claude), ); }); }); + +describe("transcriptLineGate", () => { + const gates = { + one: transcriptLineGate(['"usage"']), + several: transcriptLineGate(['"token_count"', '"session_meta"', '"a.b(c)"']), + }; + const markers = { one: ['"usage"'], several: ['"token_count"', '"session_meta"', '"a.b(c)"'] }; + + it("passes only lines containing one of its markers", () => { + for (const kind of ["one", "several"] as const) { + for (const marker of markers[kind]) { + assert.isTrue(gates[kind](Buffer.from(`{"payload":{${marker}:1}}`))); + // The same text without its quotes, or with one character changed. + assert.isFalse(gates[kind](Buffer.from(`{"text":"see ${marker.slice(1, -1)} here"}`))); + assert.isFalse(gates[kind](Buffer.from(`{${marker.replace(/[._]|usage/, "x")}:1}`))); + } + assert.isFalse(gates[kind](Buffer.alloc(0))); + } + }); + + it("matches markers beside multi-byte and malformed UTF-8", () => { + const invalid = Buffer.from([0xff, 0xc3]); + for (const kind of ["one", "several"] as const) { + for (const marker of markers[kind]) { + assert.isTrue(gates[kind](Buffer.from(`{"text":"日本語 é",${marker}:1}`))); + assert.isTrue(gates[kind](Buffer.concat([invalid, Buffer.from(marker), invalid]))); + assert.isFalse(gates[kind](Buffer.from(`{"text":"日本語 é ${marker.slice(1, -1)}"}`))); + } + } + }); +}); diff --git a/apps/server/src/usage/usageTranscriptReader.ts b/apps/server/src/usage/usageTranscriptReader.ts index dbf0fd7aeb35f80718071ee8fecfc611f9daaba2..aed2e8465f1ccd67d3f7715854da1cc88573b1eb 100644 --- a/apps/server/src/usage/usageTranscriptReader.ts +++ b/apps/server/src/usage/usageTranscriptReader.ts @@ -206,6 +206,26 @@ async function guardMatches( } } +/** + * Builds the byte-level test for a format's `usageMarkers`: whether a line + * contains any of them. + * + * One marker is a single `Buffer` search. Several would be one pass over the + * line each, so they are matched together as one alternation over the line's + * latin1 view instead: latin1 maps every byte to one character without + * validating it, so an ASCII marker matches exactly where its bytes do. + */ +export function transcriptLineGate(markers: readonly [string, ...string[]]) { + if (markers.length === 1) { + const marker = Buffer.from(markers[0]); + return (line: Buffer) => line.includes(marker); + } + const pattern = new RegExp( + markers.map((marker) => marker.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")).join("|"), + ); + return (line: Buffer) => pattern.test(line.toString("latin1")); +} + /** * Streams one transcript and returns the usage records it contains, or `null` * when the file could not be read. @@ -259,11 +279,6 @@ export async function readTranscriptRecords( resumed = true; } - const parseLine = (line: string, lineState: State, out: UsageRecord[]): void => { - if (!format.mightCarryUsage(line)) return; - for (const record of format.parseLine(line, lineState)) out.push(record); - }; - const toLineString = (lineBuffer: Buffer): string => { const content = lineBuffer.length > 0 && lineBuffer[lineBuffer.length - 1] === CARRIAGE_RETURN @@ -272,6 +287,14 @@ export async function readTranscriptRecords( return content.toString("utf8"); }; + // Gate on the raw bytes so the lines that fail it, which is most of them, + // are never decoded. + const mightCarryUsage = transcriptLineGate(format.usageMarkers); + const parseLine = (lineBuffer: Buffer, lineState: State, out: UsageRecord[]): void => { + if (!mightCarryUsage(lineBuffer)) return; + for (const record of format.parseLine(toLineString(lineBuffer), lineState)) out.push(record); + }; + const records: UsageRecord[] = []; // Byte offsets remain independent of UTF-8 decoding. Only complete lines // commit the resume point; an unfinished tail is replayed on the next scan. @@ -310,7 +333,7 @@ export async function readTranscriptRecords( pendingChunks.length === 1 ? pendingChunks[0]! : Buffer.concat(pendingChunks, pendingBytes); - parseLine(toLineString(line), lineState, out); + parseLine(line, lineState, out); } pendingChunks = []; pendingBytes = 0; @@ -333,7 +356,7 @@ export async function readTranscriptRecords( // Most lines fit in the current chunk. Avoid buffering/streaming // machinery on this hot path. if (!streaming && pendingBytes === 0) { - parseLine(toLineString(chunk.subarray(lineStart, newlineIndex)), state, records); + parseLine(chunk.subarray(lineStart, newlineIndex), state, records); } else { append(chunk.subarray(lineStart, newlineIndex)); finish(state, records); diff --git a/packages/provider-core/src/server/usage.ts b/packages/provider-core/src/server/usage.ts index 0f1235d7037d8d6a719a4b998b70d450d8017fd4..f69e9b77175d6221d614f3e4baaa5d8812310fc2 100644 --- a/packages/provider-core/src/server/usage.ts +++ b/packages/provider-core/src/server/usage.ts @@ -90,11 +90,12 @@ export type SelectedFields = { readonly [key: string]: true | SelectedFields }; export interface TranscriptUsageFormat { readonly selectFields: SelectedFields; /** - * Cheap substring gate applied before `JSON.parse`. Transcripts are mostly - * tool output; skipping lines that cannot matter is worth an order of - * magnitude. Lines that only update `State` must pass too. + * ASCII substrings, at least one of which every line that matters contains. + * Transcripts are mostly tool output, so the engine matches these against a + * line's raw bytes and neither decodes nor parses the rest, which is worth an + * order of magnitude. Lines that only update `State` must match too. */ - mightCarryUsage(line: string): boolean; + readonly usageMarkers: readonly [string, ...string[]]; parseLine(line: string, state: State): readonly UsageRecord[]; parseProjected(projected: unknown, state: State): readonly UsageRecord[]; /** diff --git a/packages/provider-grok/src/server/usage.ts b/packages/provider-grok/src/server/usage.ts index 4359cd11959ca3d06233e9cdcd1662d36838ae90..be78318753d6d2e6bfa03d3cf6b5bfbfd898b175 100644 --- a/packages/provider-grok/src/server/usage.ts +++ b/packages/provider-grok/src/server/usage.ts @@ -218,7 +218,7 @@ export const grokUsageFormat: TranscriptUsageFormat = { update: { sessionUpdate: true, prompt_id: true, usage: true }, }, }, - mightCarryUsage: (line) => line.includes('"turn_completed"'), + usageMarkers: ['"turn_completed"'], parseLine: (line) => parseGrokLine(line), parseProjected: (projected) => parseGrokRecord(projected), }; -- 2.54.0