feat(intake): stream session transcripts past the 512MB string ceiling

src/contract/index.ts

@@ -297,6 +297,7 @@ export {
   type AgentTraceRange,
   type AgentTraceRecord,
   type AuthoringProvenance,
+  type CodeAgentJsonlLine,
   type CodeAgentSessionAction,
   type CodeAgentSessionActionKind,
   type CodeAgentSessionActionStatus,
@@ -330,6 +331,8 @@ export {
   type PartitionByAuthoringModelResult,
   parseAgentTrace,
   parseCodeAgentJsonl,
+  parseCodeAgentJsonlFile,
   partitionRunsByAuthoringModel,
   type RunRecordRejection,
+  streamCodeAgentJsonlFile,
 } from './intake'

src/contract/intake/code-agent-session.ts

@@ -1,4 +1,5 @@
 import { createHash } from 'node:crypto'
+import { createReadStream } from 'node:fs'
 import { estimateCost, isModelPriced } from '../../metrics'
 import type {
   RunCostProvenance,
@@ -121,17 +122,82 @@ export interface CodeAgentSessionIntakeOptions {
   execution?: CodeAgentSessionExecutionReceipt
 }
 
+/** One transcript line after the intake rule ran on it. A blank line produces
+ *  nothing, so every value here is either a parsed entry or a counted defect. */
+export type CodeAgentJsonlLine =
+  | { kind: 'entry'; lineNumber: number; entry: unknown }
+  | { kind: 'malformed'; lineNumber: number }
+
+/** The single per-line rule. Both the string path and the file path call this,
+ *  so malformed-line handling and entry validation cannot drift apart. */
+function readCodeAgentJsonlLine(line: string, lineNumber: number): CodeAgentJsonlLine | undefined {
+  const trimmed = line.trim()
+  if (!trimmed) return undefined
+  try {
+    return { kind: 'entry', lineNumber, entry: JSON.parse(trimmed) }
+  } catch {
+    return { kind: 'malformed', lineNumber }
+  }
+}
+
 export function parseCodeAgentJsonl(jsonl: string): ParsedCodeAgentJsonl {
   const entries: unknown[] = []
   let malformedLines = 0
+  let lineNumber = 0
   for (const line of jsonl.split('\n')) {
-    const trimmed = line.trim()
-    if (!trimmed) continue
-    try {
-      entries.push(JSON.parse(trimmed))
-    } catch {
-      malformedLines += 1
+    lineNumber += 1
+    const read = readCodeAgentJsonlLine(line, lineNumber)
+    if (!read) continue
+    if (read.kind === 'malformed') malformedLines += 1
+    else entries.push(read.entry)
+  }
+  return { entries, malformedLines }
+}
+
+/** Reads a transcript one line at a time and never holds the file as a single
+ *  string. `parseCodeAgentJsonl` needs the whole file in one string, so a
+ *  session above V8's ~512MB string ceiling throws `ERR_STRING_TOO_LONG` and
+ *  cannot be ingested at all; the largest real Codex rollout on record is 695MB.
+ *
+ *  Lines break on `\n` only, which is what the string path's `split('\n')` does.
+ *  `node:readline` also breaks on a bare `\r`, so it is deliberately not used
+ *  here: a lone carriage return inside a line must stay inside that line for the
+ *  two paths to report the same malformed count. */
+export async function* streamCodeAgentJsonlFile(path: string): AsyncGenerator<CodeAgentJsonlLine> {
+  const stream = createReadStream(path, { encoding: 'utf8' })
+  let pending = ''
+  let lineNumber = 0
+  try {
+    for await (const chunk of stream) {
+      pending += chunk
+      let start = 0
+      for (let at = pending.indexOf('\n'); at !== -1; at = pending.indexOf('\n', start)) {
+        lineNumber += 1
+        const read = readCodeAgentJsonlLine(pending.slice(start, at), lineNumber)
+        if (read) yield read
+        start = at + 1
+      }
+      // Only the unterminated tail is retained, so live memory is bounded by
+      // the longest single line rather than by the file size.
+      pending = pending.slice(start)
     }
+  } finally {
+    stream.destroy()
+  }
+  const last = readCodeAgentJsonlLine(pending, lineNumber + 1)
+  if (last) yield last
+}
+
+/** Streaming counterpart to `parseCodeAgentJsonl` for a transcript on disk.
+ *  It returns the same shape, so a caller that holds every entry keeps working
+ *  above the string ceiling. The entry array still grows with the transcript;
+ *  consume `streamCodeAgentJsonlFile` directly when memory must stay flat. */
+export async function parseCodeAgentJsonlFile(path: string): Promise<ParsedCodeAgentJsonl> {
+  const entries: unknown[] = []
+  let malformedLines = 0
+  for await (const read of streamCodeAgentJsonlFile(path)) {
+    if (read.kind === 'malformed') malformedLines += 1
+    else entries.push(read.entry)
   }
   return { entries, malformedLines }
 }

src/contract/intake/index.ts

@@ -32,6 +32,7 @@ export {
   partitionRunsByAuthoringModel,
 } from './agent-trace'
 export {
+  type CodeAgentJsonlLine,
   type CodeAgentSessionAction,
   type CodeAgentSessionActionKind,
   type CodeAgentSessionActionStatus,
@@ -53,6 +54,8 @@ export {
   observeCodeAgentSession,
   type ParsedCodeAgentJsonl,
   parseCodeAgentJsonl,
+  parseCodeAgentJsonlFile,
+  streamCodeAgentJsonlFile,
 } from './code-agent-session'
 export {
   type FeedbackTableMeta,

tests/contract-code-agent-intake.test.ts

@@ -1,13 +1,19 @@
-import { readFileSync } from 'node:fs'
-import { describe, expect, it } from 'vitest'
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { afterAll, describe, expect, it } from 'vitest'
 import {
   analyzeRuns,
+  type CodeAgentJsonlLine,
   fromClaudeCodeSession,
   fromCodexSession,
   fromKimiCodeSession,
   fromOpenCodeSession,
   fromPiSession,
   parseCodeAgentJsonl,
+  parseCodeAgentJsonlFile,
+  streamCodeAgentJsonlFile,
 } from '../src/contract'
 import { validateRunRecord } from '../src/run-record'
 
@@ -331,7 +337,11 @@ describe('code-agent session intake', () => {
         {
           timestamp: '2026-08-14T00:00:03.000Z',
           type: 'response_item',
-          payload: { type: 'message', role: 'assistant', content: [{ type: 'text', text: 'done' }] },
+          payload: {
+            type: 'message',
+            role: 'assistant',
+            content: [{ type: 'text', text: 'done' }],
+          },
         },
         // The execution record. Reasoning and CommandExecution repeat the
         // transcript above; the rest appear nowhere else in the file.
@@ -352,7 +362,8 @@ describe('code-agent session intake', () => {
           changes: {
             '/repo/src/lib/billing.ts': {
               type: 'update',
-              unified_diff: '@@ -448,2 +448,3 @@\n interface BillingServiceDeps {\n+  enabled?: boolean;\n',
+              unified_diff:
+                '@@ -448,2 +448,3 @@\n interface BillingServiceDeps {\n+  enabled?: boolean;\n',
             },
           },
           stdout: '',
@@ -1187,3 +1198,89 @@ describe('code-agent session intake', () => {
     },
   )
 })
+
+describe('streaming code-agent JSONL intake', () => {
+  const dir = mkdtempSync(join(tmpdir(), 'code-agent-jsonl-'))
+  const write = (name: string, body: string) => {
+    const path = join(dir, name)
+    writeFileSync(path, body)
+    return path
+  }
+
+  afterAll(() => rmSync(dir, { recursive: true, force: true }))
+
+  it('matches the string path on a real committed session fixture', async () => {
+    const url = new URL('./fixtures/codex-exec-0.144.1.jsonl', import.meta.url)
+    const fromString = parseCodeAgentJsonl(readFileSync(url, 'utf8'))
+    const fromFile = await parseCodeAgentJsonlFile(fileURLToPath(url))
+
+    expect(fromFile.entries).toHaveLength(15)
+    expect(fromFile).toEqual(fromString)
+  })
+
+  it('counts malformed lines exactly as the string path does', async () => {
+    const body = '{"type":"ok"}\nnot-json\n\n   \n{"type":"also-ok"}\n{"unterminated":\n'
+    const path = write('malformed.jsonl', body)
+
+    const fromFile = await parseCodeAgentJsonlFile(path)
+
+    expect(fromFile.malformedLines).toBe(2)
+    expect(fromFile).toEqual(parseCodeAgentJsonl(body))
+  })
+
+  it.each([
+    ['no trailing newline', '{"a":1}\n{"b":2}'],
+    ['CRLF line breaks', '{"a":1}\r\n{"b":2}\r\n'],
+    ['blank and whitespace-only lines', '\n{"a":1}\n \t \n{"b":2}\n\n'],
+    ['empty file', ''],
+    ['only malformed content', 'nope\nalso nope\n'],
+  ])('agrees with the string path on %s', async (name, body) => {
+    const path = write(`${name.replace(/\s+/g, '-')}.jsonl`, body)
+
+    expect(await parseCodeAgentJsonlFile(path)).toEqual(parseCodeAgentJsonl(body))
+  })
+
+  it('keeps a bare carriage return inside its line, as split by newline does', async () => {
+    // `node:readline` breaks on a lone `\r`, which would report two malformed
+    // lines here instead of one and silently diverge from the string path.
+    const body = '{"a":1}\nbroken\rline\n{"b":2}\n'
+    const path = write('bare-cr.jsonl', body)
+
+    const fromFile = await parseCodeAgentJsonlFile(path)
+
+    expect(fromFile.malformedLines).toBe(1)
+    expect(fromFile).toEqual(parseCodeAgentJsonl(body))
+  })
+
+  it('reassembles entries that span read-stream chunk boundaries', async () => {
+    // Each line is far larger than the 64KB default chunk, so a splitter that
+    // does not retain the unterminated tail loses or corrupts every entry.
+    const long = Array.from({ length: 8 }, (_, i) =>
+      JSON.stringify({ index: i, filler: 'x'.repeat(200_000) }),
+    ).join('\n')
+    const path = write('chunked.jsonl', `${long}\n`)
+
+    const fromFile = await parseCodeAgentJsonlFile(path)
+
+    expect(fromFile.entries).toHaveLength(8)
+    expect(fromFile.entries.map((e) => (e as { index: number }).index)).toEqual([
+      0, 1, 2, 3, 4, 5, 6, 7,
+    ])
+    expect(fromFile).toEqual(parseCodeAgentJsonl(long))
+  })
+
+  it('reports line numbers and stops reading when the consumer stops', async () => {
+    const path = write('numbered.jsonl', '\n{"a":1}\nnot-json\n{"b":2}\n')
+
+    const seen: CodeAgentJsonlLine[] = []
+    for await (const line of streamCodeAgentJsonlFile(path)) {
+      seen.push(line)
+      if (seen.length === 2) break
+    }
+
+    expect(seen).toEqual([
+      { kind: 'entry', lineNumber: 2, entry: { a: 1 } },
+      { kind: 'malformed', lineNumber: 3 },
+    ])
+  })
+})