From 42f1b50435ff27d87bcc967a9546d3a2d6827298 Mon Sep 17 00:00:00 2001 From: ifBars Date: Sun, 16 Aug 2026 14:48:47 -0700 Subject: [PATCH] fix(server): bound usage transcript lines --- .../src/usage/usageTranscriptReader.test.ts | 96 +++++++++++++++++++ .../server/src/usage/usageTranscriptReader.ts | 90 +++++++++++++++-- 2 files changed, 179 insertions(+), 7 deletions(-) create mode 100644 apps/server/src/usage/usageTranscriptReader.test.ts diff --git a/apps/server/src/usage/usageTranscriptReader.test.ts b/apps/server/src/usage/usageTranscriptReader.test.ts new file mode 100644 index 000000000000..4359f7690401 --- /dev/null +++ b/apps/server/src/usage/usageTranscriptReader.test.ts @@ -0,0 +1,96 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + +import { afterEach, describe, expect, it } from "@effect/vitest"; + +import { readTranscriptRecords } from "./usageTranscriptReader.ts"; + +const temporaryDirectories: string[] = []; + +async function writeTranscript(lines: readonly string[], trailingNewline = true): Promise { + const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-usage-reader-")); + temporaryDirectories.push(directory); + const filePath = NodePath.join(directory, "rollout.jsonl"); + await NodeFSP.writeFile(filePath, lines.join("\n") + (trailingNewline ? "\n" : "")); + return filePath; +} + +const sessionMeta = JSON.stringify({ + type: "session_meta", + timestamp: "2026-08-01T05:17:41.289Z", + payload: { type: "session_meta", id: "session-1" }, +}); + +const turnContext = JSON.stringify({ + type: "turn_context", + timestamp: "2026-08-01T05:17:42.694Z", + payload: { type: "turn_context", model: "gpt-5.6-sol" }, +}); + +const tokenCount = JSON.stringify({ + type: "event_msg", + timestamp: "2026-08-01T05:17:49.919Z", + payload: { + type: "token_count", + info: { + last_token_usage: { + input_tokens: 1200, + cached_input_tokens: 200, + cache_write_input_tokens: 0, + output_tokens: 100, + reasoning_output_tokens: 25, + }, + }, + }, +}); + +afterEach(async () => { + await Promise.all( + temporaryDirectories + .splice(0) + .map((directory) => NodeFSP.rm(directory, { recursive: true, force: true })), + ); +}); + +describe("readTranscriptRecords", () => { + it("skips an oversized irrelevant line and continues with later usage", async () => { + const oversizedToolResult = JSON.stringify({ + type: "event_msg", + payload: { type: "patch_apply_end", output: `token_count:${"x".repeat(2048)}` }, + }); + const filePath = await writeTranscript([ + sessionMeta, + turnContext, + oversizedToolResult, + tokenCount, + ]); + + const records = await readTranscriptRecords(filePath, "codex", { maxLineBytes: 512 }); + + expect(records).toHaveLength(1); + expect(records?.[0]).toMatchObject({ + model: "gpt-5.6-sol", + sessionId: "session-1", + totals: { + uncachedInputTokens: 1000, + cachedInputTokens: 200, + outputTokens: 100, + reasoningTokens: 25, + }, + }); + }); + + it("preserves CRLF records and an unterminated final line", async () => { + const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-usage-reader-")); + temporaryDirectories.push(directory); + const filePath = NodePath.join(directory, "rollout.jsonl"); + await NodeFSP.writeFile(filePath, [sessionMeta, turnContext, tokenCount].join("\r\n")); + + const records = await readTranscriptRecords(filePath, "codex"); + + expect(records).toHaveLength(1); + expect(records?.[0]?.totals.outputTokens).toBe(100); + }); +}); diff --git a/apps/server/src/usage/usageTranscriptReader.ts b/apps/server/src/usage/usageTranscriptReader.ts index c72f0c24db65..bf54cf456f32 100644 --- a/apps/server/src/usage/usageTranscriptReader.ts +++ b/apps/server/src/usage/usageTranscriptReader.ts @@ -4,16 +4,17 @@ * * Isolated here so the rest of the usage code stays on Effect's `FileSystem`. * The direct `node:fs` streaming is deliberate: a cold 30-day window is ~1.4 GB - * across ~1,500 files, and `readline` over a read stream is roughly an order of + * across ~1,500 files, and scanning a read stream is roughly an order of * magnitude cheaper than materialising each file. The equivalent Effect stream - * pipeline is idiomatic but not fast enough to sit behind a page load. + * pipeline is idiomatic but not fast enough to sit behind a page load. The + * byte-oriented line reader also lets us discard pathological records before + * constructing a string that could cross V8's maximum length. * * @module usageTranscriptReader */ import * as NodeFS from "node:fs"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; -import * as NodeReadline from "node:readline"; import type { UsageProviderKind } from "@t3tools/contracts"; @@ -31,6 +32,83 @@ export interface TranscriptFile { readonly mtimeMs: number; } +export interface TranscriptReadOptions { + /** Maximum UTF-8 bytes retained for one JSONL record before it is skipped. */ + readonly maxLineBytes?: number; +} + +/** + * Well above observed valid provider records while remaining safely below + * V8's maximum string length. Usage-bearing records are ordinarily tiny; the + * largest transcript lines are tool outputs and embedded media. + */ +const DEFAULT_MAX_LINE_BYTES = 64 * 1024 * 1024; + +function decodeLine(chunks: readonly Buffer[], byteLength: number): string { + const bytes = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, byteLength); + const end = bytes?.[byteLength - 1] === 0x0d ? byteLength - 1 : byteLength; + return bytes?.toString("utf8", 0, end) ?? ""; +} + +/** + * Streams newline-delimited UTF-8 without ever retaining an unbounded record. + * + * Node's `readline` concatenates a whole line before yielding it. A rollout + * can legitimately contain a huge tool result on one line, which lets that + * internal string cross V8's limit and terminate the process before the + * caller's `try/catch` can run. Once a line crosses this reader's limit, its + * remaining bytes are drained through the next newline and scanning resumes. + */ +async function* readBoundedLines( + filePath: string, + maxLineBytes: number, +): AsyncGenerator { + const input = NodeFS.createReadStream(filePath); + let chunks: Buffer[] = []; + let byteLength = 0; + let discarding = false; + + for await (const rawChunk of input) { + const chunk = Buffer.isBuffer(rawChunk) ? rawChunk : Buffer.from(rawChunk); + let start = 0; + + while (start < chunk.length) { + const newline = chunk.indexOf(0x0a, start); + const end = newline === -1 ? chunk.length : newline; + const segmentLength = end - start; + + if (!discarding && segmentLength > 0) { + if (byteLength + segmentLength <= maxLineBytes) { + chunks.push(chunk.subarray(start, end)); + byteLength += segmentLength; + } else { + chunks = []; + byteLength = 0; + discarding = true; + } + } + + if (newline === -1) break; + + if (!discarding) { + const line = decodeLine(chunks, byteLength); + chunks = []; + byteLength = 0; + yield line; + } else { + chunks = []; + byteLength = 0; + discarding = false; + } + start = newline + 1; + } + } + + if (!discarding && byteLength > 0) { + yield decodeLine(chunks, byteLength); + } +} + /** * Lists `.jsonl` transcripts under `root` last modified at or after `sinceMs`. * @@ -105,15 +183,13 @@ export async function readDirectoryVolumeId(path: string): Promise { export async function readTranscriptRecords( filePath: string, provider: UsageProviderKind, + options: TranscriptReadOptions = {}, ): Promise { const records: UsageRecord[] = []; const codexState = initialCodexScanState(); try { - const lines = NodeReadline.createInterface({ - input: NodeFS.createReadStream(filePath, { encoding: "utf8" }), - crlfDelay: Infinity, - }); + const lines = readBoundedLines(filePath, options.maxLineBytes ?? DEFAULT_MAX_LINE_BYTES); for await (const line of lines) { if (provider === "codex") {