fix(server): bound usage transcript lines

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<string> {
+  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);
+  });
+});

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<string, void> {
+  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<string> {
 export async function readTranscriptRecords(
   filePath: string,
   provider: UsageProviderKind,
+  options: TranscriptReadOptions = {},
 ): Promise<readonly UsageRecord[] | null> {
   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") {